实用提示：你的代理图不应存在于 Python 中：编译方法

以下笔记为“你的智能体图谱不应使用 Python：如何从单个 YAML 文件构建多智能体工作流”提供了实用的操作路径。重点在于契约、校验规则以及可直接插入的代码占位符，而非激励性陈述。 在完成概览阶段时，首先明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。

没人会告诉你的问题

这个问题在被视为可测量的对象时最容易处理。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。 在讲解循环之前，先确定解释器及依赖项的锁定文件。笔记本电脑与持续集成环境之间的差异是API演示中最常见的隐性故障原因。

工作流程的数据表现形式

将工作流中的该阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现悄无声息的半完成状态。 在讲解循环逻辑之前，先固定解释器及依赖项的锁定文件。在笔记本电脑与持续集成环境之间切换是API演示中最常见的隐性故障来源。

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

技巧一：根据架构在运行时生成状态类

技巧1：将测试环境视为可度量的对象来构建，效果最佳。在扩大范围之前，先记录一个理想的测试用例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免从演示环境过渡到共享环境时出现意外费用。在讲解循环逻辑之前，先锁定解释器及依赖项文件。笔记本电脑与持续集成环境之间的差异是API演示中最常见的隐性故障来源。技巧1：将测试环境视为可度量的对象来构建，效果最佳。在扩大范围之前，先记录一个理想的测试用例、一个失败案例以及回滚说明。需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。

# 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
    ...
)

技巧#2：通过点号分隔的字符串引用代理，由importlib解析

在技巧2的代理引用阶段，应在修改代码之前定义输入参数、该步骤的负责人以及退出条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 建议使用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任主体，而非复杂的流程链。 将客户端构建与消息循环分开，这样就可以更换提供者而无需重写对话状态机。

impl: your_package.agents.agent_one.agent_one_fn

# 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}

技巧#3：编译器——YAML节点变为图节点

在技巧3的编译阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 将客户端构建与消息循环分开，这样即便更换提供方，也无需重写对话状态机。

# 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

路由器：以查找表实现条件分支

对于以条件分支为阶段的路由器，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在流程从演示环境切换到共享环境时出现意外费用。 将客户端构建与消息循环分开，这样就可以在不重写对话状态机的情况下更换服务提供商。 对于以条件分支为阶段的路由器，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非额外的开销。

抛光处理。

# 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)

技巧#4：自适应调用——开发者可自行定义签名格式

在实施技巧4的自适应阶段时，首先需明确合同条款：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任模块，而非复杂的流程链。 每次调用时都要记录请求ID、模型ID以及延迟时间。没有这些记录，间歇性的服务提供商错误就会被视为应用程序的缺陷。

# 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

技巧#5：按会话热切换节点（人工干预机制）

在实现“技巧5：热插拔阶段”时，首先需列出相关契约：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与已验证输出之间的契约。为相关产物命名，明确成功判定标准，并杜绝无声的半完成状态。 每次调用时都要记录请求ID、模型ID以及延迟时间。没有这些记录，间歇性的服务提供方错误就会被误认为是应用程序的故障。

# 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)
    )

这种架构实际能为你带来什么

在处理“该架构的实际功能”这一阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 每次调用都要记录请求ID、模型ID以及延迟时间。如果没有这些记录，间歇性的服务提供商错误就会被视为应用程序的缺陷。 在处理“该架构的实际功能”这一阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 需同时记录正常流程和故障恢复流程。重试机制、人工干预环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。

核心要点

在“提取成果”阶段，若能将其视为可度量的对象，效果会最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。相比庞大的脚本，应优先选择小型且可测试的单元。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。在讲解循环之前，先确定解释器及依赖项的锁定文件。在笔记本电脑与持续集成环境之间的差异是API演示中最常见的隐性故障原因。

操作检查清单

在“操作检查清单”阶段，修改代码之前需明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。

将配置置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一处，以便操作员无需查看全部架构即可进行审计。

将客户端构建与消息处理循环分开，这样在更换提供方时无需重写对话状态机。

在耗时的操作之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一次大型语言模型调用收费。

锁定依赖项的版本，并记录用于演示的镜像哈希值。可重复性比经验知识更为可靠。

将此阶段视为输入与经过验证的输出之间的契约。为相关产物命名，明确成功标准，拒绝默许的半完成状态。

在推广该技术栈之前，应先冻结版本，为关键流程记录标准输出日志，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及负责密钥轮换的明确责任人。与其展示花哨的一次性演示，不如注重扎实的可靠性。

关于2822ea5988ca的批注：请将提供商密钥移出代码仓库，设定单会话令牌上限，并将日志存储在评估用示例文件旁，以便后续模型更换时仍能保持数据可比性。