AI创想

标题: 简单学习 --＞ LangGraph [打印本页]
作者: 米落枫    时间: 3 小时前
标题: 简单学习 --＞ LangGraph
作者：CSDN博客
什么是LangGraph?

1. LangGraph 与 传统 LangChain

技术定义： LangChain 主要用于构建线性的工作流（DAG，有向无环图），数据从头传到尾。而 LangGraph 是 LangChain 的扩展框架，专门用于构建包含“循环（Cycles）”和“状态记忆（Stateful）”的多智能体（Multi-Agent）应用。
简单举例： 传统的 LangChain 像是一条单向流水线，原材料进去，经过几道工序，成品出来，不能回头。LangGraph 像是一个带环岛的交通网络或者一个带有反馈机制的办公室。员工（大模型）在遇到无法解答的问题时，可以去查资料（调用工具），查完资料后带着结果重新循环回到大脑中进行二次思考，直到问题真正解决才输出最终答案。
  1. # 传统 LangChain (单向 Pipeline，无法循环)
  2. chain = prompt | llm | output_parser
  4. # LangGraph (基于状态图，支持复杂循环)
  5. builder = StateGraph(AgentState)
  6. # ... 添加节点与边 ...
  7. graph = builder.compile()
复制代码
2. State (状态)

技术定义： 图中所有节点共享的数据结构（通常是 Python 的 TypedDict 或 Pydantic 模型）。每次节点执行完毕后，都会返回一个更新，LangGraph 会将这个更新合并到全局状态中，传递给下一个节点。
简单举例： State 就像是会议室桌子上的一块公共白板。所有的参会者（节点）都能看到白板上的历史讨论记录（上下文对话），当轮到某个人发言或工作时，他会把自己的新发现写在白板上，供下一个人参考。
  1. from typing import TypedDict
  3. # 定义这块“公共白板”上能写哪些字段
  4. class AgentState(TypedDict):
  5.     input_query: str          # 用户的初始问题
  6.     search_results: list[str] # 存放查找到的资料
  7.     final_answer: str         # 最终输出的答案
复制代码
3. Node (节点)

技术定义： 图中的处理单元，本质上就是接收 State、执行特定逻辑（如调用 LLM 或执行 Python 代码），然后返回 State 更新的普通 Python 函数。
简单举例： 节点就是公司里的具体岗位员工。比如“研究员节点”负责去搜索引擎找资料，“总结员节点”负责把资料写成报告。他们只负责自己份内的工作。
  1. # 一个典型的节点函数（员工）
  2. def researcher_node(state: AgentState):
  3.     query = state["input_query"]
  4.     # 模拟执行外部搜索
  5.     results = [f"关于 {query} 的搜索结果A", "结果B"]
  6.    
  7.     # 只需要返回需要更新到白板上的增量数据
  8.     return {"search_results": results}
复制代码
4. Edge (边) 与 Conditional Edge (条件边)

技术定义： 边定义了节点之间的执行顺序。条件边（Conditional Edge）则是一个路由函数，它根据当前的 State 动态决定下一步该走向哪个节点。
简单举例： 边就像是公司的汇报流程图。而“条件边”则是项目经理，他会根据当前白板（State）上的情况做判断：“如果资料查全了，就发给总结员（走向结束）；如果资料不够，就打回去让研究员继续查（形成循环）。”
  1. # 路由函数（项目经理）
  2. def route_logic(state: AgentState):
  3.     if len(state.get("search_results", [])) > 0:
  4.         return "summarizer" # 资料够了，去写总结
  5.     return "researcher"     # 资料不够，打回去继续查
  7. # 普通边：researcher 节点做完后，强制走到 router_node
  8. builder.add_edge("researcher", "router_node")
  10. # 条件边：根据 route_logic 的返回值动态决定去向
  11. builder.add_conditional_edges("router_node", route_logic)
复制代码
5. Reducer (聚合器)

技术定义： 定义在 State 字段上的数据合并逻辑。默认情况下，节点返回的数据会覆盖全局 State 中的同名字段；如果配置了 Reducer，LangGraph 会调用该函数，将新数据合并或追加到旧数据上。
简单举例： 如果没有 Reducer，会议白板上的字会被下一个人直接擦掉重写；配置了 Reducer，相当于在白板上划分了专门的“留言区”，新来的内容会排在旧留言下面，保留完整历史。
  1. import operator
  2. from typing import Annotated, TypedDict
  4. class AgentState(TypedDict):
  5.     # 没有 Reducer：状态被直接覆盖 (擦掉重写)
  6.     current_status: str
  7.    
  8.     # 有 Reducer (operator.add)：新列表会追加到老列表后面 (保留历史)
  9.     messages: Annotated[list[str], operator.add]
复制代码
6. Checkpointer (检查点)

