LangManus 是一个基于 LangGraph 构建的多 Agent 协作系统,通过专业化的角色分工和工作流编排,实现复杂任务的智能处理。

系统架构

LangManus 采用了以图为中心的 Agent 协作架构,主要包含以下组件:

  1. Agent 层: 多个专业化 Agent(研究员、程序员、浏览器操作员等)
  2. 工具层: 为各 Agent 提供能力的工具集
  3. LLM 层: 差异化配置的大语言模型
  4. 协调层: 工作流编排和任务调度
  5. API 层: 面向用户的接口服务

核心组件详解

1. Agent 体系 (src/agents/)

Agent 层是 LangManus 系统的核心执行单元,由一系列专业化角色组成,每个 Agent 都有明确的职责、工具集和工作模式。

系统包含多个专业 Agent,每个 Agent 有特定职责:

  • Coordinator (协调员): 用户交互入口,处理简单问答和任务分发
  • Planner (规划师): 根据用户需求创建详细执行计划
  • Supervisor (监督员): 决策下一步由哪个 Agent 执行
  • Researcher (研究员): 负责搜索和网页抓取信息收集
  • Coder (程序员): 执行 Python/Bash 代码进行数据处理
  • Browser (浏览器操作员): 自动化浏览器操作和交互
  • Reporter (报告员): 生成最终报告

1. Coordinator Agent (协调员)

职责:

  • 作为用户交互的第一入口
  • 处理打招呼和简单对话
  • 拒绝不适当的请求
  • 引导用户提供足够的上下文
  • 将复杂任务转交给 Planner

实现细节:

def coordinator_node(state: State) -> Command[Literal["planner", "__end__"]]:
    messages = apply_prompt_template("coordinator", state)
    response = get_llm_by_type(AGENT_LLM_MAP["coordinator"]).invoke(messages)
    response_content = repair_json_output(response.content)
    goto = "__end__"
    if "handoff_to_planner" in response_content:
        goto = "planner"
    return Command(goto=goto)

工作流程:

  1. 接收用户输入并应用协调员提示模板
  2. 使用基础型 LLM 处理请求
  3. 检测响应中是否包含 handoff_to_planner
  4. 若包含则转给 Planner,否则直接结束(适用于简单问答)

提示词特点:

  • 明确限定处理简单问候和闲聊
  • 包含明确的执行规则和分流标准

1.2. Planner Agent (规划师)

职责:

  • 分析复杂任务需求
  • 将大任务分解为子任务
  • 为每个子任务分配合适的 Agent
  • 创建详细的执行计划

实现细节:

def planner_node(state: State) -> Command[Literal["supervisor", "__end__"]]:
    messages = apply_prompt_template("planner", state)
    llm = get_llm_by_type("basic")
    if state.get("deep_thinking_mode"):
        llm = get_llm_by_type("reasoning")
    if state.get("search_before_planning"):
        searched_content = tavily_tool.invoke({"query": state["messages"][-1].content})
        messages[-1].content += f"\n\n# Relative Search Results\n\n{json.dumps([...])}"
    stream = llm.stream(messages)
    # ...处理和修复JSON输出...
    return Command(
        update={"messages": [HumanMessage(content=full_response, name="planner")],
                "full_plan": full_response},
        goto="supervisor",
    )

工作流程:

  1. 根据是否启用深度思考模式选择 LLM 类型
  2. 若启用了规划前搜索,则先进行背景搜索
  3. 生成详细的 JSON 格式执行计划
  4. 修复和规范化 JSON 输出
  5. 将计划添加到状态,转交 Supervisor

特殊功能:

  • 深度思考模式:使用更强大的推理型 LLM
  • 规划前搜索:先获取背景信息再规划
  • 结构化输出:产生 JSON 格式计划

1.3. Supervisor Agent (监督员)

职责:

  • 分析当前状态和任务进度
  • 决定下一步由哪个 Agent 执行
  • 判断任务是否已完成

实现细节:

