实用笔记：多智能体协调——构建与观察多智能体系统

本指南将逐步构建从原始材料到可运行系统的完整流程，主题为“多智能体编排：使用LangGraph和LangSmith构建与观察多智能体系统”。重点在于可操作的步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 配置信息应与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，以便操作人员无需查看整个系统结构即可进行审核。

简介：什么是LangSmith智能体编排？

在完成“Introduction：什么是LangSmith阶段”这一部分时，首先列出相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续的优化工作。 在成本较高的步骤之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的LLM接口。

架构：LangGraph中的编排机制

在处理“架构与编排”阶段时，首先需明确相关约定：所需的输入参数、成功信号，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。

编排的设置方式：

在处理“编排机制如何实现”这一阶段时，首先需明确相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关产物命名，定义成功判定标准，杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在处理“编排机制如何实现”这一阶段时，首先需明确相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作员无需查看整个架构即可进行审计。

逐步可复现的示例

将“逐步可复现示例”阶段视为一个可度量的对象来处理，效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个故障案例以及回滚说明。同时记录正常流程与恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。保持图结构的状态简洁且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。

前提条件

将“前置条件”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一责任模块，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。

步骤1：环境搭建

将“第一步：环境搭建”阶段视为可度量的工作面，效果最佳。在扩大范围之前，需记录一份理想状态下的运行日志、一个故障案例以及回滚说明。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现悄无声息的半完成状态。 保持图结构扁平且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。 将“第一步：环境搭建”阶段视为可度量的工作面，效果最佳。在扩大范围之前，需记录一份理想状态下的运行日志、一个故障案例以及回滚说明。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放于一个位置，以便操作人员无需查看整个图结构即可进行审计。

uv add langgraph langchain-anthropic langsmith python-dotenv

第二步：Python代码

在“步骤2：Python阶段”，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 应将客户端构建逻辑与消息循环分离，这样在更换服务提供商时无需重写对话状态机。

import os
from typing import TypedDict, Literal
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END

# ==========================================
# 0. Load Environment Variables
# ==========================================
# This loads the API keys and LangSmith configs from the .env file
load_dotenv()

# ==========================================
# 1. Define the Shared State
# ==========================================
class AgentState(TypedDict):
    messages: list
    next_agent: str

# ==========================================
# 2. Define the Nodes (The Agents)
# ==========================================
# Initialize Claude 3.5 Sonnet
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)

def router_node(state: AgentState):
    """Acts as the router. Classifies the user query and directs it to the correct department."""
    system_prompt = SystemMessage(content=(
        "You are a router agent. Look at the user's message and classify it as either "
        "'billing' or 'technical'. Reply with ONLY the word 'billing' or 'technical'."
    ))
    response = llm.invoke([system_prompt] + state["messages"])
    classification = response.content.strip().lower()

    # Update state with the routing decision
    return {"next_agent": classification, "messages": [response]}

def billing_node(state: AgentState):
    """Handles billing-related queries."""
    system_prompt = SystemMessage(content=(
        "You are a billing support agent. Help the user with invoices, refunds, and payments. "
        "Be polite and professional."
    ))
    response = llm.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

def tech_support_node(state: AgentState):
    """Handles technical issues."""
    system_prompt = SystemMessage(content=(
        "You are a technical support agent. Help the user troubleshoot bugs, login issues, "
        "and software errors. Be analytical and helpful."
    ))
    response = llm.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

# ==========================================
# 3. Define the Routing Logic
# ==========================================
def route_decision(state: AgentState) -> Literal["billing", "technical"]:
    """Reads the state to decide which node to visit next."""
    next_agent = state.get("next_agent", "technical")
    # Claude is highly instruction-following, but we use 'in' to safely handle
    # any edge cases where it might add conversational filler.
    if "billing" in next_agent:
        return "billing"
    return "technical"

# ==========================================
# 4. Build and Compile the Graph
# ==========================================
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("router", router_node)
workflow.add_node("billing", billing_node)
workflow.add_node("technical", tech_support_node)

# Define edges
workflow.set_entry_point("router")
# The magic of orchestration: Conditional routing based on state
workflow.add_conditional_edges(
    "router",
    route_decision,
    {
        "billing": "billing",
        "technical": "technical",
    }
)

# Both specialized agents end the workflow
workflow.add_edge("billing", END)
workflow.add_edge("technical", END)

# Compile the graph
app = workflow.compile()

# ==========================================
# 5. Run the Orchestration
# ==========================================
if __name__ == "__main__":
    # Test Case 1: Billing Query
    print("--- Running Billing Test ---")
    inputs = {"messages": [HumanMessage(content="I was charged twice for my subscription!")]}
    result = app.invoke(inputs)
    print(result["messages"][-1].content)
    print("\n")

    # Test Case 2: Tech Support Query
    print("--- Running Tech Support Test ---")
    inputs = {"messages": [HumanMessage(content="My app keeps crashing when I click the save button.")]}
    result = app.invoke(inputs)
    print(result["messages"][-1].content)

步骤3：运行脚本进行测试

在第三步“运行阶段”中，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任主体，而非错综复杂的流程。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。

uv run multiagent-orchestration.py

第四步：在 LangSmith 中查看编排流程

在进入第4步“查看阶段”之前，需先明确输入参数、该步骤的负责人以及结束标准，然后再进行代码修改。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 应将此阶段视为输入与已验证输出之间的契约。为相关工件命名，设定成功检测标准，并杜绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接关系并不等同于业务上的完整性。 在进入第4步“查看阶段”之前，需先明确输入参数、该步骤的负责人以及结束标准，然后再进行代码修改。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 应将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。

在 LangSmith 中你会看到什么：

在“你将看到什么”这一阶段，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 在成本较高的步骤之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大型语言模型。

结论

操作检查清单