当前位置： 首页 > news >正文

LangGraph工作流与智能体核心模式总结

news 2025/10/24 7:58:47

LangGraph工作流与智能体核心模式总结

该教程围绕LangGraph构建智能体系统的核心场景，系统梳理了工作流（Workflows）与智能体（Agents）的关键差异，并详细拆解6种高频实现模式（提示词链、并行化、路由、编排者-工作者、评估器-优化器、智能体），每种模式均提供适用场景、实现逻辑（Graph API与Functional API）及代码示例，同时强调LangGraph在持久化、流式输出、部署等方面的底层支持，具体内容如下：

一、核心概念：工作流（Workflows）vs 智能体（Agents）

教程基于Anthropic《Building Effective Agents》博客定义两者差异，是理解后续模式的基础：

维度工作流（Workflows）智能体（Agents）
核心逻辑通过预定义代码路径编排LLM与工具，流程固定LLM动态自主控制流程与工具使用，自主决策任务执行方式
适用场景任务可拆解为固定子步骤（如标准化报告生成、固定流程处理）开放式问题、步骤不可预测的场景（如复杂编码、灵活信息检索）
控制方式代码硬编码流程，无自主决策能力LLM基于环境反馈（如工具结果、用户输入）循环决策，具备自主性

二、基础准备：环境配置与增强型LLM

所有模式均需先完成基础依赖安装与LLM配置，确保支持结构化输出与工具调用：

1. 安装依赖

pip install langchain_core langchain-anthropic langgraph

2. 初始化LLM（以Anthropic Claude为例）

import os
import getpass
from langchain_anthropic import ChatAnthropic# 配置API密钥
def _set_env(var: str):if not os.environ.get(var):os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")# 初始化Claude-3-5-Sonnet模型
llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

3. 增强型LLM：结构化输出与工具调用

LLM需通过“结构化输出”（定义Pydantic模型）或“工具绑定”（bind_tools）扩展能力，是后续模式的核心“积木”：

  • 结构化输出：让LLM按固定格式返回结果（如路由决策、评估反馈），示例：
    from pydantic import BaseModel, Field# 定义“搜索查询”结构化输出模型
class SearchQuery(BaseModel):search_query: str = Field(None, description="优化后的网页搜索查询")justification: str = Field(None, description="查询与用户需求的相关性说明")# 为LLM绑定结构化输出模型
structured_llm = llm.with_structured_output(SearchQuery)
  • 工具调用：让LLM可调用外部工具（如计算、搜索），示例：
    # 定义工具（乘法计算）
def multiply(a: int, b: int) -> int:return a * b# 为LLM绑定工具
llm_with_tools = llm.bind_tools([multiply])

三、6种核心模式：实现逻辑与适用场景

模式1：提示词链（Prompt Chaining）

核心逻辑

将任务拆解为线性序列的子步骤，每个LLM调用处理前一步输出，可插入“网关（Gate）”检查中间结果是否符合要求，不符合则跳转至优化步骤。

适用场景
  • 任务可清晰拆解为固定子步骤（如“生成笑话→检查笑点→优化笑话→最终润色”）；
  • 需通过分步处理降低单步LLM负担，提升结果准确性（用 latency 换精度）。
实现示例（Graph API）
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END# 1. 定义状态（存储任务各阶段数据）
class State(TypedDict):topic: str  # 笑话主题joke: str   # 初始笑话improved_joke: str  # 优化后笑话final_joke: str     # 最终笑话# 2. 定义节点（子步骤）
def generate_joke(state: State):# 步骤1：生成初始笑话msg = llm.invoke(f"写一个关于{state['topic']}的短笑话")return {"joke": msg.content}def check_punchline(state: State):# 网关：检查笑话是否有笑点（含“?”或“!”）return "Pass" if ("?" in state["joke"] or "!" in state["joke"]) else "Fail"def improve_joke(state: State):# 步骤2：优化无笑点的笑话msg = llm.invoke(f"给这个笑话加双关语，让它更有趣：{state['joke']}")return {"improved_joke": msg.content}def polish_joke(state: State):# 步骤3：最终润色（加意外转折）msg = llm.invoke(f"给这个笑话加一个意外转折：{state['improved_joke']}")return {"final_joke": msg.content}# 3. 构建流程
workflow = StateGraph(State)
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)# 4. 连接节点（线性+条件分支）
workflow.add_edge(START, "generate_joke")
# 条件边：检查通过则结束，失败则进入优化
workflow.add_conditional_edges("generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END})
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)# 5. 编译并调用
chain = workflow.compile()
state = chain.invoke({"topic": "cats"})
print("初始笑话：", state["joke"])
if "improved_joke" in state:print("优化后笑话：", state["improved_joke"])print("最终笑话：", state["final_joke"])

模式2：并行化（Parallelization）

核心逻辑

将任务拆解为多个独立子任务，让LLM同时处理（并行执行），最后通过“聚合器（Aggregator）”合并所有子任务结果。

适用场景
  • 子任务无依赖关系，可并行提速（如同时生成“笑话、故事、诗歌”）；
  • 需多视角结果提升置信度（如多次生成同一任务取最优，或多维度内容组合）。