def supervisor_node(state: State) -> Command[Literal[*TEAM_MEMBERS, "__end__"]]:
    messages = apply_prompt_template("supervisor", state)
    messages = deepcopy(messages)
    for message in messages:
        if isinstance(message, BaseMessage) and message.name in TEAM_MEMBERS:
            message.content = RESPONSE_FORMAT.format(message.name, message.content)
    response = (
        get_llm_by_type(AGENT_LLM_MAP["supervisor"])
        .with_structured_output(schema=Router, method="json_mode")
        .invoke(messages)
    )
    goto = response["next"]
    if goto == "FINISH":
        goto = "__end__"
    return Command(goto=goto, update={"next": goto})

工作流程:

  1. 预处理历史消息,格式化之前 Agent 的响应
  2. 使用结构化输出要求返回特定格式的路由决策
  3. 分析输出,决定下一步交由哪个 Agent 处理
  4. 如果返回 "FINISH",则结束整个工作流

关键特点:

  • 使用 JSON Schema 强制输出结构化决策
  • 作为控制中心协调整个工作流程

1.4. Researcher Agent (研究员)

职责:

  • 搜索互联网获取信息
  • 抓取和分析网页内容
  • 整合和总结研究发现
  • 提供信息支持给其他 Agent

工具集:

  • tavily_tool:互联网搜索工具
  • crawl_tool:网页内容抓取工具

实现细节:

research_agent = create_react_agent(
    get_llm_by_type(AGENT_LLM_MAP["researcher"]),
    tools=[tavily_tool, crawl_tool],
    prompt=lambda state: apply_prompt_template("researcher", state),
)

def research_node(state: State) -> Command[Literal["supervisor"]]:
    result = research_agent.invoke(state)
    response_content = repair_json_output(result["messages"][-1].content)
    return Command(
        update={"messages": [HumanMessage(content=response_content, name="researcher")]},
        goto="supervisor",
    )

工作流程:

  1. 使用 ReAct 框架(思考-行动-观察循环)执行研究
  2. 通过搜索获取相关信息
  3. 抓取识别到的关键网页
  4. 合成研究发现,整理成结构化结果
  5. 将结果传递给 Supervisor

关键特点:

  • 无法执行数学计算或编程任务
  • 专注于信息收集和整理
  • 输出 Markdown 格式的研究报告

1.5. Coder Agent (程序员)

职责:

  • 编写和执行 Python/Bash 代码
  • 执行数据分析和处理
  • 进行数学计算
  • 构建技术解决方案

工具集:

  • python_repl_tool:Python 代码执行环境
  • bash_tool:Bash 命令执行环境

实现细节:

coder_agent = create_react_agent(
    get_llm_by_type(AGENT_LLM_MAP["coder"]),
    tools=[python_repl_tool, bash_tool],
    prompt=lambda state: apply_prompt_template("coder", state),
)

def code_node(state: State) -> Command[Literal["supervisor"]]:
    result = coder_agent.invoke(state)
    response_content = repair_json_output(result["messages"][-1].content)
    return Command(
        update={"messages": [HumanMessage(content=response_content, name="coder")]},
        goto="supervisor",
    )

工作流程:

  1. 分析编程需求和上下文
  2. 使用 ReAct 框架动态编写和执行代码
  3. 对结果进行评估和修正
  4. 以 Markdown 格式记录过程和结果
  5. 将结果传递给 Supervisor

特殊能力:

  • 数据分析:使用 pandas、numpy 等库
  • 金融数据处理:通过 yfinance 访问市场数据
  • 系统交互:执行 Bash 命令进行环境操作

1.6. Browser Agent (浏览器操作员)

职责:

  • 自动化浏览器操作
  • 与网页进行交互(点击、输入、滚动)
  • 执行网页导航和检索
  • 提取网页视觉和文本信息

工具集:

  • browser_tool:基于 browser-use 的浏览器自动化工具

实现细节:

browser_agent = create_react_agent(
    get_llm_by_type(AGENT_LLM_MAP["browser"]),
    tools=[browser_tool],
    prompt=lambda state: apply_prompt_template("browser", state),
)

def browser_node(state: State) -> Command[Literal["supervisor"]]:
    result = browser_agent.invoke(state)
    response_content = repair_json_output(result["messages"][-1].content)
    return Command(
        update={"messages": [HumanMessage(content=response_content, name="browser")]},
        goto="supervisor",
    )

