AI创想

标题: LangGraph [打印本页]
作者: AI小编    时间: 2026-7-6 14:05
标题: LangGraph
作者：CSDN博客
目录


一、LangGraph 是什么？

LangGraph 是 LangChain 生态下的状态驱动图结构工作流框架，专为构建复杂、可恢复、长运行的AI智能体(Agent) 设计。简单说：
核心优势：
二、快速安装
  1. # 基础安装
  2. pip install langgraph
  4. # 完整安装（含LangChain核心、LLM集成、可视化等）
  5. pip install langgraph[standard]
复制代码
三、核心概念

概念通俗解释类比
Graph（图）整个工作流的容器，定义所有节点和边一张完整的流程图
Node（节点）工作流中的单个步骤（如调用LLM、工具、判断）流程图中的方框
Edge（边）节点间的连接关系，定义执行顺序流程图中的箭头
State（状态）工作流的全局数据存储，所有节点共享读写流水线的物料箱，每个工位都能取放东西
Conditional Edge（条件边）根据状态动态选择下一个节点流程图中的判断分支（是/否）
Checkpoint（检查点）State状态快照，支持暂停/恢复执行游戏存档，随时读档继续
四、基础语法API

1. 状态定义（最关键）

状态是LangGraph的灵魂，用TypedDict定义结构化数据：
  1. from typing import TypedDict, List
  3. classAgentState(TypedDict):# 必须包含的字段：消息历史
  4.     messages: List[dict]# 自定义字段：当前任务
  5.     task:str# 自定义字段：工具调用结果
  6.     tool_results:dict# 自定义字段：是否完成
  7.     is_finished:bool=False# 带默认值
复制代码
2. 节点函数（业务逻辑）

每个节点是一个函数，输入状态，输出状态更新：
  1. # 1. 简单节点：调用LLM思考defllm_think(state: AgentState)-> AgentState:# 调用LLM生成思考结果
  2.     llm_output ="我需要搜索天气信息来回答用户问题"# 更新状态：添加思考消息
  3.     state["messages"].append({"role":"assistant","content": llm_output})return state
  5. # 2. 工具调用节点defcall_weather_api(state: AgentState)-> AgentState:# 调用天气API
  6.     weather_data ={"temperature":25,"condition":"sunny"}# 更新状态：保存工具结果
  7.     state["tool_results"]["weather"]= weather_data
  8.     state["messages"].append({"role":"tool","name":"weather_api","content":str(weather_data)})return state
  10. # 3. 条件判断节点（用于条件边）defshould_continue(state: AgentState)->str:# 根据状态判断下一步：继续工具调用还是结束if state["is_finished"]:return"end"# 指向结束节点else:return"continue"# 指向下一个处理节点
复制代码
3. 构建图（核心步骤）