技术定义： LangGraph 的状态持久层组件。它会在图的每一步执行完毕后，将完整的 State 拍一个快照（Snapshot）保存下来。这是实现时光倒流 (Time Travel) 和人类介入 (Human-in-the-loop) 的基石。
简单举例： 就像单机游戏的“自动存档”功能。如果打 Boss（调用工具）时报错了，你可以直接读档回到打 Boss 前的状态重试。
  1. from langgraph.checkpoint.memory import MemorySaver
  3. # 1. 初始化保存器 (也支持 SQLite/Postgres 等数据库)
  4. memory = MemorySaver()
  6. # 2. 编译时注入保存器
  7. app = builder.compile(checkpointer=memory)
  9. # 3. 运行时必须指定线程 ID，用于隔离不同用户的“游戏存档”
  10. config = {"configurable": {"thread_id": "user_123_session_1"}}
  11. app.invoke({"input_query": "你好"}, config=config)
复制代码
7. Subgraph (子图)

技术定义： 将一个已经编译好的 StateGraph 实例作为一个普通的 Node，无缝嵌入到另一个更大的图（父图）中。专门用于降低复杂 Agent 系统的代码耦合度。
简单举例： 就像大公司里的“部门外包”。父图是总公司，遇到复杂的财务问题时，直接把任务扔给“财务部（子图）”。财务部内部有自己的一套流转节点，搞定后直接把最终结果交回给总公司。
  1. # 1. 编译子图 (财务部门)
  2. finance_graph = StateGraph(FinanceState)
  3. # ... 给财务部添加主管、会计节点 ...
  4. finance_app = finance_graph.compile()
  6. # 2. 将子图作为普通节点，无缝接入父图 (总公司)
  7. main_graph = StateGraph(MainState)
  8. main_graph.add_node("finance_department", finance_app)
复制代码
8. Send API (并行分发)

技术定义： LangGraph 提供的用于实现动态并行执行的核心 API。在条件边中返回 Send 对象，LangGraph 会同时拉起多个相同的节点并发处理不同数据，最后汇总结果。
简单举例： 就像老师批改 50 份期末试卷。老师不自己挨个批改（串行），而是把试卷同时分发给 5 个助教（并行 Send 拉起 5 个节点），改完后再将分数统一汇总给老师。
  1. from langgraph.constants import Send
  3. def distribute_grading(state: GlobalState):
  4.     papers = state["exam_papers"]
  5.     # 动态返回多个 Send 对象，并行拉起多个 grader_node
  6.     return [Send("grader_node", {"paper_content": p}) for p in papers]
  8. # 起点直接触发分发逻辑
  9. builder.add_conditional_edges("start", distribute_grading)
复制代码
9. Command (控制指令)

技术定义： 一种在 Node 内部直接返回的特殊数据类。它允许节点在更新 State 的同时，动态指定下一步要跳转的目标节点，将“状态更新”和“路由跳转”合二为一。
简单举例： 以前员工干完活（节点），要把结果写在白板上等经理（条件边）决定下一步。现在员工自带分配权（Command），干完活不仅更新白板，还能直接指定“转交给谁干”。
  1. from langgraph.types import Command
  3. def smart_worker_node(state: AgentState):
  4.     # 干完活了
  5.     result = "数据分析报告已生成"
  6.    
  7.     # 以前只能 return {"final_answer": result}，然后靠外部判断
  8.     # 现在直接用 Command 决定去向：更新状态并强制跳到 manager_node
  9.     return Command(
  10.         update={"final_answer": result},
  11.         goto="manager_node"
  12.     )
复制代码
LangGraph的基本使用

第一步：环境安装与导入

安装 LangGraph 和基础依赖环境。
  1. # 安装基础包
  2. # pip install -U langgraph langchain-openai
  3. from typing import TypedDict, Annotated
  4. from langgraph.graph import StateGraph, START, END
  5. from langgraph.graph.message import add_messages
复制代码
第二步：定义全局状态 (State)

我们需要定义图在流转过程中传递的数据结构，最典型的用法是保存对话历史记录。
  1. # 定义状态字典，Annotated[list, add_messages] 引入了 Reducer 聚合器
  2. # 告诉框架：新消息要追加到列表中，而不是覆盖
  3. class AgentState(TypedDict):
  4.     messages: Annotated[list, add_messages]
复制代码
第三步：定义节点 (Nodes)

创建具体的处理函数。这里模拟两个节点：思考的大模型和外部工具。
  1. # 1. 模拟大模型节点：根据当前消息思考
  2. def call_llm(state: AgentState):
  3.     messages = state['messages']
  4.     last_message = messages[-1]
  5.    
  6.     if "天气" in last_message:
  7.         return {"messages": ["CALL_TOOL: weather_api"]}
  8.     else:
  9.         return {"messages": ["我是一个AI，我已经为你解答了问题。"]}
  11. # 2. 模拟工具节点：执行具体的外部查询
  12. def execute_tool(state: AgentState):
  13.     return {"messages": ["工具执行结果：今天天气晴朗，气温25度。"]}
