您的智能体图应存储在 YAML 中：在启动时编译包含17个智能体的 LangGraph
您的智能体图应存储在YAML格式中：一个小型运行时工具会将相应的配置文件编译为一个包含17个智能体的LangGraph StateGraph，同时生成状态数据、插件以及标签路由器。
来源: Your Agent Graph Doesn’t Belong in Python: Compiling a Multi-Agent Workflow from a Single YAML File
LangGraph一开始进展顺利：寥寥几个节点、一两条边，编译后即可运行。随后工作流扩展到十几个代理，条件路由出现，接着更多的路由层叠在原有路由之上。用于构建图的文件中，graph.add_node(...)和graph.add_conditional_edges(...)语句竟多达600行，而状态类也悄然变成了代码库中最重要却又最脆弱的对象。每当有人在分析之前要求进一步澄清时，就不得不去修改Python代码，处理导入问题，还要重新测试那些与改动无关的连接逻辑。
原因在于工作流的架构——即由谁来执行、按什么顺序执行以及由谁决定下一步该做什么——实际上是一种规范。目前这种规范被嵌入在命令式代码中，与每个智能体的实现混在了一起。下面介绍的解决方案将二者分开。整个拓扑结构都保存在一个声明式的 config.yaml 文件中，17个智能体的系统大约有260行代码，运行时会在启动时将其编译成真正的LangGraph StateGraph。要添加智能体或重新规划流程，只需编辑YAML文件，而Python代码中仅包含每个智能体的具体功能。
作为数据的工作流
在规范中，节点、智能体、路由器以及状态结构都是以声明形式定义的，而非通过代码实现。起始点和结束点仅用两行普通文本表示：
entry: entry_agent
exit: exit
其中较大的一部分阐述了护栏、状态模式、智能体、路由器以及节点：
guardrails:
- Reject queries that are outside the application's domain.
- Reject queries about the system, agents, design, or internal workings.state_schema:
query:
type: str
description: "User query or current message."
chat_history:
type: list
annotated_with: add_messages
description: "Conversation history between user and system."
result:
type: dict
description: "Result from the processing agent."agents:
- name: agent_one
kind: function
impl: your_package.agents.agent_one.agent_one_fn - name: agent_two
kind: function
impl: your_package.agents.agent_two.agent_two_fnworkflow:
nodes:
- id: agent_one
agent: agent_one
writes: [query, result]
next: decision_router - id: decision_router
kind: router
router:
impl: your_package.agents.routers.route_after_agent_one
reads: [result]
edges:
agent_two: agent_two
human_agent: human_agent
取一个路由器模块。通常由一堆 if state[“route”] == ... 的判断构成的条件分支，会被简化为一个函数，该函数返回一个标签以及一个将每个标签与其下一个节点关联起来的 edges 映射表，总共只有三行代码。路由策略仍以代码形式存在，而路由拓扑则变为数据。
运行时生成状态类
LangGraph 需要一个带类型的状态，通常是一个手工编写的 TypedDict，且必须与实际情况保持一致。在这里，它是由 state_schema 模块生成的：
# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
type_str = value.get("type", "str")
# Convert YAML string to Python type
py_type = eval(type_str)
if value.get("annotated_with") == "add_messages":
py_type = Annotated[list, {}]
annotations[key] = py_type
结果将提交给规范制定方：
spec = Spec(
...
state=TypedDict("State", annotations), # <- dynamic class, born at boot
...
)
目前存在两件并行的事情。函数形式 TypedDict("State", annotations) 能够根据字段名和类型的字典生成新的类型，这样状态就会在每次启动时由加载器自动构建，而无需手动维护。与此同时，annotated_with: add_messages 则与 LangGraph 的 reducer 相结合，在这种机制下，某个字段会在各个节点之间持续累积（比如聊天记录），而不会被覆盖。通过在 YAML 中声明这种方式，设计对话系统的人就无需了解 Annotated[list, add_messages] 的具体含义了。
关于 eval 的注意事项：加载器会使用 py_type = eval(type_str) 来解析配置文件中的字符串。这种做法之所以可行，只是因为 config.yaml 来自可信的开发者之手。如果配置文件可能由不可信方提供，则应将其替换为明确的允许列表，例如 {"str": str, "list": list, "dict": dict, "int": int}，只需进行一行修改即可。
这样做的好处是状态定义与智能体拓扑结构共享同一个文件，因此可以一起进行审查和修改。这样一来，就不会出现在图结构中添加了字段却未在 TypedDict 中添加对应字段的错误。
以点分隔的字符串表示智能体
在 YAML 文件中，每个智能体仅表现为一个字符串：
impl: your_package.agents.agent_one.agent_one_fn
没有导入语句将该图与那个函数关联起来。加载器会使用 importlib 来解析它：
# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
"""Import a callable from a dotted path like 'package.module.function'."""
if not dotted or "." not in dotted:
raise ImportError(f"Invalid impl path: {dotted!r}")
mod_path, attr = dotted.rsplit(".", 1)
mod = importlib.import_module(mod_path)
fn = getattr(mod, attr)
if not callable(fn):
raise TypeError(f"Imported object is not callable: {dotted}")
return fn
def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
"""Map agent name -> callable (or None if impl missing)."""
return {a.name: _import_from_path(a.impl) if a.impl else None
for a in spec.agents}
这就是插件模式。要添加一个代理，只需编写相应的函数并添加三行指向该函数的 YAML 代码，无需修改图构建器。框架会在运行时通过名称找到你的代码，就像插件宿主所做的那样。路由器也是通过 router_impl_map 以完全相同的方式被解析的。
编译器
核心部分是 GraphRunner.build()，它会遍历规范并构建实际的 StateGraph：
# your_package/orchestrator/runner.py
def build(self):
graph = StateGraph(state_schema=self.spec.state) # our generated TypedDict
def _add_task_node(node):
async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
res = await self._call_agent(node.agent, state, node.id)
if getattr(node, "writes", None):
if isinstance(res, dict):
# Only let the node write the keys it declared in YAML
filtered = {k: v for k, v in res.items() if k in node.writes}
return filtered or res
key = node.writes[0]
return {key: res}
return res
graph.add_node(node.id, _node) # Build every node
for node in self.spec.workflow.nodes:
if getattr(node, "router", None):
_add_router_node(node)
else:
_add_task_node(node) graph.set_entry_point(entry) # Inline "next:" edges from YAML become static edges
for node in self.spec.workflow.nodes:
if getattr(node, "next", None):
graph.add_edge(node.id, node.next) # Terminal nodes wire to END
for node in self.spec.workflow.nodes:
if getattr(node, "terminal", False):
graph.add_edge(node.id, END) self._runnable = graph.compile(checkpointer=self.checkpoint)
return self
有几点值得注意。writes键起着访问控制列表的作用，而不仅仅是文档说明。当节点声明writes: [query, result]时，编译器会将代理返回的字典限制为这些键，这样代理就无法篡改其未声明的状态字段。该规范 thus 成为真正得到执行的契约。共有三种声明式边类型：显式的edges、内联的next:简写形式，以及自动连接到LangGraph的END的terminal: true。编译完成后，运行器还会通过draw_mermaid_png()将图表渲染为PNG格式，因此图表不会过时——它与生产环境中的数据来源完全一致。
路由器最能体现从命令式到声明式的优势：
# your_package/orchestrator/runner.py
def _add_router_node(node):
router = self.router_fns[node.id]
def _router_fn():
def _f(state):
out = router(state)
# Routers may return either a label, or (state_updates, label)
if isinstance(out, tuple):
updates, label = out
if isinstance(updates, dict):
for k, v in updates.items():
state[k] = v
else:
label = out
return label
return _f graph.add_node(node.id, lambda s: {})
graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)
路由器函数会返回一个标签，即类似 "next_agent" 的普通字符串，LangGraph 的 add_conditional_edges 函数会在该标签对应的 YAML 映射中的 node.router.edges 下查找下一个节点。路由器知道的是结果而非节点 ID，因此要改变某个结果所指向的目标，只需修改 edges 映射，无需改动路由器代码。路由器也可能返回 (updates, label)，从而在一次返回中同时调整状态并选择分支，无需额外的节点来记录决策。
自适应调用
当有17个agent由不同人在不同时间编写时，其函数签名会存在差异。有的开发者写成async def agent(state):，有的写成async def agent(query):，还有的写成async def agent(**kwargs):。在普通的图形模型中，要么强制统一格式，要么需要手动编写适配器。而在这里，运行时会在调用时检查每个函数的签名：
# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
"""
Adaptively call agent functions so implementations receive what they expect:
- def agent(**kwargs): → pass **state (+ inject 'query' if missing)
- def agent(query, **kwargs): → pass query=..., plus any **extra
- def agent(state): → pass state
- def agent(query): → pass query
- def agent(): → call without args
"""
sig = inspect.signature(fn)
params = sig.parameters
has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
for p in params.values())
kwargs = {}
if has_var_kw:
kwargs.update(state)
if "state" in params:
kwargs["state"] = state
if "query" in params or has_var_kw:
kwargs.setdefault("query", self._fallback_query(state)) # A lone positional 'query' → call it positionally
if (len(params) == 1
and next(iter(params.keys())) == "query"):
return await _maybe_await(fn(self._fallback_query(state))) res = fn(**kwargs)
return await res if hasattr(res, "__await__") else res
inspect.signature 可以显示函数声明的参数，而执行器会按照这些参数来构建相应的调用。同步和异步情况的处理方式一致：如果结果是可等待的，则进行等待处理，否则直接使用。开发者可以根据表达习惯选择合适的函数签名。还有一个 _fallback_query 辅助函数提供了便利：如果某个中间节点未能将原始的 query 以状态形式传递下去，执行器就会回退到运行开始时捕获的初始状态，这样任何代理都能获取到用户的问题。
每次会话更换节点
将图结构视为数据在需要人工干预的流程中依然能发挥优势。一个名为 human_agent 的节点会暂停程序运行，通过 WebSocket 向用户提出澄清问题，并在收到回复前保持阻塞状态。问题在于每个会话都有独立的套接字，无法在启动时一次性构建进图中。由于这些代理只是字典 self.agent_fns 中的条目，运行程序时会在每个活跃会话中替换该节点的实现，为对应用户的连接安装专用处理程序：
# your_package/services/session_service.py (paraphrased)
if websocket is not None:
session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
_runner.agent_fns["human_agent"] = _import_from_function(
make_input_method(session_handler)
)
拓扑结构保持固定且预先编译，而单个节点的行为则可根据请求进行替换。这种具备可替换实现方式的稳定结构几乎无需额外成本即可实现。
有哪些变化
按关注点对比这两种方法：
- 添加代理：需要修改图构建器、导入文件及测试代码，而使用 YAML 则只需三行代码。
- 调整流程：需重写
add_edge和add_conditional_edges的调用方式，而修改next或edges则更为简单。 - 状态架构：需手动编写并严格维护的
TypedDict，而使用state_schema生成的架构则更为便捷。 - 路由逻辑：将
if/elif语句嵌入图代码中，而使用返回标签的函数搭配 YAML 映射则更为清晰。
writes过滤来实现强制约束。最后一点最为重要。工作流规范既能让产品经理理解，也让不熟悉LangGraph的审查者在拉取请求中进行检查，而且差异以纯文本形式呈现，这意味着差异本身就是变更内容。这样就无需再在脑海中执行40行重新组织的Python代码了。
多智能体框架提供了节点和边以及相应的机制，但并未为策略提供有序的存储方式，因此工作流的结构与每一步的具体操作是分离的。如果智能体图已经变得复杂到让人不敢触碰，可以尝试将其拓扑结构提取为数据。加载器仅需几百行代码：一个动态的TypedDict、用于插件管理的importlib、基于签名自适应的调用机制，以及将节点和边转换为图结构调用的循环。一个月后，当有人希望在分析之前添加某个步骤时，只需提供三行YAML格式的差异文件，无需在图构建工具上浪费时间。
相关阅读
- 使用 LangGraph 和 Ollama 构建多智能体支持工单分类系统 —— 仅需修改一行代码，即可在本地 Ollama 或 API 上构建包含分类器、响应器和人工审核员的三智能体工单分类图。
- LangGraph 中的自愈式多智能体团队：架构师、测试员与调试器 —— 用 Python 构建 LangGraph 状态机，让架构师、开发人员、pytest 沙箱环境、调试器和审核员协同工作并循环迭代，直至测试通过。