实用指南：多智能体架构实地操作手册

本指南将逐步构建从原始材料到可运行系统的完整流程，内容为《多智能体架构实用指南》。重点在于具体的操作步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 可将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功标准，杜绝默许的半完成状态。

1. 层次化多智能体系统

在处理“1层递归多智能体系统”阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。

from fundamental_analysis_tools import evaluate_fundamentals
from langchain.agents import create_agent
from technical_analysis_tools import technical_analysis

agent = create_agent(
    model=llm,
    tools=[technical_analysis, evaluate_fundamentals]
)

Supervisor
├── Technical Analysis Tool
└── Fundamental Analysis Tool

Supervisor
├── Technical Analyst Agent
│   ├── Tool A
│   ├── Tool B
│   └── ...
└── Fundamental Analyst Agent
    ├── Tool C
    ├── Tool D
    └── ...

from langchain.tools import tool
from langgraph_supervisor import create_supervisor

@tool
def get_weather(city: str) -> str:
    """Use this tool to get the weather of a city or location"""
    return f"The weather is sunny in {city}"

weather_expert = create_agent(
    model=llm,
    tools=[get_weather],
    name="weather_expert"
)

technical_analyst_agent = create_agent(
    model=llm,
    tools=[technical_analysis],
    name="technical_analyst"
)

fundamental_analyst_agent = create_agent(
    model=llm,
    tools=[evaluate_fundamentals],
    name="fundamental_analyst"
)

analysis_squad = create_supervisor(
    [technical_analyst_agent, fundamental_analyst_agent],
    model=llm,
    supervisor_name="analysis_supervisor",
    prompt=(
        "You are a team supervisor managing a fundamental analyst and a "
        "technical analyst..."
    )
)

analysis_app = analysis_squad.compile(
    name="fundamental_and_technical_analyst"
)

supervisor_graph = create_supervisor(
    [analysis_app, weather_expert],
    model=llm,
    prompt=(
        "You are a team supervisor managing a fundamental analyst, a "
        "technical analyst and a weather expert..."
    )
)

supervisor = supervisor_graph.compile()

何时应使用分层架构？

在规划“何时应使用该阶段”时，首先需明确合同条款：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，以便操作员无需查看整个流程即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。

2. 显式多智能体工作流

在处理“2个显式多智能体阶段”时，首先需明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程与恢复流程。重试、人工审核以及死信处理都是产品功能的一部分，而非后续的优化工作。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM接口。 在处理“2个显式多智能体阶段”时，首先需明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的合同。为相关成果命名，定义成功检查标准，并拒绝默许的部分完成状态。

def execute_plan(
    plan: Plan,
    state: State,
) -> Command[
    Literal[
        "fundamental_analysis_agent",
        "technical_analysis_agent",
        "respond",
    ]
]:
    gotos = [
        Send(
            step.action.agent_to_use,
            {"query": step.action.query_to_send},
        )
        for step in plan.steps
    ]

    if not gotos:
        gotos.append(
            Send("respond", {"messages": state["messages"]})
        )

    return Command(goto=gotos)

def router(state: State) -> Command[
    Literal["fundamental_analysis_agent", "technical_analysis_agent", "respond"]
]:
    # Get plan
    query = state['messages'][-1].content
    response = llm_planner.invoke(query) #invoke planner
    return execute_plan(response, state)

何时使用显式多智能体工作流

将“明确阶段的使用时机”视为可度量的指标会更为有效。在扩大范围之前，先记录一份理想的测试用例、一个故障案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解这些成本信息，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在进程中断后导致无法继续执行。

3. 智能体群

将“3个智能体集群”阶段视为可测量的结构来处理效果最佳。在扩大范围之前，先收集一份成功的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个结构就能进行审计。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致恢复失败。

from langgraph_swarm import (
    create_handoff_tool,
    create_swarm,
)

