跳转至

langgraph实现六种典型Agent设计模式（纯干）

来源:https://mp.weixin.qq.com/s/aOWfTIhs2MtA2XOb0TxmBQ

img

最近Herness大火，我就在反思，我们在日常进行智能体开发的过程中，是否也在做类似的事，我们用过claude code sdk、codex sdk、copilot cli等通用agent做封装，也用过dify或者coze搭工作流，也用过langchain做过定制化助手，其实本质都是一样的： 在大模型的外层打一个脚手架，来控制输入信息的内容和质量、来控制什么时候把该给的信息传给大模型，来处理大模型的信息做拼装，力求大模型可以每次稳定、优质的输出生产级产物。

本次，我就拿langgraph，实现几个常用且典型的Agent设计模式的demo，来和大家详细讲讲这几种设计模式实现逻辑。

说到Agent，先问大家，什么是Agent？

大模型？工作流？智能体？

Agent主要可以分为两类:

一种是，workflow，以固定的流程为主，不同节点分别连接着逻辑处理、文本处理、大模型请求等，以dify、coze、n8n等为主流。

另一种是，智能体，可以实现执行过程中自主动态决策：计划、反思、调用等动作，可以完成复杂任务的执行，需要定制开发。

真正的智能体Agent架构可分为两大类：单智能体 vs 多智能体。

根据Anthropic、OpenAI、Google三大厂的agent范式定义可分为三个流派：

  • 增强X型LLM+Workflow + Agent：1+4+1种

  • Handoffs + Tools：生产模式

  • Sequential/Loop/Parallel：21种（最全）

整体看下来，用的最多的、最典型的几种，vibe coding了一个langraph agent demo来说明这几种范式的原理

img

img

单Agent设计范式

1、ReAct模式 核心思想 Reason + Act = ReAct —— 推理与行动交错进行，"边想边做" Agent 在每一步都先生成思考（Thought），再决定行动（Action），观察结果（Observation），循环往复直到完成任务。

img

流程 | Thought（思考） | 智能体分析当前情况、分解任务、制定下一步计划，或反思上一步结果 | | Action（行动） | 智能体决定采取的具体动作，通常是调用一个外部工具 | | Observation（观察） | 执行 Action 后从外部工具返回的结果，如搜索摘要或 API 返回值 |

img

伪代码

-

-

-

-

-

-

-

-

defreact_agent(user_task):    state.messages.append(user_task)
    whileTrue:        # Thought：LLM 推理，可能携带工具调用请求        reply = llm_with_tools(state.messages)        state.messages.append(reply)
        # Action + Observation：执行工具，结果写回状态        if reply.has_tool_calls():            for tool_call in reply.tool_calls:                observation = execute_tool(tool_call)                state.messages.append(observation)            continue  # 带着 Observation 再次推理
        # LLM 不再调用工具，输出最终答案        return reply.content

2、Plan-and-Solve 模式 核心思想 先全局规划，再分步执行 Agent 首先生成完整的执行计划，然后按照计划逐步执行每个步骤。适合任务步骤明确、 需要整体优化的场景。通过预先规划，Agent 能够优化整体执行路径，避免局部最优。

img

流程

| 规划（Planning） | 生成完整执行计划 | | 执行（Execution） | 按顺序执行每个步骤 | | 评估（Evaluation） | 检查结果，决定继续或重规划 | | 完成（Finish） | 返回最终结果 |

img

伪代码

-

-

-

-

-

-

-

-

-

-

-

-

-

-

-

-

defplan_and_solve_agent(user_task, max_replans=3):    plan = planner(user_task)   # 生成 JSON 步骤列表    results = []    current_step = 0
    while current_step < len(plan):        step = plan[current_step]
        # 执行当前步骤（可能触发工具调用）        execution = executor(step, previous_results=results)
        if execution.has_tool_calls():            tool_results = run_tools(execution.tool_calls)            results.append(record(step, tool_results))        else:            results.append(record(step, execution.content))
        # 评估：继续、跳过还是重规划        decision = evaluator(plan, current_step, results)        if decision == "continue":            current_step += 1        elif decision == "replan"and max_replans > 0:            plan = planner(user_task, done=results)            current_step = 0            max_replans -= 1        else:            break
    # 汇总所有步骤结果，生成最终回答    return aggregator(user_task, results)

3、Reflection 模式 核心思想 自我评估与迭代改进 Agent 生成初始输出后，通过自我反思（Reflection）评估质量，识别问题，然后进行 优化（Refinement），循环迭代直到达到质量标准。 其特点是为智能体引入一种事后（post-hoc）的自我校正循环，使智能体能够像人类 一样审视自己的工作，发现不足并进行迭代优化。完整的"执行-反思-优化"轨迹形成宝贵 的经验记录，还可以扩展至多模态输出（代码、图像等）。

img

流程 | 生成器（Generator） | 生成初始输出 | | 评估器（Evaluator） | 评估输出质量，产出反馈 | | 优化器（Refiner） | 基于反馈改进输出，进入下一轮迭代 |

img

伪代码

-

-

-

-

-

-

-

-

-

-

defreflection_agent(user_task, max_rounds=2):# 第一轮：直接生成初稿    draft = generator(user_task)    round_count = 0
while round_count < max_rounds:# 反思节点：基于原始要求 + 当前草稿，输出改进建议（不重写正文）        feedback = reflector(            original_request=user_task,            current_draft=draft,        )
# 生成节点：把反馈当作新的 Human 消息，输出完整改进稿        draft = generator(            messages=[user_task, draft, feedback_as_human_message(feedback)]        )        round_count += 1
# 最后一次 generate 的输出即为最终结果return draft