关键设计
  • 并行节点：从START直接连接多个LLM调用节点，实现同时执行；
  • 聚合节点：接收所有并行节点的输出，按规则合并（如拼接文本、筛选最优）。

模式3：路由（Routing）

核心逻辑

先通过LLM（或传统分类模型）对输入进行分类，再将输入路由到对应“专项节点”处理（如“生成故事→故事节点”“生成笑话→笑话节点”），实现“分而治之”。

适用场景
  • 输入存在明确分类，且不同类别需专项处理（如用户需求分“故事、笑话、诗歌”三类）；
  • 需避免“通用处理”对特定场景的性能损耗（如优化一类输入时不影响其他类）。
实现核心
  • 路由决策：用结构化输出定义路由规则（如Pydantic模型Route，指定下一步节点）；
  • 条件边：根据路由结果动态跳转至对应节点，示例：
    from pydantic import BaseModel
from typing_extensions import Literal# 1. 定义路由结构化模型（仅允许“poem/story/joke”三类）
class Route(BaseModel):step: Literal["poem", "story", "joke"] = Field(None, description="路由目标节点")# 2. 路由节点：判断输入应跳转至哪个专项节点
def llm_call_router(state: State):router = llm.with_structured_output(Route)decision = router.invoke([SystemMessage(content="根据用户需求，路由到story/joke/poem节点"),HumanMessage(content=state["input"])])return {"decision": decision.step}# 3. 条件边：按路由结果跳转
def route_decision(state: State):if state["decision"] == "story":return "llm_call_1"  # 故事节点elif state["decision"] == "joke":return "llm_call_2"  # 笑话节点elif state["decision"] == "poem":return "llm_call_3"  # 诗歌节点

模式4：编排者-工作者（Orchestrator-Worker）

核心逻辑
  • 编排者（Orchestrator）：动态拆解任务，生成子任务列表（如“生成报告→拆解为‘引言、方法论、结论’3个子章节”）；
  • 工作者（Worker）：并行处理编排者分配的子任务；
  • 合成者（Synthesizer）：合并所有工作者的输出，生成最终结果。
适用场景
  • 任务子步骤不可预定义，需动态拆解（如编码任务中“需修改的文件数量/类型”取决于需求）；
  • 复杂任务需分工协作（如报告生成、多文档汇总）。
关键API：Send

LangGraph通过Send API动态创建工作者节点，每个工作者处理一个子任务，示例：

from langgraph.types import Send
from typing import Annotated, List
import operator# 1. 定义状态（含“待处理子任务”与“已完成子任务”）
class State(TypedDict):topic: str  # 报告主题sections: list[Section]  # 待处理章节（编排者生成）completed_sections: Annotated[list, operator.add]  # 已完成章节（工作者输出，自动合并）final_report: str  # 最终报告（合成者生成）# 2. 编排者节点：动态拆解任务为章节
def orchestrator(state: State):planner = llm.with_structured_output(Sections)  # Sections为章节列表模型report_sections = planner.invoke([SystemMessage(content="生成报告大纲"),HumanMessage(content=f"报告主题：{state['topic']}")])return {"sections": report_sections.sections}# 3. 分配工作者：为每个章节创建工作者节点
def assign_workers(state: State):return [Send("llm_call", {"section": s}) for s in state["sections"]]  # 每个章节对应一个工作者

模式5：评估器-优化器（Evaluator-Optimizer）

核心逻辑

构建“生成→评估→优化”的循环闭环：

  1. 生成器（Generator）：LLM生成初始结果；
  2. 评估器（Evaluator）：LLM按预设标准（如“笑话是否有趣”“报告是否准确”）评估结果，返回反馈；
  3. 优化器（Optimizer）：若评估不通过，生成器基于反馈迭代优化，直至符合标准。
适用场景
  • 有明确评估标准（如“笑话趣味性”“回答准确性”）；
  • 需迭代优化提升结果质量（如文案润色、报告修订）。
评估逻辑示例
from pydantic import BaseModel
from typing_extensions import Literal# 1. 定义评估结构化模型
class Feedback(BaseModel):grade: Literal["funny", "not funny"] = Field(description="笑话是否有趣")feedback: str = Field(description="不有趣时的优化建议")# 2. 评估器节点：判断笑话是否符合标准
def llm_call_evaluator(state: State):evaluator = llm.with_structured_output(Feedback)grade = evaluator.invoke(f"评价这个笑话：{state['joke']}")return {"funny_or_not": grade.grade, "feedback": grade.feedback}# 3. 循环逻辑：评估不通过则重新生成
def route_joke(state: State):return "Accepted" if state["funny_or_not"] == "funny" else "Rejected + Feedback"# 流程连接：生成器→评估器→（通过则结束/不通过则返回生成器）
workflow.add_edge("llm_call_generator", "llm_call_evaluator")
workflow.add_conditional_edges("llm_call_evaluator", route_joke, {"Accepted": END, "Rejected + Feedback": "llm_call_generator"})

模式6：智能体（Agent）

核心逻辑

