从零构建可观测的Agent执行图

为了更好地理解 LangChain 和 LangGraph 的内部执行机制，我实现了一个轻量级的图执行框架，内置日志和可观测性。该框架不再将 Agent 视为单个执行循环，而是将其工作流显式地表示为模块化节点和转换的图结构。这使得执行流程在单个节点的粒度上变得可观测。

该图的第一个版本支持三种基本执行机制：

节点执行和有向边

扇出和扇入执行

该设计遵循 Bloch 所描述的原则——"所有程序员都是 API 设计者。好的程序是模块化的，模块间的边界定义了 API。好的模块会被复用。"因此，该图将每个节点视为一个独立的计算单元，具有良好的、面向图其余部分的接口定义。

为简单起见，每个节点最初被限制为单个传出转换。因此，顺序执行可以通过返回单个节点名称来表示，而更复杂的执行模式则通过路由或扇出机制显式表示。

每个节点接收当前图状态并返回一个 StateUpdate，其中包含应修改的状态字段。同样的原则也应用于并行执行。扇出操作可以对同一个输入状态执行多个节点，之后收集并合并它们的结果状态更新。扇入机制负责合并这些更新，同时防止意外的状态覆盖。

因此，图结构可以用节点、边、路由器、扇出和扇入的集合来表示：

@dataclass
class Edge:
    source: str
    target: str

@dataclass
class FanOut:
    source: str
    targets: list[str]

@dataclass
class FanIn:
    sources: list[str]
    target: str

@dataclass
class StateUpdate:
    values: dict[str, Any] = field(default_factory=dict)
    observation: dict[str, Any] = field(default_factory=dict)

@dataclass
class Node:
    name: str
    func: Callable[[Any], Awaitable[StateUpdate]]

在定义数据类之后，我们通过一组统一函数实现了图本身，用于添加节点、边、路由器、扇出和扇入操作。每个函数在添加相应的图组件时执行类型检查和基本验证，确保尽早检测到无效的图结构。

class Graph:
    def __init__(self):
        self.nodes: dict[str, Node] = {}
        self.edges: dict[str, Edge] = {}
        self.routers: dict[str, Callable] = {}
        self.fan_out: dict[str, FanOut] = {}
        self.fan_in: dict[str, FanIn] = {}
        self.compiled: CompiledGraph | None = None

    def add_node(self, name, func):
        self.nodes[name] = Node(name, func)

    def add_edge(self, source, target):
        if source in self.edges:
            raise ValueError(f"Node '{source}' already has an edge")
        self.edges[source] = Edge(source, target)

    def add_conditional_edges(self, source, router):
        if source in self.routers:
            raise ValueError(f"Node '{source}' already has a router")
        self.routers[source] = router

    def add_fanout(self, source, targets):
        if source in self.fan_out:
            raise ValueError(f"Node '{source}' already has a fanout")
        self.fan_out[source] = FanOut(source, targets)

    def add_fanin(self, sources, target):
        if target in self.fan_in:
            raise ValueError(f"Target '{target}' already has a fanin")
        self.fan_in[target] = FanIn(sources, target)

图通过注册节点及其对应的转换来构建。节点和边以其各自的数据类表示形式作为字典来维护。路由器单独表示为可调用对象，因为其输出取决于当前执行状态。验证会检查图的结构正确性。例如，图必须包含有效的 START 和 END 节点，每条边必须引用在图内显式定义的节点。这确保了每条边的源节点和目标节点都是有效的，防止执行引擎在运行时尝试遍历未定义的节点。

    def _validate(self):
        if "START" not in self.nodes:
            raise ValueError("START node is required")

        if "END" not in self.nodes:
            raise ValueError("END node is required")

        for source, edge in self.edges.items():
            if source not in self.nodes:
                raise ValueError(f"Unknown source node: {source}")

            if edge.source != source:
                raise ValueError(
                    f"Edge source mismatch: {source} != {edge.source}"
                )

            if edge.target not in self.nodes:
                raise ValueError(f"Unknown target node: {edge.target}")

        for source, fanout in self.fan_out.items():
            if source not in self.nodes:
                raise ValueError(f"Unknown fan-out source: {source}")

            if fanout.source != source:
                raise ValueError(
                    f"Fan-out source mismatch: {source} != {fanout.source}"
                )

            for target in fanout.targets:
                if target not in self.nodes:
                    raise ValueError(f"Unknown fan-out target: {target}")