## We'll have to redefine our all agents to include the handoff tool
weather_expert = create_agent(
    model=llm,
    tools=[
        get_weather,
        create_handoff_tool(
            agent_name="technical_analyst",
            description="Transfer for technical analysis related questions"
        ),
        create_handoff_tool(
            agent_name="fundamental_analyst",
            description="Transfer for fundamental analysis related questions"
        ),
    ],
    name="weather_expert"
)
technical_analyst_agent = ...
fundamental_analyst_agent = ...

swarm_workflow = create_swarm(
    [fundamental_analyst_agent, technical_analyst_agent, weather_expert],
    default_active_agent="weather_expert" #a default agent must be specified.
)

swarm = swarm_workflow.compile()

何时应该使用智能体集群？

“何时应使用该阶段”这一概念在被视为可度量的对象时效果最佳。在扩大范围之前，需记录一份理想状态下的流程记录、一个故障案例以及回滚说明。同时将正常流程与恢复流程都记录下来。重试机制、人工审核环节以及错误处理都是产品本身的一部分，而非后续需要补充的内容。保持图表状态简洁且具有明确类型；嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续处理。将“何时应使用该阶段”视为输入与经过验证的输出之间的契约，为相关成果命名，明确成功标准，绝不允许出现无声无息的半完成状态。

4. 黑板多智能体系统

在4个Blackboard多智能体阶段中，修改代码之前需先明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境切换到共享环境时出现意外费用。对于那些会消耗资金或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。

a. 黑板

对于黑板架构，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，需经过人工审批。编译时的连接方式并不等同于业务流程的完整性。

class Blackboard(TypedDict, total=False):
    query: str

    # Problem frame
    problem_framed: bool
    ticker: str | None
    location: str | None
    wants_technical: bool
    wants_fundamental: bool

    # Specialist panels
    technical: dict | None
    fundamental: dict | None
    environmental: dict | None

    # Integrated solution
    synthesis: str | None

    # Control state
    next_knowledge_source: str | None
    cycles: int

b. 知识来源

在知识源阶段，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。 在知识源阶段，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，绝不允许出现无声无息的半完成状态。

def can_analyse_technicals(board: Blackboard) -> bool:
    return (
        board.get("ticker") is not None
        and board.get("wants_technical", False)
        and board.get("technical") is None
    )

c. 控制组件

在处理“控制组件”阶段时，首先列出相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一次大语言模型调用收费。

def control_component(board: Blackboard) -> dict:
    eligible = [
        source
        for source in KNOWLEDGE_SOURCES #these are subagents
        if source.precondition(board)
    ]
    if not eligible:
        return {"next_knowledge_source": None}
    chosen = max(
        eligible,
        key=lambda source: source.priority,
    )
    return {
        "next_knowledge_source": chosen.name,
        "cycles": board.get("cycles", 0) + 1,
    }

inspect → identify eligible specialists → activate one
       → contribute → inspect again

执行过程是刻意按顺序进行的

在处理按顺序执行的阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会出错。 将配置信息与应用程序代码分开。环境文件、密钥存储以及功能开关应集中存放于一处，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。

何时应该使用黑板架构？

在规划“何时应使用该阶段”时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 在成本较高的操作之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大型语言模型。 在规划“何时应使用该阶段”时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，杜绝无声的半完成状态。

5. 行为者-评论家（对抗式）多智能体系统

将5种演员-评论家对抗式多阶段方法视为可度量的模型时，其效果最佳。在扩大范围之前，需记录一个成功的案例、一个失败案例以及回滚说明。 在功能结果旁还需记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 要保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，且在中断后会导致流程无法继续。

actor → critic → judge
  ↑                 |
  └──── revise ─────┘

a. 演员

将“演员”阶段视为可度量的界面时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一处，这样操作人员无需查看整个结构即可进行审计。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致恢复失败。

