核心设计逻辑

核心架构：Supervisor 中心化路由 + 三个功能智能体循环执行，是典型的「主管 - 执行体」多智能体模式。

流程特点：所有智能体执行后均回到 Supervisor 进行下一轮决策，通过迭代次数限制防止无限循环。

容错机制：每个智能体节点都有异常捕获逻辑，出错时生成错误消息并返回 Supervisor，保证系统不会直接崩溃。

Supervisor（主管）协调的多智能体系统，包含 researcher（研究员）、coder（程序员）、coder_tester（测试员）三个核心智能体，整体是循环路由式的工作流：

定义多个智能体

multi_agents.py

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

from src.config import settings

# 模型初始化
llm = init_chat_model(
    base_url=settings.LLM_BASE_URL,
    api_key=settings.LLM_API_KEY,
    model=settings.LLM_MODEL_NAME,
    model_provider=settings.LLM_MODEL_PROVIDER,
    temperature=settings.LLM_TEMPERATURE,
    max_tokens=settings.LLM_MAX_TOKENS,
    streaming=False,
)

# 1. 产品经理 Agent
researcher = create_agent(
    model=llm,
    system_prompt="你是一个资深产品经理。只负责产品需求分析和产品功能设计。",
)

# 2. 程序员 Agent
coder = create_agent(
    model=llm,
    system_prompt="你是一个资深程序员。只负责编写代码。",
)

# 3. 测试员 Agent
coder_tester = create_agent(
    model=llm,
    system_prompt="你是一个资深测试员。只负责基于已有的代码编写测试代码。",
)

智能体控制中枢

multi_agents_supervisor.py

# multi_agent_system.py
from typing import Annotated, Literal, TypedDict

from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

from src.llm.agents_re.multi_agents import coder, coder_tester, llm, researcher


# --- 1. 定义增强版状态 ---
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    context: dict  # 例如: {"search_results": "...", "code_snippet": "..."}
    iteration_count: int  # 用于监控循环次数，防止死循环


# --- 2. 定义节点函数 (增强鲁棒性) ---
def code_node(state: AgentState) -> dict:
    """调用编码 Agent，并安全地更新状态"""
    print("[code_node] 开始执行...")
    try:
        # 只传递 messages，避免污染子 agent 的上下文
        result = coder.invoke({"messages": state["messages"]})
        new_messages = result["messages"]
        current_count = state.get("iteration_count", 0)

        print(f"[code_node] 完成，消息数: {len(new_messages)}")
        print(f"new_messages: {new_messages}")
        return {"messages": new_messages, "iteration_count": current_count + 1}
    except Exception as e:
        print(f"[code_node] 发生错误: {e}")
        # 错误处理：将错误信息作为消息加入，并终止或转交主管
        error_msg = AIMessage(content=f"Code node encountered an error: {e!s}")
        return {"messages": [error_msg], "iteration_count": current_count + 1}


def research_node(state: AgentState) -> dict:
    """调用研究 Agent，并安全地更新状态"""
    print("[research_node] 开始执行...")
    try:
        result = researcher.invoke({"messages": state["messages"]})
        new_messages = result["messages"]
        current_count = state.get("iteration_count", 0)

        print(f"[research_node] 完成，消息数: {len(new_messages)}")
        print(f"new_messages: {new_messages}")
        return {"messages": new_messages, "iteration_count": current_count + 1}
    except Exception as e:
        print(f"[research_node] 发生错误: {e}")
        error_msg = AIMessage(content=f"Research node encountered an error: {e!s}")
        return {"messages": [error_msg], "iteration_count": current_count + 1}


def test_node(state: AgentState) -> dict:
    """调用测试 Agent，并安全地更新状态"""
    print("[test_node] 开始执行...")
    try:
        result = coder_tester.invoke({"messages": state["messages"]})
        new_messages = result["messages"]
        current_count = state.get("iteration_count", 0)

        print(f"[test_node] 完成，消息数: {len(new_messages)}")
        print(f"new_messages: {new_messages}")
        return {"messages": new_messages, "iteration_count": current_count + 1}
    except Exception as e:
        print(f"[test_node] 发生错误: {e}")
        error_msg = AIMessage(content=f"Test node encountered an error: {e!s}")
        return {"messages": [error_msg], "iteration_count": current_count + 1}