多Agent设计范式 1、Multi-agent Network（网络型） 这种架构，存在多个智能体节点，每个节点之间都和相互通信，并且相互委派，去中心化

| 拓扑结构 | 去中心化，点对点通信 | | 适用场景 | 强互动、无固定调用顺序的场景 | | 优点 | 信息、资源和任务共享，具备弹性和灵活性 | | 缺点 | 通信管理复杂，决策一致性较难保证 |

img

伪代码

-

-

-

-

-

-

-

-

-

-

-

-

-

defmulti_agent_network(user_task):    state.messages.append(user_task)
# 入口路由：首先派给最合适的 Agent    current_agent = router(user_task)   # 返回 "researcher" / "coder" / "writer"
whileTrue:        reply = current_agent.handle(state.messages)        state.messages.append(reply)
# 如果 Agent 发起了工具调用，执行后回到同一 Agent 继续if reply.has_tool_calls():            tool_results = run_tools(reply.tool_calls)            state.messages.append(tool_results)# 工具结果路由回发起工具调用的 Agent            current_agent = agent_who_called_tools(state.messages)continue
# 解析 Agent 在回复末尾写的 NEXT: 指令        next_name = parse_next_directive(reply.content)   # "researcher" / "coder" / "writer" / "FINISH"if next_name == "FINISH":return strip_directive(reply.content)
        current_agent = get_agent(next_name)

代码片段：可切换语言，无法单独设置文字格式

2、Agent Supervisor（监督者型） 一主多从，一个管理者，多个执行者，监督者可以动态调整多个执行者的角色（当然也可以固定死）

| 拓扑结构 | 星形，监督者居中协调 | | 适用场景 | 多领域任务分发、并行运行多 Agent、map-reduce 模式 | | 优点 | 层级结构清晰，易于管理和扩展 | | 缺点 | 存在单点故障和性能瓶颈风险 |

img

伪代码

-

-

-

-

-

-

-

-

-

-

-

defagent_supervisor(user_task):    state.messages.append(user_task)
    whileTrue:        # Supervisor 分析任务进度，发出 ROUTE: 指令        decision = supervisor(state.messages)
        # Supervisor 认为任务完成，直接输出最终答案        if decision.route == "FINISH":            return decision.final_answer
        # 派发给对应的工作 Agent        worker = get_worker(decision.route)     # "searcher" / "coder" / "analyst"        reply = worker.handle(state.messages)        state.messages.append(reply)
        # 工作 Agent 可能调用工具        if reply.has_tool_calls():            tool_results = run_tools(reply.tool_calls)            state.messages.append(tool_results)
        # 工作 Agent 汇报结果，Supervisor 下一轮继续决策        state.messages.append(worker_report(reply))

3、Hierarchical Agent Teams（层级型） Agent Supervisor的进阶版，多层级指挥，和公司一样，有CEO、经理、员工...

| 拓扑结构 | 多层树形，高层监督者管理低层监督者，底层为操作智能体 | | 适用场景 | 大型复杂任务、需要分布式决策的系统 | | 优点 | 便于扩展，分层管理清晰，支持分布式决策 | | 缺点 | 架构复杂，层级间通信开销较大 |

img

伪代码

-

-

-

-

-

-

-

-

-

-

-

-

-

-

-

-

-

defhierarchical_agent_teams(user_task):    state.messages.append(user_task)
    whileTrue:        # 顶层管理者决定交给哪个团队，或直接收尾        top_decision = top_supervisor(state.messages)        if top_decision.route == "FINISH":            return top_decision.final_answer
        # 把任务交给对应团队的子图（research_team / dev_team）        selected_team = get_team(top_decision.route)        team_result = run_team_subgraph(selected_team, state)        state.messages.append(team_result)

defrun_team_subgraph(team, state):    """每个团队内部也是一个监督者循环"""    whileTrue:        # 团队负责人决定由哪个成员处理        lead_decision = team.lead(state.messages)        if lead_decision.route == "FINISH":            return lead_decision.summary   # 阶段性成果返回顶层
        member = team.get_member(lead_decision.route)        reply = member.handle(state.messages)        state.messages.append(reply)
        # 成员可能调用工具        if reply.has_tool_calls():            tool_results = run_tools(reply.tool_calls)            state.messages.append(tool_results)
        # 成员执行完毕后汇报，负责人下一轮继续决策        state.messages.append(member_report(reply))

img

总结 整体看，这几中范式可以覆盖我们大部分的场景，但是很多情况，不只是使用某一种范式，而是多场景混合，比如ReAct+Planer、ReAct+Reflection，或者multi-agent套ReAct等，需要更具实际的业务或提效场景进行搭配和调整。 langgraph框架使用下来，相较于dify或coze肯定灵活性很强，而且比通用agent sdk更易微调和节点追踪，更有langfuse和langsmith这类可追溯观测工具，可以非常好的适配于生产环境开发和部署。

参考 参考：Anthropic:building-effective-agents https://www.anthropic.com/research/building-effectiveagents 参考：Google技术指南《Agentic Design Patterns》系统阐述了 21种智能体模式 参考：OpenAI:《A practical guide to building agents 》

END

img

            预览时标签不可点









































<div class="