LLM基于环境反馈（如工具结果、用户输入）在“思考→工具调用→结果处理”的循环中自主决策，无需预定义流程，是最灵活的模式。

适用场景
  • 开放式问题，步骤不可预测（如“解决数学题”“自主检索信息回答复杂问题”）；
  • 需LLM自主选择工具、判断何时停止（如多步计算、多轮搜索）。
实现核心：工具调用循环
from langgraph.graph import MessagesState
from langchain_core.messages import ToolMessage# 1. LLM节点：决策是否调用工具
def llm_call(state: MessagesState):return {"messages": [llm_with_tools.invoke(state["messages"])]}# 2. 工具节点：执行LLM指定的工具调用
def tool_node(state: dict):result = []for tool_call in state["messages"][-1].tool_calls:tool = tools_by_name[tool_call["name"]]  # tools_by_name为工具名→工具的映射observation = tool.invoke(tool_call["args"])result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))return {"messages": result}# 3. 循环逻辑：有工具调用则执行，无则结束
def should_continue(state: MessagesState) -> Literal["Action", END]:last_msg = state["messages"][-1]return "Action" if last_msg.tool_calls else END# 流程连接：LLM→（有工具调用→工具节点→LLM；无则结束）
workflow.add_edge(START, "llm_call")
workflow.add_conditional_edges("llm_call", should_continue, {"Action": "environment", END: END})
workflow.add_edge("environment", "llm_call")
预构建智能体：create_react_agent

LangGraph提供预构建函数，简化智能体创建：

from langgraph.prebuilt import create_react_agent# 直接创建ReAct智能体（无需手动定义节点与边）
pre_built_agent = create_react_agent(llm, tools=tools)
# 调用智能体（计算“3+4”）
messages = pre_built_agent.invoke({"messages": [HumanMessage(content="Add 3 and 4.")]})

四、LangGraph的底层支持能力

无论采用哪种模式，LangGraph均提供以下核心能力，降低开发与部署门槛：

  1. 持久化（Persistence）：
    • 人机协同（Human-in-the-Loop）：支持流程中断与人工审批（如工具调用前需人工确认）；
    • 记忆功能：支持短期对话记忆（多轮上下文）与长期记忆（跨会话状态存储）。
  2. 流式输出（Streaming）：支持流式返回中间结果（如LLM逐token输出、工具调用过程），提升用户体验。
  3. 部署与可观测性：
    • 部署：支持本地服务器、LangGraph Platform等多种部署方式；
    • 可观测性：集成LangSmith，追踪流程执行轨迹、状态转换，便于调试与评估。

五、总结

该教程通过“概念定义→基础准备→模式拆解→底层能力”的逻辑，覆盖了LangGraph构建智能体系统的核心场景：

  • 工作流模式（提示词链、并行化、路由、编排者-工作者）适合流程固定、可拆解的任务，强调稳定性与效率；
  • 智能体模式适合开放式、步骤不可预测的任务，强调自主性与灵活性；
  • 评估器-优化器模式可作为“增强模块”，嵌入其他模式提升结果质量。

所有模式均基于LangGraph的“状态图（StateGraph）”核心，通过节点（子任务）、边（流程连接）、状态（数据存储）的组合实现，同时依托LangSmith的可观测性与持久化能力，支持从开发到生产的全流程落地。

http://www.dtcms.com/a/519800.html

相关文章：

  • redis常见问题解决
  • 网站顶部有空白网络营销是什么时候出现的
  • NFS文件存储
  • 07_朴素贝叶斯
  • 【源码项目】简单实现的WPF浏览器,有兴趣的可以自己扩展（带源码）
  • 大连做网站哪家好一点商务网站建设用的是什么软件
  • Mybatis一级缓存
  • Java+OpenCV实现图片切割
  • Linux云计算基础篇（24）-PXE批量安装和Kickstart工具
  • 从零搭建 RAG 智能问答系统 6：Text2SQL 与工作流实现数据库查询
  • 创建Django项目
  • 注册个人网站的方法建设企业网站费用
  • 初识C语言13.自定义类型（联合体与枚举）
  • WebRTC入门指南：实时通信零基础
  • 深圳外贸营销型网站建设引擎搜索网站
  • Django 表单验证详解Form
  • k8s相关学习
  • 【2026计算机毕业设计】基于Django的社区婴幼儿预防接种系统
  • D030知识图谱科研文献论文推荐系统vue+django+Neo4j的知识图谱|论文本文相似度推荐|协同过滤
  • pg库、MySQL库和sqlite3基本操作
  • Android Studio新手开发第三十天
  • 住房建设部官方网站公示公告手机网站内容模块
  • 网站建设 数据库连接网站建设公司软件开
  • 深入理解栈与队列：核心特性与实战应用
  • 如何建一个免费的网站南昌做网站优化哪家好
  • 机器学习概念，算法原理及应用
  • Pytorch常用层总结
  • 【Deepseek OCR】重磅测试，mac环境下的体验【本人已经本地实验成功】
  • 轻量化（Lightweight）概念
  • [人工智能-大模型-58]：模型层技术 - 深度神经网络的本质是一个复杂的复合数学函数