用Graph类组装节点和边：
  1. from langgraph.graph import Graph
  3. # 1. 创建图，指定状态类型
  4. graph = Graph(AgentState)# 2. 添加节点（name, function）
  5. graph.add_node("llm_think", llm_think)
  6. graph.add_node("call_weather", call_weather_api)
  7. graph.add_node("check_finish", should_continue)# 条件判断节点# 3. 添加边（定义执行顺序）# 固定边：从A到B
  8. graph.add_edge("llm_think","call_weather")# 条件边：根据判断结果选择下一个节点
  9. graph.add_conditional_edges("check_finish",# 源节点
  10.     should_continue,# 判断函数（返回目标节点名）{"continue":"llm_think",# 循环"end":"__end__"# 内置结束节点})# 4. 设置起始节点
  11. graph.set_entry_point("llm_think")# 5. 编译图（生成可运行的应用）
  12. app = graph.compile()
复制代码
4. 运行图
  1. # 初始状态
  2. initial_state ={"messages":[{"role":"user","content":"今天苏州天气怎么样？"}],"task":"查询苏州天气","tool_results":{}}# 运行图（两种方式）# 方式1：一次性运行到结束
  3. result = app.invoke(initial_state)# 方式2：流式运行（适合实时交互）for step in app.stream(initial_state):print("Step:", step)print("State:", step["messages"][-1])
复制代码
五、实战案例一：智能问答Agent

代码：
  1. from typing import TypedDict, List, Optional
  2. from langgraph.graph import Graph
  3. from langchain_openai import ChatOpenAI
  4. from langchain.tools import tool
  5. import os
  7. # 1. 环境准备（替换为你的API密钥）
  8. os.environ["OPENAI_API_KEY"]="your-api-key"
  9. llm = ChatOpenAI(model="gpt-3.5-turbo")# 2. 定义工具（示例：天气查询工具）@tooldefweather_api(city:str)->str:"""查询指定城市的天气"""# 模拟API调用returnf"{city}天气：25℃，晴，微风"
  11. tools =[weather_api]# 3. 定义状态classAgentState(TypedDict):
  12.     messages: List[dict]# 消息历史
  13.     tool_name: Optional[str]# 要调用的工具名
  14.     tool_input: Optional[str]# 工具输入
  15.     tool_result: Optional[str]# 工具结果
  16.     is_finished:bool=False# 是否完成# 4. 定义节点函数defllm_decision(state: AgentState)-> AgentState:"""LLM决策：判断是否需要调用工具"""
  17.     messages = state["messages"]# 调用LLM生成响应（含工具调用判断）
  18.     response = llm.invoke(messages +[{"role":"system","content":"你可以调用weather_api工具查询天气，格式：TOOL[工具名,参数]。不需要工具时直接回答。"}])
  19.    
  20.     content = response.content
  21.     state["messages"].append({"role":"assistant","content": content})# 检查是否需要调用工具if"TOOL["in content:# 解析工具调用（简化示例）
  22.         tool_call = content.split("TOOL[")[1].split("]")[0]
  23.         state["tool_name"], state["tool_input"]= tool_call.split(",")
  24.         state["is_finished"]=Falseelse:
  25.         state["is_finished"]=Truereturn state
  27. deftool_execution(state: AgentState)-> AgentState:"""工具执行节点"""
  28.     tool_name = state["tool_name"]
  29.     tool_input = state["tool_input"]# 执行工具for tool in tools:if tool.name == tool_name:
  30.             result = tool.run(tool_input)
  31.             state["tool_result"]= result
  32.             state["messages"].append({"role":"tool","name": tool_name,"content": result})breakreturn state
  34. defshould_continue(state: AgentState)->str:"""判断是否继续循环"""return"end"if state["is_finished"]else"continue"# 5. 构建图
  35. graph = Graph(AgentState)# 添加节点
  36. graph.add_node("llm_decision", llm_decision)
  37. graph.add_node("tool_execution", tool_execution)# 添加边
  38. graph.add_edge("llm_decision","tool_execution")
  39. graph.add_conditional_edges("tool_execution",
  40.     should_continue,{"continue":"llm_decision",# 循环：继续决策"end":"__end__"# 结束})# 设置起始节点
  41. graph.set_entry_point("llm_decision")# 编译图
  42. app = graph.compile()# 6. 运行测试if __name__ =="__main__":
  43.     initial_state ={"messages":[{"role":"user","content":"苏州今天天气如何？"}],"tool_name":None,"tool_input":None,"tool_result":None}# 流式运行并打印结果for step in app.stream(initial_state):for node, data in step.items():print(f"\n=== {node} ===")if"messages"in data:print("最新消息:", data["messages"][-1])
复制代码
运行流程解析：
六、实战案例二：多智能体协作

核心流程：研究员 → 程序员 → 审核员 → 【不通过→退回程序员 / 通过→总结员】
[code]from typing import TypedDict, Literal
from langgraph.graph import StateGraph

# ===================== 1. 定义全局状态（仅新增2个审核字段，其他不变）=====================classMultiAgentState(TypedDict):# 用户原始需求
    query:str# 研究员搜集的资料
    research_info:str# 程序员编写的代码
    code:str# 审核意见（新增）
    review_opinion:str# 代码是否通过审核（新增）
    is_code_approved:bool# 总结员的最终结果
    final_result:str# ===================== 2. 原有3个智能体（完全没改动！）=====================defresearcher_agent(state: MultiAgentState)-> MultiAgentState:"""研究员：搜集资料"""
    query = state["query"]
    research_info =f"【研究员资料】关于{query}：核心逻辑是输入处理+函数实现，适配Python语法"print("✅ 研究员完成工作")return{"research_info": research_info}defcoder_agent(state: MultiAgentState)-> MultiAgentState:"""程序员：根据资料/审核意见写代码"""# 如果有审核意见，打印提示（模拟修改代码）if state["review_opinion"]:print(f"



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