工作流程:

  1. 接收自然语言形式的浏览指令
  2. 使用视觉型 LLM 理解网页内容
  3. 通过 Browser 工具执行导航、点击等操作
  4. 生成操作过程的 GIF 动画记录
  5. 返回操作结果和屏幕截图

独特特点:

  • 唯一使用视觉型 LLM 的 Agent
  • 支持动态交互式网页操作
  • 提供可视化操作记录

1.7. Reporter Agent (报告员)

职责:

  • 整合所有收集的信息
  • 编写结构化、专业化的最终报告
  • 突出关键发现和见解
  • 提供清晰的结论和建议

实现细节:

def reporter_node(state: State) -> Command[Literal["supervisor"]]:
    messages = apply_prompt_template("reporter", state)
    response = get_llm_by_type(AGENT_LLM_MAP["reporter"]).invoke(messages)
    response_content = repair_json_output(response.content)
    return Command(
        update={"messages": [HumanMessage(content=response_content, name="reporter")]},
        goto="supervisor",
    )

工作流程:

  1. 分析所有 Agent 的工作成果
  2. 应用专业报告提示模板
  3. 使用基础型 LLM 生成报告
  4. 以 Markdown 格式构建报告
  5. 将报告提交给 Supervisor

报告特点:

  • 结构化:包含摘要、关键发现、详细分析、结论
  • 客观性:严格基于已有信息,避免臆测
  • 专业性:采用清晰、精确的表达方式

1.8 Agent 间协作模式

LangManus 的 Agent 协作遵循以下模式:

  1. 分阶段执行:

    • Coordinator → Planner → Supervisor → 专业 Agent → Supervisor → … → Reporter
  2. 集中决策:

    • Supervisor 作为中央控制器,决定每步调用哪个 Agent
    • 每个专业 Agent 完成任务后都返回给 Supervisor
  3. 状态共享:

    • 所有 Agent 共享同一个状态对象
    • 每个 Agent 的输出被添加到状态中,供后续 Agent 访问
  4. 能力互补:

    • Researcher 负责信息收集但不做计算
    • Coder 负责所有数学计算和代码执行
    • Browser 处理需要直接网页交互的任务
    • Reporter 整合所有结果形成最终输出

这种分工明确、协作紧密的 Agent 团队设计,使 LangManus 能够高效处理复杂任务,充分发挥各个 Agent 的专业优势。

2. 工作流图 (src/graph/)

工作流的执行路径并不是预定义的,而是动态确定的,主要由以下部分控制:

主要执行路径

  1. 初始路径:START→coordinator
  2. 规划路径:coordinator→planner→supervisor
  3. 执行路径:supervisor→{researcher|coder|browser|reporter}→supervisor→ …
  4. 结束路径:supervisor→end

执行流程细节

  1. 协调阶段

    • Coordinator接收用户输入
    • 判断是否需要复杂任务处理
    • 简单问题直接回答,复杂任务转给Planner
  2. 规划阶段

    • Planner创建详细执行计划
    • 可选择是否启用深度思考
    • 可选择是否进行计划前搜索
    • 生成结构化JSON计划
  3. 执行循环

    • Supervisor分析当前状态和计划
    • 决定下一步执行的Agent
    • 选定Agent执行任务
    • 结果返回Supervisor
    • 循环直到任务完成
  4. 结束条件

    • 当Supervisor判断任务完成

    • 返回

      {“next”: “FINISH”}

      ,转为

      end

使用 LangGraph 的 StateGraph 构建工作流:

def build_graph():
    builder = StateGraph(State)
    builder.add_edge(START, "coordinator")
    builder.add_node("coordinator", coordinator_node)
    builder.add_node("planner", planner_node) 
    builder.add_node("supervisor", supervisor_node)
    builder.add_node("researcher", research_node)
    builder.add_node("coder", code_node)
    builder.add_node("browser", browser_node)
    builder.add_node("reporter", reporter_node)
    return builder.compile()

工作流程:

  1. Coordinator 首先处理用户输入
  2. 视情况交给 Planner 创建执行计划
  3. Supervisor 基于计划决定下一步执行的 Agent
  4. 专业 Agent 执行任务并返回结果
  5. 循环直到 Supervisor 决定完成任务