当执行编译后的图时，初始状态在运行时传入图执行器。因此，状态属于单个图执行，而不是属于 LLM Agent 本身。图执行器负责确定下一步应执行哪个图元素，并通过 _execute_node 将其传递执行。为统一每个节点的响应，除终端 END 节点外，每个节点都应返回一个 StateUpdate。空的值字典表示节点未修改图状态的有效执行。

@dataclass
class CompiledGraph:
    nodes: dict[str, Node]
    edges: dict[str, str]
    routers: dict[str, Callable[[Any], str | list[str]]]
    fanouts: dict[str, list[str]]
    fanins: dict[str, str]
    repo: ObservationRepository
    start: str = "START"
    end: str = "END"
    trace_id: str = None

    async def run(self, state):
        return await self._execute(self.start, state)

    async def _execute_node(self, name, state):
        try:
            node = self.nodes[name]
            start = time.perf_counter()
            update = await node.func(state)
            elapsed = time.perf_counter() - start
            await self._record_observation(
                name=name,
                elapsed=elapsed,
                update=update,
            )
            return update

        except Exception as e:
            await self._record_observation(
                name=name,
                elapsed=time.perf_counter() - start,
                update=StateUpdate(),
                success=False,
                error=str(e),
            )
            raise

    async def _execute(self, name, state):
        if name == self.end:
            return state

        update = await self._execute_node(name, state)
        state = self._apply_update(state, update)

        if name in self.routers:
            targets = self.routers[name](state)

            if isinstance(targets, str):
                return await self._execute(targets, state)

            return await self._execute_fanout(name, targets, state)

        if name in self.fanouts:
            return await self._execute_fanout(name, self.fanouts[name], state)

        edge = self.edges.get(name)

        if edge is None:
            raise ValueError(f"Node '{name}' has no outgoing edge")

        return await self._execute(edge.target, state)

    def _apply_update(self, state, update: StateUpdate):
        for field, value in update.values.items():
            if field not in state.model_fields:
                raise ValueError(f"Unknown state field: '{field}'")
            setattr(state, field, value)

        return state

当遇到扇出时，图执行器解析目标节点并通过相同的 execute_node 函数执行它们。因此，扇出不需要第二份日志、异常处理或观测逻辑的实现。

    async def _execute_fanout(self, source, targets, state):
        results = await asyncio.gather(
            *[self._execute_node(target, state) for target in targets]
        )

        update = self._merge_updates(results)
        state = self._apply_update(state, update)

        fanin = self.fanins.get(source)
        if fanin is None:
            return state

        return await self._execute(fanin.target, state)

    def _merge_updates(self, updates: list[StateUpdate]):
        fields = {}
        observations = []

        for update in updates:
            for field, value in update.values.items():
                if field in fields:
                    raise ValueError(f"State conflict: '{field}'")
                fields[field] = value

            if update.observation is not None:
                observations.append(update.observation)

        return StateUpdate(
            values=fields,
            observation={"fanout": observations} if observations else None,
        )

由于所有节点执行都必须经过 _execute_node，因此可以轻松地一致记录每个节点执行期间发生的情况。这提供了一个单一的观测点，我们可以在此处捕获诸如节点名称、输入状态、执行时间、返回的状态更新、执行状态以及发生的任何异常等信息。更重要的是，这种方法确保了可观测性均匀地应用于所有节点，无需每个节点实现自己的日志或监控逻辑。

    async def _record_observation(
        self,
        name: str,
        elapsed: float,
        update: StateUpdate | None = None,
        success: bool = True,
        error: str | None = None,
    ):
        def _jsonable(value):
            if hasattr(value, "model_dump"):
                return value.model_dump(mode="json")
            return value
        metadata = update.observation
        metadata = {
            key: _jsonable(value)
            for key, value in metadata.items()
        }

        kwargs = {
            "node": name,
            "duration_ms": elapsed * 1000,
            "success": success,
            "observations": metadata,
            "error": error,
        }

        if self.trace_id is not None:
            kwargs["trace_id"] = self.trace_id

        record = AgentObservation(**kwargs)

        await self.repo.create(record)

        if self.trace_id is None:
            self.trace_id = record.trace_id

通过这个实现，我更清楚地理解了如何用相对简单的执行原语构建 Agent 图。尽管当前的实现是有意简化的，但相同的原则可以扩展到更复杂的 Agent 工作流。节点提供计算单元，边定义执行拓扑，路由器提供条件转换，扇出/扇入机制提供并行执行和状态聚合。