"""Declarative workflow API: goals + tools, orchestrated by the FGAO controller. LangGraph wires nodes with edges you write by hand. Here you declare *what must become true* (goals with mechanical checks) and *what each tool claims to do* (gain, harm, produced predicates, preconditions, cost); FGAO decides the order at run time from those declarations plus online feedback, and stops when the goals are verified or nothing left is worth its cost. wf = Workflow() wf.add_goal("has_data", check=lambda s: "rows" in s, weight=1) wf.add_tool("load", load_fn, gain={"has_data": 0.9}, cost=1) app = wf.compile(budget=10) result = app.invoke({"path": "data.csv"}) State is any object you choose (a dict is the common case). Tools receive the state and their declared arguments, may mutate it, and may return None state was mutated in place dict merged into a dict state (``state.update(d)``) ToolResult(update, cost, error) the same, with a measured cost / error Checks map the state to a value in [0, 1] (``bool`` is accepted). """ from __future__ import annotations import asyncio import inspect from collections.abc import MutableMapping from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from typing import Any, AsyncIterator, Callable, Dict, FrozenSet, Hashable, Iterable, Iterator, List, Mapping, Optional, Sequence, Union from .controller import FGAOConfig, FGAOController from .types import Action, ActionSpec, Condition, Decision, Observation, Semantics, Status @dataclass class ToolResult: """Optional rich return value of a tool.""" update: Optional[Mapping[str, Any]] = None cost: Optional[float] = None # measured cost; default: declared cost error: Optional[str] = None @dataclass class Tool: name: str fn: Optional[Callable[..., Any]] # sync implementation (None for async-only tools) action: Action semantics: Semantics group: Optional[Hashable] = None description: str = "" afn: Optional[Callable[..., Any]] = None # async implementation (used by the async API) resources: FrozenSet[str] = frozenset() # tools sharing a resource never run in the same parallel round @dataclass class Goal: name: str check: Callable[[Any], Union[float, bool]] weight: float = 1.0 mandatory: bool = True description: str = "" @dataclass class Step: """One executed step (yielded by ``stream``).""" step: int action: Action decision: Decision values_before: Dict[str, float] values_after: Dict[str, float] cost: float budget_left: float error: Optional[str] = None @property def tool(self) -> str: return self.action.tool def to_dict(self) -> Dict[str, Any]: s = self.decision.scores return { "step": self.step, "action": str(self.action), "Q": s.Q, "predicted_gain": s.gain, "continuation": s.continuation, "estimated_cost": s.cost, "cost": self.cost, "budget_left": self.budget_left, "values": dict(self.values_after), "error": self.error, } @dataclass class WorkflowResult: """Final outcome of a run (also the last item yielded by ``stream``).""" state: Any status: str success: bool values: Dict[str, float] cost: float steps: List[Step] = field(default_factory=list) def trace(self) -> List[Dict[str, Any]]: return [s.to_dict() for s in self.steps] Callback = Callable[[Step], None] def _as_value(x: Union[float, bool]) -> float: v = float(x) if not 0.0 <= v <= 1.0: raise ValueError(f"goal check returned {x!r}; expected a value in [0, 1]") return v class Workflow: """Builder for an FGAO-orchestrated workflow.""" def __init__(self, name: str = "workflow") -> None: self.name = name self.goals: List[Goal] = [] self.predicates: Dict[str, Callable[[Any], bool]] = {} self.tools: List[Tool] = [] # ------------------------------------------------------------ declaration def add_goal( self, name: str, check: Callable[[Any], Union[float, bool]], *, weight: float = 1.0, mandatory: bool = True, description: str = "", ) -> "Workflow": """Declare a verifiable goal. ``check(state)`` must use evidence, not a tool's own claim.""" if any(g.name == name for g in self.goals): raise ValueError(f"goal {name!r} already declared") self.goals.append(Goal(name, check, weight, mandatory, description)) return self def add_predicate(self, name: str, fn: Callable[[Any], bool]) -> "Workflow": """Declare an observable predicate (used in preconditions and ``produces``).""" self.predicates[name] = fn return self def add_tool( self, name: str, fn: Callable[..., Any], *, args: Optional[Mapping[str, Any]] = None, cost: float = 1.0, gain: Optional[Mapping[str, float]] = None, harm: Optional[Mapping[str, float]] = None, produces: Optional[Mapping[str, float]] = None, requires: Iterable[str] = (), requires_goals: Iterable[str] = (), group: Optional[Hashable] = None, description: str = "", afn: Optional[Callable[..., Any]] = None, resources: Iterable[str] = (), ) -> Action: """Declare one concrete call ``fn(state, **args)`` and its declared effects. Register the same function several times with different ``args`` to offer several variants (e.g. one per retrieval source). Variants that share a ``group`` share online feedback. ``fn`` may be an ``async def`` function (async-only tool), and ``afn`` an optional async twin of a sync ``fn``. The async API awaits async implementations and runs sync ones in worker threads. """ if inspect.iscoroutinefunction(fn): fn, afn = None, fn action = Action.of(name, args) if any(t.action == action for t in self.tools): raise ValueError(f"tool call {action} already declared") sem = Semantics( gain=dict(gain or {}), harm=dict(harm or {}), produces=dict(produces or {}), requires_predicates=frozenset(requires), requires_conditions=frozenset(requires_goals), cost=float(cost), ) self.tools.append(Tool(name, fn, action, sem, group, description, afn, frozenset(resources))) return action def add_agent( self, name: str, agent: Any, *, inputs: Callable[[Any], Any], output: Callable[[Any, Any], Optional[Mapping[str, Any]]], measured_cost: Union[bool, Callable[[Any], float]] = True, budget: Optional[Callable[[Any], Optional[float]]] = None, **declared: Any, ) -> Action: """Register a sub-agent as one tool: anything with ``invoke`` (and optionally ``ainvoke``), e.g. another compiled ``Workflow``, an ``FGAORAG``, a LangGraph graph or a LangChain agent. inputs(state) -> the sub-agent's input; output(state, result) -> the update for this state. measured_cost: True charges ``result.cost`` when the result has one (FGAO sub-workflows and FGAORAG do), a function computes it from the result, False charges the declared ``cost``. budget(state) -> per-call budget passed to FGAO sub-agents (``budget=`` keyword). """ def _kw(state): b = budget(state) if budget else None return {"budget": b} if b is not None else {} def _wrap(state, res): if callable(measured_cost): cost = float(measured_cost(res)) elif measured_cost and hasattr(res, "cost"): cost = float(res.cost) else: cost = None return ToolResult(update=output(state, res), cost=cost) def fn(state, **_): return _wrap(state, agent.invoke(inputs(state), **_kw(state))) afn = None if hasattr(agent, "ainvoke"): async def afn(state, **_): # noqa: F811 return _wrap(state, await agent.ainvoke(inputs(state), **_kw(state))) return self.add_tool(name, fn, afn=afn, **declared) def tool(self, name: Optional[str] = None, **declared: Any) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """Decorator form of ``add_tool``.""" def deco(fn: Callable[..., Any]) -> Callable[..., Any]: self.add_tool(name or fn.__name__, fn, **declared) return fn return deco def validate(self) -> None: goals = {g.name for g in self.goals} if not goals: raise ValueError("declare at least one goal") if not self.tools: raise ValueError("declare at least one tool") for t in self.tools: s = t.semantics for c in list(s.gain) + list(s.harm) + list(s.requires_conditions): if c not in goals: raise ValueError(f"{t.action}: unknown goal {c!r}") for p in list(s.produces) + list(s.requires_predicates): if p not in self.predicates: raise ValueError(f"{t.action}: unknown predicate {p!r} (declare it with add_predicate)") for d in (s.gain, s.harm, s.produces): for k, x in d.items(): if not 0.0 <= x <= 1.0: raise ValueError(f"{t.action}: declared rate {k}={x} outside [0, 1]") if s.cost < 0: raise ValueError(f"{t.action}: negative cost") def compile( self, config: Optional[FGAOConfig] = None, *, budget: float = float("inf"), on_error: str = "raise", max_parallel: int = 1, independent: bool = True, ) -> "CompiledWorkflow": """Freeze the declarations into a runnable app. on_error="raise" propagates tool exceptions; "record" charges the declared cost, records the error and lets feedback learn from it. max_parallel > 1 runs up to that many tool calls concurrently per round (see ``FGAOController.select_batch``): the first is the v2 choice, further slots take calls of positive value in the expected state that, with independent=True, touch none of the batch's goals or predicates and share no ``resources``. max_parallel=1 is sequential v2. """ if on_error not in ("raise", "record"): raise ValueError("on_error must be 'raise' or 'record'") if max_parallel < 1: raise ValueError("max_parallel must be >= 1") self.validate() return CompiledWorkflow(self, config or FGAOConfig(), budget, on_error, max_parallel, independent) class CompiledWorkflow: """A runnable workflow. Every run gets its own controller, so runs are independent and can execute concurrently (``batch`` / ``abatch``) as long as the tools themselves are thread-safe.""" def __init__(self, wf: Workflow, config: FGAOConfig, budget: float, on_error: str, max_parallel: int = 1, independent: bool = True) -> None: self.name = wf.name self.config = config self.budget = budget self.on_error = on_error self.max_parallel = max_parallel self.independent = independent self._goals = list(wf.goals) self._preds = dict(wf.predicates) self._tools = {t.action: t for t in wf.tools} self._specs = [ActionSpec(t.action, t.semantics, t.group) for t in wf.tools] self.conditions = [Condition(g.name, g.weight, g.mandatory, g.description) for g in self._goals] self.async_only = [t.action for t in wf.tools if t.fn is None] @property def actions(self) -> List[Action]: return [s.action for s in self._specs] def observe(self, state: Any) -> Observation: return Observation( {g.name: _as_value(g.check(state)) for g in self._goals}, {p: bool(fn(state)) for p, fn in self._preds.items()}, ) def controller(self, *, budget: Optional[float] = None, seed: int = 0) -> FGAOController: """A fresh step-wise controller over this workflow's goals and tools.""" return FGAOController(self.conditions, self._specs, self.config, budget=self.budget if budget is None else budget, seed=seed) # ------------------------------------------------------------- execution def _call(self, tool: Tool, state: Any): ok, out = self._raw(tool, state) if not ok: return out return self._apply(tool, state, out) def _raw(self, tool: Tool, state: Any): """(True, output) or (False, (None, error)) without touching the state.""" if tool.fn is None: raise TypeError(f"{tool.action} is async-only; use ainvoke / astream / abatch") try: return True, tool.fn(state, **tool.action.kwargs) except Exception as e: # noqa: BLE001 if self.on_error == "raise": raise return False, (None, f"{type(e).__name__}: {e}") async def _araw(self, tool: Tool, state: Any): try: if tool.afn is not None: return True, await tool.afn(state, **tool.action.kwargs) return True, await asyncio.to_thread(tool.fn, state, **tool.action.kwargs) except asyncio.CancelledError: raise except Exception as e: # noqa: BLE001 if self.on_error == "raise": raise return False, (None, f"{type(e).__name__}: {e}") def _conflicts(self, a: Action, b: Action) -> bool: return bool(self._tools[a].resources & self._tools[b].resources) def _finish_round(self, ctrl, state, obs, batch, outs, callbacks): results = [] for d, (ok, out) in zip(batch, outs): # merge updates in batch order results.append(self._apply(self._tools[d.action], state, out) if ok else out) after = self.observe(state) recs = ctrl.update_batch([d.action for d in batch], after, [c for c, _ in results], [e for _, e in results]) steps = [] for d, rec, (_, err) in zip(batch, recs, results): st = Step(rec["step"], d.action, d, dict(obs.values), dict(after.values), rec["cost"], rec["budget_left"], err) steps.append(st) for cb in callbacks: cb(st) return after, steps async def _acall(self, tool: Tool, state: Any): try: if tool.afn is not None: out = await tool.afn(state, **tool.action.kwargs) else: out = await asyncio.to_thread(tool.fn, state, **tool.action.kwargs) except asyncio.CancelledError: raise except Exception as e: # noqa: BLE001 if self.on_error == "raise": raise return None, f"{type(e).__name__}: {e}" return self._apply(tool, state, out) def _apply(self, tool: Tool, state: Any, out: Any): cost, err, upd = None, None, None if isinstance(out, ToolResult): upd, cost, err = out.update, out.cost, out.error elif isinstance(out, Mapping): upd = out elif out is not None: raise TypeError(f"{tool.action} returned {type(out).__name__}; expected None, dict or ToolResult") if upd: if not isinstance(state, MutableMapping): raise TypeError(f"{tool.action} returned an update but the state is not a mutable mapping") state.update(upd) return cost, err def stream( self, state: Any, *, seed: int = 0, budget: Optional[float] = None, callbacks: Sequence[Callback] = (), ) -> Iterator[Union[Step, WorkflowResult]]: """Run step by step, yielding each executed ``Step`` and finally the ``WorkflowResult``.""" ctrl = self.controller(budget=budget, seed=seed) obs = self.observe(state) ctrl.reset(obs) steps: List[Step] = [] if self.max_parallel > 1: with ThreadPoolExecutor(max_workers=self.max_parallel) as ex: while ctrl.should_stop() is None: batch = ctrl.select_batch(self.max_parallel, self._conflicts, self.independent) outs = list(ex.map(lambda d: self._raw(self._tools[d.action], state), batch)) obs, new = self._finish_round(ctrl, state, obs, batch, outs, callbacks) steps.extend(new) yield from new yield WorkflowResult(state, ctrl.status, ctrl.status == Status.SUCCESS, dict(ctrl.v), ctrl.spent, steps) return while ctrl.should_stop() is None: d = ctrl.select_action() tool = self._tools[d.action] cost, err = self._call(tool, state) after = self.observe(state) ctrl.update(d.action, after, cost, err) rec = ctrl.history[-1] st = Step(d.step, d.action, d, dict(obs.values), dict(after.values), rec["cost"], ctrl.budget, err) obs = after steps.append(st) for cb in callbacks: cb(st) yield st yield WorkflowResult(state, ctrl.status, ctrl.status == Status.SUCCESS, dict(ctrl.v), ctrl.spent, steps) def start(self, state: Any, *, seed: int = 0, budget: Optional[float] = None, interrupt_before: Iterable[str] = (), callbacks: Sequence[Callback] = ()) -> "Run": """A step-by-step run that can pause for approval and be checkpointed (see ``Run``). A Run executes one call per step (sequential v2), whatever ``max_parallel`` is. """ return Run(self, state, seed=seed, budget=budget, interrupt_before=interrupt_before, callbacks=callbacks) def resume(self, snapshot: Mapping[str, Any], state: Any, *, interrupt_before: Iterable[str] = (), callbacks: Sequence[Callback] = ()) -> "Run": """Continue a run from ``Run.snapshot()`` with the (restored) application state.""" run = Run(self, state, seed=snapshot["seed"], budget=snapshot["controller"]["budget0"], interrupt_before=interrupt_before, callbacks=callbacks, _snapshot=snapshot) return run def invoke(self, state: Any, *, seed: int = 0, budget: Optional[float] = None, callbacks: Sequence[Callback] = ()) -> WorkflowResult: """Run to termination and return the ``WorkflowResult``.""" out = None for out in self.stream(state, seed=seed, budget=budget, callbacks=callbacks): pass return out # type: ignore[return-value] def batch(self, states: Sequence[Any], *, max_concurrency: int = 8, seed: int = 0, budget: Optional[float] = None) -> List[WorkflowResult]: """Run many independent states concurrently in threads; results keep input order.""" if max_concurrency <= 1: return [self.invoke(s, seed=seed, budget=budget) for s in states] with ThreadPoolExecutor(max_workers=max_concurrency) as ex: return list(ex.map(lambda s: self.invoke(s, seed=seed, budget=budget), states)) async def astream( self, state: Any, *, seed: int = 0, budget: Optional[float] = None, callbacks: Sequence[Callback] = (), ) -> AsyncIterator[Union[Step, WorkflowResult]]: """Native async run: async tools are awaited, sync tools run in worker threads. Makes exactly the same decisions as ``stream``; cancelling the task cancels the tool call in flight. """ ctrl = self.controller(budget=budget, seed=seed) obs = self.observe(state) ctrl.reset(obs) steps: List[Step] = [] if self.max_parallel > 1: while ctrl.should_stop() is None: batch = ctrl.select_batch(self.max_parallel, self._conflicts, self.independent) outs = await asyncio.gather(*(self._araw(self._tools[d.action], state) for d in batch)) obs, new = self._finish_round(ctrl, state, obs, batch, outs, callbacks) steps.extend(new) for st in new: yield st yield WorkflowResult(state, ctrl.status, ctrl.status == Status.SUCCESS, dict(ctrl.v), ctrl.spent, steps) return while ctrl.should_stop() is None: d = ctrl.select_action() tool = self._tools[d.action] cost, err = await self._acall(tool, state) after = self.observe(state) ctrl.update(d.action, after, cost, err) rec = ctrl.history[-1] st = Step(d.step, d.action, d, dict(obs.values), dict(after.values), rec["cost"], ctrl.budget, err) obs = after steps.append(st) for cb in callbacks: cb(st) yield st yield WorkflowResult(state, ctrl.status, ctrl.status == Status.SUCCESS, dict(ctrl.v), ctrl.spent, steps) async def ainvoke(self, state: Any, *, seed: int = 0, budget: Optional[float] = None, callbacks: Sequence[Callback] = ()) -> WorkflowResult: out = None async for out in self.astream(state, seed=seed, budget=budget, callbacks=callbacks): pass return out # type: ignore[return-value] async def abatch(self, states: Sequence[Any], *, max_concurrency: int = 64, **kw: Any) -> List[WorkflowResult]: sem = asyncio.Semaphore(max(1, max_concurrency)) async def one(s: Any) -> WorkflowResult: async with sem: return await self.ainvoke(s, **kw) return list(await asyncio.gather(*(one(s) for s in states))) @dataclass class Interrupt: """Returned by ``Run.step`` instead of executing a tool listed in ``interrupt_before``.""" step: int action: Action decision: Decision class Run: """A resumable, interruptible workflow run. run = app.start(state, interrupt_before=["send_email"]) out = run.run() # runs until done or an Interrupt if isinstance(out, Interrupt): save(run.snapshot(), state) # e.g. wait for a human run.approve() # or run.reject() to exclude that call out = run.run() ``snapshot()`` is JSON-serialisable; ``app.resume(snapshot, state)`` continues with exactly the decisions the uninterrupted run would make. The application state itself is yours to persist alongside it. """ def __init__(self, app: CompiledWorkflow, state: Any, *, seed: int, budget: Optional[float], interrupt_before: Iterable[str], callbacks: Sequence[Callback], _snapshot: Optional[Mapping] = None): self.app, self.state, self.seed = app, state, seed self.interrupt_before = set(interrupt_before) self.callbacks = list(callbacks) self.ctrl = app.controller(budget=budget, seed=seed) self.steps: List[Step] = [] self.interrupted: Optional[Interrupt] = None self._approved: Optional[Action] = None self.result: Optional[WorkflowResult] = None if _snapshot is None: self._obs = app.observe(state) self.ctrl.reset(self._obs) else: self.ctrl.restore(_snapshot["controller"]) self._obs = Observation(dict(self.ctrl.v), dict(self.ctrl.preds)) if _snapshot.get("interrupted") is not None: d = self.ctrl.select_action() if d.action != self.app.actions[_snapshot["interrupted"]]: # pragma: no cover - guarded by determinism raise RuntimeError("snapshot does not reproduce the interrupted decision") self.interrupted = Interrupt(d.step, d.action, d) if _snapshot.get("approved") is not None: self._approved = self.app.actions[_snapshot["approved"]] @property def done(self) -> bool: return self.result is not None @property def status(self) -> str: return self.result.status if self.result else ( "INTERRUPTED" if self.interrupted else self.ctrl.status) def _finish(self) -> WorkflowResult: self.result = WorkflowResult(self.state, self.ctrl.status, self.ctrl.status == Status.SUCCESS, dict(self.ctrl.v), self.ctrl.spent, list(self.steps)) return self.result def _next(self): """Decision to execute now, or an Interrupt / final result.""" if self.result is not None: return self.result if self.interrupted is not None and self._approved is None: return self.interrupted if self.ctrl.should_stop() is not None: return self._finish() d = self.ctrl.select_action() gated = d.action.tool in self.interrupt_before or str(d.action) in self.interrupt_before if gated and self._approved != d.action: self.interrupted = Interrupt(d.step, d.action, d) return self.interrupted return d def _record(self, d: Decision, cost, err) -> Step: after = self.app.observe(self.state) self.ctrl.update(d.action, after, cost, err) rec = self.ctrl.history[-1] st = Step(d.step, d.action, d, dict(self._obs.values), dict(after.values), rec["cost"], self.ctrl.budget, err) self._obs = after self.steps.append(st) self.interrupted, self._approved = None, None for cb in self.callbacks: cb(st) return st def step(self) -> Union[Step, Interrupt, WorkflowResult]: """Execute one tool call (sync); returns the Step, an Interrupt, or the final result.""" d = self._next() if not isinstance(d, Decision): return d cost, err = self.app._call(self.app._tools[d.action], self.state) return self._record(d, cost, err) async def astep(self) -> Union[Step, Interrupt, WorkflowResult]: d = self._next() if not isinstance(d, Decision): return d cost, err = await self.app._acall(self.app._tools[d.action], self.state) return self._record(d, cost, err) def run(self) -> Union[WorkflowResult, Interrupt]: """Step until the run ends or pauses.""" while True: out = self.step() if not isinstance(out, Step): return out async def arun(self) -> Union[WorkflowResult, Interrupt]: while True: out = await self.astep() if not isinstance(out, Step): return out def approve(self) -> None: """Allow the interrupted call; it executes on the next ``step`` / ``run``.""" if self.interrupted is None: raise RuntimeError("nothing to approve") self._approved = self.interrupted.action def reject(self) -> None: """Exclude the interrupted call for the rest of the run; FGAO picks something else.""" if self.interrupted is None: raise RuntimeError("nothing to reject") self.ctrl.block(self.interrupted.action) self.interrupted, self._approved = None, None def snapshot(self) -> Dict[str, Any]: """JSON-serialisable checkpoint of the orchestration (not of your application state).""" index = {a: i for i, a in enumerate(self.app.actions)} return { "workflow": self.app.name, "seed": self.seed, "controller": self.ctrl.snapshot(), "interrupted": index[self.interrupted.action] if self.interrupted else None, "approved": index[self._approved] if self._approved is not None else None, }