3. LLM 工厂 (src/llms/)

LangManus 的 LLM 工厂是系统的核心引擎,它采用工厂模式提供差异化的大语言模型服务,根据任务特性智能匹配最合适的模型

  • 基础型 LLM (basic): 处理简单指令和普通任务
  • 推理型 LLM (reasoning): 处理复杂推理和深度思考
  • 视觉型 LLM (vision): 处理需要视觉理解的任务

支持多种 LLM 提供商:

  • OpenAI API
  • DeepSeek API
  • Azure OpenAI API
  • 通过 LiteLLM 支持的其他模型
def get_llm_by_type(llm_type: LLMType):
    # 返回缓存实例或创建新实例
    if llm_type in _llm_cache:
        return _llm_cache[llm_type]
    
    # 根据类型和配置选择实现
    if llm_type == "reasoning":
        # 创建推理型 LLM...
    elif llm_type == "basic":
        # 创建基础型 LLM...
    elif llm_type == "vision":
        # 创建视觉型 LLM...

4. 工具集 (src/tools/)

为 Agent 提供的能力的工具:

  • 研究工具:

    • tavily_tool: 网络搜索
    • crawl_tool: 网页内容抓取
  • 编码工具:

    • python_repl_tool: Python 代码执行
    • bash_tool: Bash 命令执行
  • 浏览工具:

    • browser_tool: 控制浏览器交互
  • 文件工具:

    • write_file_tool: 文件写入

5. API 服务 (src/api/)

LangManus 的 API 服务基于 FastAPI 构建,作为系统与外部世界的交互层,负责接收用户请求、调用核心工作流引擎并将结果返回给客户端。

5.1. API 架构概览

app = FastAPI(
    title="LangManus API",
    description="API for LangManus LangGraph-based agent workflow",
    version="0.1.0",
)

API 服务提供了两个主要端点:

  • /api/chat/stream

    聊天流式接口 (POST)

  • /api/browser_history/{filename}

    浏览器历史记录获取接口 (GET)

基于 FastAPI 的 HTTP 接口:

  • /api/chat/stream: 流式聊天接口
  • /api/browser_history/{filename}: 浏览器操作历史记录

支持 SSE (Server-Sent Events) 流式响应,实现实时反馈。

6 执行流程

典型的执行流程如下:

  1. 用户输入处理:

    • 用户通过 API 发送请求
    • Coordinator 接收并初步分析
    • 对于复杂任务,转交给 Planner
  2. 计划阶段:

    • Planner 创建详细执行计划
    • 可选择是否执行预搜索 (search_before_planning)
    • 输出结构化 JSON 计划
  3. 执行阶段:

    • Supervisor 分析当前状态,决定调用哪个 Agent
    • 选中的 Agent 执行任务并返回结果
    • 结果添加到状态中
    • 控制权返回给 Supervisor 做下一步决策
  4. 完成阶段:

    • 当 Supervisor 判断任务完成,返回 {"next": "FINISH"}
    • 最终结果通过 API 返回给用户

7 特色功能

  1. 深度思考模式: 通过 deep_thinking_mode 标志启用,使用更强大的推理型 LLM

  2. 计划前搜索: 通过 search_before_planning 标志启用,在制定计划前先进行搜索

  3. 浏览器自动化: 使用专门的浏览器工具实现网页交互

  4. 流式响应: 支持实时流式输出 LLM 响应

  5. 模块化设计: 每个 Agent 有明确的职责和能力

总结

LangManus 是一个复杂任务处理系统,通过多个专业化 Agent 的协作来解决问题。其核心优势在于:

  1. 职责明确分工: 每个 Agent 专注于特定任务
  2. 灵活的工作流: 动态决策下一步执行的 Agent
  3. 差异化 LLM 使用: 根据任务复杂度选择合适的 LLM
  4. 工具丰富: 提供多种工具增强 Agent 能力
  5. 流式体验: 支持实时响应

这种基于 LangGraph 的多 Agent 协作架构,为构建复杂 AI 应用提供了强大而灵活的框架。