def actor(state: AdversarialState) -> dict:
    if state.get("latest_critique") is None:
        prompt = f"Produce a first draft.\n\nTASK: {state['task']}"
    else:
        prompt = (
            "Revise the current draft.\n\n"
            f"TASK:\n{state['task']}\n\n"
            f"CURRENT DRAFT:\n{state['draft']}\n\n"
            f"CRITIQUE:\n{state['latest_critique']}\n\n"
            f"JUDGE'S PRIORITY:\n{state['latest_verdict']['focus']}"
        )

    draft = llm.invoke(prompt).content

    return {
        "draft": draft,
        "round": state.get("round", 0) + 1,
    }

b. 评论者

将“评审阶段”视为可度量的对象时，其效果最佳。在扩大范围之前，需记录一份理想状态下的完整日志、一个故障案例以及回滚说明。同时记录正常流程与恢复流程的详细信息。重试机制、人工审核环节以及错误处理都是产品本身的一部分，而非后续需要补充的内容。保持图表状态简洁且类型明确，嵌套的数据结构会掩盖具体是哪个节点修改了哪个字段，还会导致中断后无法继续处理。将“评审阶段”视为输入与已验证输出之间的契约，为相关文档命名，明确成功标准，绝不允许出现无声无息的半完成状态。

class Issue(BaseModel):
    severity: Literal["blocking", "major", "minor"]
    description: str

class Critique(BaseModel):
    issues: list[Issue]
    summary: str

def weighted_issue_score(issues: list[dict]) -> int:
    weights = {
        "blocking": 5,
        "major": 2,
        "minor": 1,
    }

    return sum(
        weights[issue["severity"]]
        for issue in issues
    )

c. 审判者

在法官阶段，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于那些会耗费资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

class Verdict(BaseModel):
    decision: Literal["accept", "revise"]
    reasoning: str
    focus: str

def judge(state: AdversarialState) -> dict:
    verdict = llm.with_structured_output(Verdict).invoke(
        [
            HumanMessage(
                content=(
                    "Evaluate the critic's findings on their merits. "
                    "Accept if only minor issues remain. "
                    "Request revision only for blocking or material problems.\n\n"
                    f"DRAFT:\n{state['draft']}\n\n"
                    f"CRITIQUE:\n{state['latest_critique']}"
                )
            )
        ]
    )

    return {
        "latest_verdict": verdict.model_dump(),
    }

为何仍需监督机制

在修改代码之前，需明确监控阶段的输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。

round 1: 12
round 2: 8
round 3: 8
round 4: 9

def watchdog(state: AdversarialState) -> dict:
    round_number = state.get("round", 0)
    scores = state.get("issue_counts", [])

    if round_number >= HARD_ROUND_CAP:
        return {
            "halt_reason": "hard round cap reached",
        }

    if len(scores) >= STALEMATE_WINDOW:
        window = scores[-STALEMATE_WINDOW:]

        if window[-1] >= window[0]:
            return {
                "halt_reason": (
                    f"stalemate detected: {window}"
                ),
            }

    return {}

最终确认需区分成功与失败

在最终确定阶段，修改代码之前必须明确成功标准、输入参数、该步骤的负责人以及退出条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。 在最终确定阶段，修改代码之前必须明确成功标准、输入参数、该步骤的负责人以及退出条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功检测标准，并杜绝无声的局部完成情况。

何时应使用对抗式架构？

在思考“何时应使用”这一阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。

选择架构

在“选择架构”阶段工作时，首先写下相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次计费相同的大型语言模型调用。

总结

在完成“最终思考”阶段时，首先需写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续的优化工作。 在成本较高的步骤之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大型语言模型。 在完成“最终思考”阶段时，首先需写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。

操作检查清单

将操作检查清单阶段视为可度量的对象，效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个故障案例以及回滚说明。

优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤出错时，故障应能指向具体的责任主体，而非复杂的流程链。

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

在预算允许的情况下，使用测试环境中的固定数据而非真实的付费 API，为关键路径添加冒烟测试。

将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。

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

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

f6f8c689c406的批处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌使用上限，并将日志存储在评估用示例文件旁边，以便后续模型更换时仍能保持对比性。