# --- 3. 定义 Supervisor 逻辑 (结构化 Prompt + 容错) ---
class RouteDecision(BaseModel):
    reasoning: str = Field(description="简短说明为什么选择下一个代理，基于当前对话进度。")
    next_agent: Literal["researcher", "coder", "coder_tester", "__end__"] = Field(
        description="下一个要执行的代理。如果任务完成或无法继续，选择 '__end__'。"
    )


# 使用 ChatPromptTemplate 管理 Prompt 模板
router_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """你是一个多智能体系统的主管 (Supervisor)。
你的任务是协调 'researcher' (产品经理)、'coder' (程序员) 和 'coder_tester' (测试员) 三个代理的工作。

【当前对话历史】:
{messages}

【决策指南】:
1. 如果用户提出问题，需要产品经理分析和设计产品功能，优先分配给 'researcher'。
2. 如果需求已经输出，需要编写代码、修复错误或实现功能，分配给 'coder'。
3. 如果 'coder' 已完成代码编写，则需要写测试用例，分配给 'coder_tester'。
4. 如果任务已经完成，或者所有代理都已经充分工作且无需进一步操作，选择 '__end__'。
5. 避免无限循环：如果同一个代理刚刚执行过且没有产生实质性进展，请考虑切换代理或结束任务。

请根据以上信息做出决策。""",
        ),
        ("human", "{input_placeholder}"),  # 这里可以放一些额外的指令，或者留空让 model 只看 history
    ]
)

# 绑定结构化输出
router_llm = router_prompt | llm.with_structured_output(RouteDecision)


def supervisor_router(state: AgentState) -> Literal["researcher", "coder", "coder_tester", "__end__"]:
    messages = state["messages"]
    iteration = state.get("iteration_count", 0)

    # 安全限制：防止无限循环 (例如超过 10 次迭代强制结束)
    if iteration > 10:
        print("[supervisor_router] 达到最大迭代次数，强制结束。")
        return "__end__"

    print(f"[supervisor_router] 当前迭代次数: {iteration}, 消息总数: {len(messages)}")

    try:
        # 构造输入：将消息历史转换为字符串供 Prompt 使用
        # 注意：LangChain 的 prompt 通常会自动处理 messages 列表，但为了显式控制，我们可以传入
        response: RouteDecision = router_llm.invoke({"messages": messages, "input_placeholder": "请决定下一步动作。"})

        print(f"[supervisor_router] 决策理由: {response.reasoning}")
        print(f"[supervisor_router] 下一节点: {response.next_agent}")
        return response.next_agent

    except Exception as e:
        print(f"[supervisor_router] 路由决策失败: {e}")
        # 失败时默认结束，防止系统卡死
        return "__end__"


# --- 4. 构建图 ---
workflow = StateGraph(AgentState)

workflow.add_node("coder", code_node)
workflow.add_node("researcher", research_node)
workflow.add_node("coder_tester", test_node)


# 添加条件边：根据 route_logic 的返回值跳转到不同节点
workflow.add_conditional_edges(
    START,
    supervisor_router,
    {"researcher": "researcher", "coder": "coder", "coder_tester": "coder_tester", "__end__": END},
)


# 所有子节点执行后都回到路由决策
for node in ["researcher", "coder", "coder_tester"]:
    workflow.add_conditional_edges(
        node,
        supervisor_router,
        {"researcher": "researcher", "coder": "coder", "coder_tester": "coder_tester", "__end__": END},
    )

# 编译图
app = workflow.compile()

测试

# main.py
import os
import sys

from langchain_core.messages import HumanMessage

# 添加项目根目录到 Python 路径
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))


from src.llm.agents_re.multi_agents_supervisor import app

# 输入复杂任务
inputs = {"messages": [HumanMessage(content="设计一个简单的用户注册功能。")]}

response = app.invoke(inputs)

print(f"response: {response}")
print("\n✅ 任务完成！")