LangGraph 入门实战：用图结构解锁LLM的复杂工作流 - AI创想

# 安装核心依赖
pip install langgraph langchain openai python-dotenv
# 可选：若需使用其他工具（如搜索、数据库），安装对应依赖
# pip install langchain-community

复制代码

from dotenv import load_dotenv
from langgraph.graph import Graph, StateGraph, END
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Dict, Any
# 1. 加载环境变量（读取OpenAI API Key）
load_dotenv()
# 2. 定义全局状态（State）：存储所有中间结果
# 用Pydantic模型定义State，结构更清晰，支持类型校验
class QAState(BaseModel):
question: str = Field(description="用户的原始问题")
initial_answer: str = Field(default="", description="LLM的初步回答")
is_complete: bool = Field(default=False, description="回答是否完整的标记")
final_answer: str = Field(default="", description="最终完善后的回答")
# 3. 初始化LLM（这里使用OpenAI的gpt-3.5-turbo，可替换为其他模型）
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# 4. 定义各个节点（Node）的执行逻辑
# 节点1：接收用户问题（初始化状态）
def init_state(state: QAState) -> Dict[str, Any]:
"""初始化状态，将用户问题存入State"""
return {"question": state.question, "initial_answer": "", "is_complete": False}
# 节点2：LLM生成初步回答
def generate_initial_answer(state: QAState) -> Dict[str, Any]:
"""根据用户问题，生成初步回答"""
prompt = f"请回答用户的问题，简洁明了，无需展开：{state.question}"
response = llm.invoke(prompt)
return {"initial_answer": response.content}
# 节点3：校验回答完整性（条件判断节点）
def check_answer_complete(state: QAState) -> str:
"""校验初步回答是否完整，返回下一步节点的名称"""
# 简单校验逻辑：判断回答长度是否大于50字符，或是否包含关键信息（可自定义复杂逻辑）
prompt = f"""
请判断以下回答是否完整回答了用户的问题，只需返回"完整"或"不完整"：
用户问题：{state.question}
初步回答：{state.initial_answer}
"""
check_result = llm.invoke(prompt).content.strip()
if check_result == "完整":
state.is_complete = True
return "complete_node" # 流向“生成最终回答”节点
else:
state.is_complete = False
return "improve_node" # 流向“完善回答”节点
# 节点4：完善回答（循环节点）
def improve_answer(state: QAState) -> Dict[str, Any]:
"""如果回答不完整，基于初步回答进行完善"""
prompt = f"""
你的初步回答不够完整，请结合用户问题，补充关键信息，完善回答：
用户问题：{state.question}
初步回答：{state.initial_answer}
完善要求：逻辑清晰，信息全面，不冗余
"""
improved_response = llm.invoke(prompt)
return {"initial_answer": improved_response.content} # 覆盖初步回答，用于再次校验
# 节点5：生成最终回答
def generate_final_answer(state: QAState) -> Dict[str, Any]:
"""将完整的回答整理为最终格式，存入State"""
final_prompt = f"""
请将以下回答整理为最终版本，语气友好，格式规范：
用户问题：{state.question}
完整回答：{state.initial_answer}
"""
final_response = llm.invoke(final_prompt)
return {"final_answer": final_response.content}
# 5. 构建LangGraph（StateGraph）
# 初始化图，指定State类型
graph_builder = StateGraph(QAState)
# 向图中添加所有节点（参数：节点名称，节点执行函数）
graph_builder.add_node("init", init_state)
graph_builder.add_node("generate_initial", generate_initial_answer)
graph_builder.add_node("check_complete", check_answer_complete)
graph_builder.add_node("improve", improve_answer)
graph_builder.add_node("complete", generate_final_answer)
# 6. 定义节点之间的边（流向）
# 确定性边：从init节点 → generate_initial节点（固定流向）
graph_builder.add_edge("init", "generate_initial")
# 确定性边：从generate_initial节点 → check_complete节点（固定流向）
graph_builder.add_edge("generate_initial", "check_complete")
# 条件边：从check_complete节点出发，根据返回值流向不同节点
graph_builder.add_conditional_edges(
source="check_complete", # 源节点
# 条件函数：接收源节点的输出（这里是check_answer_complete的返回值），返回目标节点名称
path_selector=lambda x: x,
# 映射关系：源节点输出 → 目标节点
path_map={
"improve_node": "improve", # 输出"improve_node" → 流向improve节点
"complete_node": "complete" # 输出"complete_node" → 流向complete节点
}
)
# 确定性边：improve节点 → check_complete节点（循环校验，直到回答完整）
graph_builder.add_edge("improve", "check_complete")
# 确定性边：complete节点 → 结束节点（END是LangGraph内置的结束标记）
graph_builder.add_edge("complete", END)
# 7. 设置图的入口节点（起始点）
graph_builder.set_entry_point("init")
# 8. 编译图（生成可运行的图实例）
graph = graph_builder.compile()
# 9. 运行图，传入初始状态（用户问题）
if __name__ == "__main__":
# 初始状态：仅传入用户问题
initial_state = QAState(question="请详细说明LangGraph和LangChain的区别与联系")
# 运行图（stream_mode="values"表示返回每一步的状态，便于调试）
for step, state in enumerate(graph.stream(initial_state, stream_mode="values")):
print(f"\n=== 步骤 {step + 1} ===")
print(f"当前节点：{graph.get_current_node()}")
print(f"用户问题：{state.question}")
print(f"初步回答：{state.initial_answer}")
print(f"回答是否完整：{state.is_complete}")
print(f"最终回答：{state.final_answer}")
# 输出最终结果
final_result = graph.invoke(initial_state)
print("\n=== 最终结果 ===")
print(f"用户问题：{final_result.question}")
print(f"最终回答：{final_result.final_answer}")

复制代码

from langgraph.graph import StateGraph
# 定义两个并行执行的节点（比如两个不同的智能体）
def agent1(state: QAState) -> Dict[str, Any]:
"""智能体1：补充理论相关内容"""
prompt = f"补充LangGraph的理论基础：{state.initial_answer}"
return {"initial_answer": state.initial_answer + "\n" + llm.invoke(prompt).content}
def agent2(state: QAState) -> Dict[str, Any]:
"""智能体2：补充实战相关内容"""
prompt = f"补充LangGraph的实战案例：{state.initial_answer}"
return {"initial_answer": state.initial_answer + "\n" + llm.invoke(prompt).content}
# 构建图时，添加并行边
graph_builder.add_parallel_edges(
source="generate_initial", # 源节点
destinations=["agent1", "agent2"], # 并行的目标节点
then="check_complete" # 并行节点执行完毕后，统一流向的节点
)

复制代码

# 1. 可视化图结构（生成PNG文件）
graph.draw("qa_graph.png") # 需要安装graphviz：pip install graphviz
# 2. 持久化图的运行状态（使用LangChain的持久化工具）
from langchain.storage import LocalFileStore
from langgraph.checkpoint.memory import MemorySaver
# 初始化内存持久化器（可替换为本地文件、数据库等）
memory = MemorySaver()
# 编译图时指定持久化器
graph = graph_builder.compile(checkpointer=memory)
# 运行图时，指定会话ID，可恢复会话状态
session_id = "user_123"
final_state = graph.invoke(initial_state, config={"configurable": {"session_id": session_id}})
# 恢复会话状态（比如用户中断后，继续之前的工作流）
restored_state = graph.get_state(config={"configurable": {"session_id": session_id}})

复制代码