LangChain驱动的智能体AI——第7部分：构建多智能体系统

本指南将逐步构建从原始材料到可运行系统的完整流程，内容涵盖：使用 LangChain 的智能体 AI —— 第 7 部分：使用 LangGraph 构建多智能体系统。重点在于可操作的步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。

为何多智能体如此重要

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

本文中使用的技术概念

在处理“阶段中使用的技术概念”时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。

项目概述：财务助手智能体

在处理“项目概览：财务”阶段时，首先需明确合同条款：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持透明可溯。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。 在处理“项目概览：财务”阶段时，首先需明确合同条款：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持透明可溯。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续的优化内容。

1. 系统配置

在系统配置阶段，将其视为可度量的对象会更为有效。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在流程中断后导致无法继续执行。

OPENAI_API_KEY="<Your OpenAI API Key>"
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. 项目结构

将“项目结构设计”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份优秀的成果文档、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现悄无声息的半完成状态。 保持图表状态的简洁性与类型一致性。嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在中断后导致无法继续处理。

project/
├── tools/
│   ├── __init__.py
│   ├── financial_search.py # fetch stock prices for companies
│   ├── python_REPL.py      # async wrapper of PythonREPL()
|   └── wikipedia_search.py # fetch financial summary of companies
│
├── graphs/            # implement subagents and multi-agents
│   ├── agent.py       # implement research and analyst in a single agent
│   ├── sub_agents.py  # implement research and analyst sub agents
│   ├── multi_agent_supervisor.py # implement supervisor multi-agents
│   └── multi_agent_swarm.py # implement swamp multi-agents
│
├── config.py           # pydantic BaseConfig
│── utils.py            # utility for printing and executing agents
├── .env                # environment variable definition
├── .gitignore
└── requirements.txt     # package requirements

3. 财务助手智能体的工具

将“舞台上的三种工具”视为可度量的对象使用效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 应使用结构清晰且带有明确副作用标签的工具。主机需要在自动批准之前知道哪些调用会改变系统状态。 将“舞台上的三种工具”视为可度量的对象使用效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 需同时记录正常流程和恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。

3.1. 存量数据搜索工具

在3 1 Stock Data阶段，修改代码之前需先明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。应在网关处进行身份验证，在数据层再次授权——仅凭承载令牌并不足以界定租户边界。

# define LangChain tool to retrieve stock data
@tool
async def stock_data_tool(
        company_ticker: Annotated[str, "The ticker symbol of the company to retrieve their stock performance data."],
        num_days: Annotated[int, "The number of business days of stock data required to respond to the user query."]
) -> str:

        # fetch stock data asynchronously from yfinance
        # check full code for _fetch_stock_sync function
        df = await asyncio.to_thread(_fetch_stock_sync, company_ticker, f"{num_days}d")

        return (f"Successfully executed the stock performance data retrieval tool to retrieve "
                f"the last *{num_days} days* of data for company **{company_ticker}**:\n\n{df.to_markdown()}")

3.2. Python代码执行工具

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

@tool
async def python_repl_tool(
        code: Annotated[str, "The python code to execute to generate your chart."],
) -> str:

    # wrap LLM generated code and execute it by python.REPL()
    # repl = PythonREPL().
    clean_code = textwrap.dedent(code).strip()
    result = await asyncio.to_thread(repl.run, clean_code)

    return (
        f"Successfully executed the Python REPL tool.\n\n"
        f"Python code executed:\n```python\n{code}\n```\n\n"
        f"Code output:\n```\n{result}```"
    )

3.3. 维基百科搜索工具

在3×3维基百科搜索阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在系统从演示环境切换到共享环境时出现意外费用。 在网关处进行身份验证，在数据层面重新授权。仅凭承载令牌并不足以界定租户边界。 在3×3维基百科搜索阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。

@tool()
async def wikipedia_tool(
        query: Annotated[str, "The Wikipedia search to execute to find key summary information."],
) -> str:
    """Use this to search Wikipedia for factual information asynchronously."""
    headers = {"User-Agent": "MyLangChainAgent/1.0 (contact@example.com)"}

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:

        # Step 1: Search Wikipedia for matching pages
        search_url = "https://en.wikipedia.org/w/api.php"
        search_params = {
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json"
        }
        response = await client.get(search_url, params=search_params)
        search_data = response.json()
        results = search_data.get("query", {}).get("search", [])
        title = results[0]["title"]

        # Step 2: Fetch the page extract/summary
        summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{title}"
        summary_res = await client.get(summary_url)
        summary_data = summary_res.json()
        summary = summary_data.get("extract", "No summary available.")

        return f"Successfully executed:\nWikipedia summary: {summary}"

4. 群体多智能体

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

4.1. 群体多智能体的架构

在处理4 1架构的各个阶段时，首先需明确合同条款：所需的输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM接口。

4.2. 群体多智能体实现

在处理4 2 Swarm多智能体阶段时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM接口。 在处理4 2 Swarm多智能体阶段时，首先需明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# Create a handoff tool for researcher -> analyst
transfer_to_analyst = create_handoff_tool(
    agent_name="analyst",
    description="Transfer user to the analyst assistant, who can create visualizations of provided data.",
)

# create research agent as a subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool, transfer_to_analyst],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# define the checkpointer
checkpointer = InMemorySaver()

# create swarm agents
swarm_agent = create_swarm(
    agents=[research_agent, analyst_agent],
    default_active_agent="researcher"
).compile(checkpointer=checkpointer)

5. 监管多智能体

将“5个监督者多智能体”阶段视为可度量的模型来处理效果最佳。在扩大范围之前，先记录一份成功的示例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一责任主体，而非复杂的流程链。 保持图结构的层次清晰且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在进程中断后导致无法继续执行。

5.1. 监督者多智能体架构

将舞台作品的5 1架构视为可度量的界面最为有效。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地仅完成部分工作。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。

5.2. 监管多智能体实现

将5 2 Supervisor Multi-agents阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想运行案例、一个故障案例以及回滚说明。 在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 保持图结构简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致流程无法继续。 将5 2 Supervisor Multi-agents阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想运行案例、一个故障案例以及回滚说明。 需同时记录正常流程与恢复流程的相关文档。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# create a checkpointer
checkpointer = InMemorySaver()

# create research subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# Create the supervisor multi-agent graph and compile it
supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, analyst_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research and data collection tasks to this agent\n"
        "- an analyst agent. Assign the creation of visualizations via code to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,
    output_mode="last_message"
).compile(checkpointer=checkpointer)

6. 总结

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

操作检查清单

将“操作检查清单”阶段视为可衡量的工作面，效果最佳。在扩大范围之前，需记录一份标准操作流程、一个故障案例以及回滚说明。

将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放于一处，这样操作人员无需查看整个系统结构即可进行审计。

保持系统状态的结构扁平且具有类型约束。嵌套的数据结构会掩盖具体是哪个节点修改了哪个字段，还会导致在中断后无法继续执行。

只要预算允许，就应在持续集成过程中使用测试用例而非真实的付费 API 来执行关键路径的冒烟测试。

需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。

保持系统状态的结构扁平且具有类型约束。嵌套的数据结构会掩盖具体是哪个节点修改了哪个字段，还会导致在中断后无法继续执行。