复制代码
第四步：定义条件路由 (Conditional Edge)

写一个逻辑函数，决定大模型思考完之后，是直接结束，还是去调用工具。
  1. # 路由逻辑：判断下一步去哪里
  2. def should_continue(state: AgentState):
  3.     last_message = state['messages'][-1]
  4.    
  5.     if "CALL_TOOL" in last_message:
  6.         return "continue_to_tool"
  7.     return "end_conversation"
复制代码
第五步：构建并编译计算图 (Graph)

将状态、节点和边像拼图一样组装起来。
  1. workflow = StateGraph(AgentState)
  3. workflow.add_node("agent", call_llm)
  4. workflow.add_node("tool", execute_tool)
  6. workflow.add_edge(START, "agent")
  8. workflow.add_conditional_edges(
  9.     "agent",
  10.     should_continue,
  11.     {
  12.         "continue_to_tool": "tool",
  13.         "end_conversation": END
  14.     }
  15. )
  17. workflow.add_edge("tool", "agent")
  19. app = workflow.compile()
复制代码
第六步：运行图

向图中输入初始状态，观察数据的流转。
  1. # 测试 1：不需要工具的普通对话 (START -> agent -> END)
  2. inputs = {"messages": ["你好，请介绍一下你自己。"]}
  3. for output in app.stream(inputs):
  4.     print("当前节点输出:", output)
  6. print("---")
  8. # 测试 2：触发工具调用的复杂对话 (START -> agent -> tool -> agent -> END)
  9. inputs = {"messages": ["北京今天的天气怎么样？"]}
  10. for output in app.stream(inputs):
  11.     print("当前节点输出:", output)
复制代码
第七步：使用 Send 实现并行处理 (Map-Reduce 进阶)

在需要并行查询多个数据源时使用。
  1. from langgraph.constants import Send
  2. import operator
  4. # 状态定义：使用 operator.add 作为 Reducer 来累加多个并行节点的结果
  5. class ParallelState(TypedDict):
  6.     topics: list[str]
  7.     reports: Annotated[list[str], operator.add]
  9. def generate_report(state: str):
  10.     return {"reports": [f"【{state}】的调研报告已生成"]}
  12. def continue_to_parallel(state: ParallelState):
  13.     # 返回多个 Send 对象，LangGraph 会并行触发 generate_report 节点
  14.     return [Send("generate_report", topic) for topic in state["topics"]]
  16. parallel_workflow = StateGraph(ParallelState)
  17. parallel_workflow.add_node("generate_report", generate_report)
  18. parallel_workflow.add_conditional_edges(START, continue_to_parallel)
  19. parallel_workflow.add_edge("generate_report", END)
  21. app_parallel = parallel_workflow.compile()
复制代码
第八步：使用 Command 简化图结构路由 (高阶写法)

抛弃繁琐的 add_conditional_edges，让节点自己决定去向。
  1. from langgraph.types import Command
  3. def smart_agent_node(state: AgentState):
  4.     messages = state['messages']
  5.     last_message = messages[-1]
  6.    
  7.     if "查天气" in last_message:
  8.         # Command 方式：同时更新状态并指定去向
  9.         return Command(
  10.             goto="tool",
  11.             update={"messages": ["准备调用天气工具..."]}
  12.         )
  13.     else:
  14.         return Command(
  15.             goto=END,
  16.             update={"messages": ["直接回答问题，结束流程。"]}
  17.         )
  19. command_workflow = StateGraph(AgentState)
  20. command_workflow.add_node("agent", smart_agent_node)
  21. command_workflow.add_node("tool", execute_tool)
  22. command_workflow.add_edge(START, "agent")
  23. # 节点自带 goto 逻辑，无需再写 add_conditional_edges
复制代码
进阶与生产环境迁移

当你掌握了基础的图结构后，LangGraph 在生产环境中的两大杀手锏是记忆持久化（Persistence）和人工干预（Human-in-the-loop）。
在编译图时，你可以传入一个 checkpointer（如基于 SQLite 或 Postgres 的保存器）。加入保存器后，图不仅能在单个循环中流转状态，还能将状态永久保存在数据库中。这意味着你可以实现线程级别的断点续传：
  1. # 进阶编译示例（引入持久化检查点）
  2. # from langgraph.checkpoint.sqlite import SqliteSaver
  3. # memory = SqliteSaver.from_conn_string(":memory:")
  4. # app = workflow.compile(checkpointer=memory, interrupt_before=["tool"])
复制代码
原文地址：https://blog.csdn.net/m0_66871478/article/details/161428733



欢迎光临 AI创想 (http://llms-ai.com/) Powered by Discuz! X3.4