LangManus 是一个基于 LangGraph 构建的多 Agent 协作系统,通过专业化的角色分工和工作流编排,实现复杂任务的智能处理。
系统架构
LangManus 采用了以图为中心的 Agent 协作架构,主要包含以下组件:
- Agent 层: 多个专业化 Agent(研究员、程序员、浏览器操作员等)
- 工具层: 为各 Agent 提供能力的工具集
- LLM 层: 差异化配置的大语言模型
- 协调层: 工作流编排和任务调度
- 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)
工作流程:
- 接收用户输入并应用协调员提示模板
- 使用基础型 LLM 处理请求
- 检测响应中是否包含
handoff_to_planner
- 若包含则转给 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",
)
工作流程:
- 根据是否启用深度思考模式选择 LLM 类型
- 若启用了规划前搜索,则先进行背景搜索
- 生成详细的 JSON 格式执行计划
- 修复和规范化 JSON 输出
- 将计划添加到状态,转交 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})
工作流程:
- 预处理历史消息,格式化之前 Agent 的响应
- 使用结构化输出要求返回特定格式的路由决策
- 分析输出,决定下一步交由哪个 Agent 处理
- 如果返回
"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",
)
工作流程:
- 使用 ReAct 框架(思考-行动-观察循环)执行研究
- 通过搜索获取相关信息
- 抓取识别到的关键网页
- 合成研究发现,整理成结构化结果
- 将结果传递给 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",
)
工作流程:
- 分析编程需求和上下文
- 使用 ReAct 框架动态编写和执行代码
- 对结果进行评估和修正
- 以 Markdown 格式记录过程和结果
- 将结果传递给 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",
)
工作流程:
- 接收自然语言形式的浏览指令
- 使用视觉型 LLM 理解网页内容
- 通过 Browser 工具执行导航、点击等操作
- 生成操作过程的 GIF 动画记录
- 返回操作结果和屏幕截图
独特特点:
- 唯一使用视觉型 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",
)
工作流程:
- 分析所有 Agent 的工作成果
- 应用专业报告提示模板
- 使用基础型 LLM 生成报告
- 以 Markdown 格式构建报告
- 将报告提交给 Supervisor
报告特点:
- 结构化:包含摘要、关键发现、详细分析、结论
- 客观性:严格基于已有信息,避免臆测
- 专业性:采用清晰、精确的表达方式
1.8 Agent 间协作模式
LangManus 的 Agent 协作遵循以下模式:
分阶段执行:
- Coordinator → Planner → Supervisor → 专业 Agent → Supervisor → … → Reporter
集中决策:
- Supervisor 作为中央控制器,决定每步调用哪个 Agent
- 每个专业 Agent 完成任务后都返回给 Supervisor
状态共享:
- 所有 Agent 共享同一个状态对象
- 每个 Agent 的输出被添加到状态中,供后续 Agent 访问
能力互补:
- Researcher 负责信息收集但不做计算
- Coder 负责所有数学计算和代码执行
- Browser 处理需要直接网页交互的任务
- Reporter 整合所有结果形成最终输出
这种分工明确、协作紧密的 Agent 团队设计,使 LangManus 能够高效处理复杂任务,充分发挥各个 Agent 的专业优势。
2. 工作流图 (src/graph/
)
工作流的执行路径并不是预定义的,而是动态确定的,主要由以下部分控制:
主要执行路径
- 初始路径:START→coordinator
- 规划路径:coordinator→planner→supervisor
- 执行路径:supervisor→{researcher|coder|browser|reporter}→supervisor→ …
- 结束路径:supervisor→end
执行流程细节
协调阶段:
- Coordinator接收用户输入
- 判断是否需要复杂任务处理
- 简单问题直接回答,复杂任务转给Planner
规划阶段:
- Planner创建详细执行计划
- 可选择是否启用深度思考
- 可选择是否进行计划前搜索
- 生成结构化JSON计划
执行循环:
- Supervisor分析当前状态和计划
- 决定下一步执行的Agent
- 选定Agent执行任务
- 结果返回Supervisor
- 循环直到任务完成
结束条件:
当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()
工作流程:
- Coordinator 首先处理用户输入
- 视情况交给 Planner 创建执行计划
- Supervisor 基于计划决定下一步执行的 Agent
- 专业 Agent 执行任务并返回结果
- 循环直到 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 执行流程
典型的执行流程如下:
用户输入处理:
- 用户通过 API 发送请求
- Coordinator 接收并初步分析
- 对于复杂任务,转交给 Planner
计划阶段:
- Planner 创建详细执行计划
- 可选择是否执行预搜索 (
search_before_planning
) - 输出结构化 JSON 计划
执行阶段:
- Supervisor 分析当前状态,决定调用哪个 Agent
- 选中的 Agent 执行任务并返回结果
- 结果添加到状态中
- 控制权返回给 Supervisor 做下一步决策
完成阶段:
- 当 Supervisor 判断任务完成,返回
{"next": "FINISH"}
- 最终结果通过 API 返回给用户
- 当 Supervisor 判断任务完成,返回
7 特色功能
深度思考模式: 通过
deep_thinking_mode
标志启用,使用更强大的推理型 LLM计划前搜索: 通过
search_before_planning
标志启用,在制定计划前先进行搜索浏览器自动化: 使用专门的浏览器工具实现网页交互
流式响应: 支持实时流式输出 LLM 响应
模块化设计: 每个 Agent 有明确的职责和能力
总结
LangManus 是一个复杂任务处理系统,通过多个专业化 Agent 的协作来解决问题。其核心优势在于:
- 职责明确分工: 每个 Agent 专注于特定任务
- 灵活的工作流: 动态决策下一步执行的 Agent
- 差异化 LLM 使用: 根据任务复杂度选择合适的 LLM
- 工具丰富: 提供多种工具增强 Agent 能力
- 流式体验: 支持实时响应
这种基于 LangGraph 的多 Agent 协作架构,为构建复杂 AI 应用提供了强大而灵活的框架。