简单的LangGraph示例：从基础到实践的全流程解析

2026年1月7日 互联网

简单的LangGraph示例：从基础到实践的全流程解析

LangGraph作为一种用于构建语言处理流程图的框架，近年来在自然语言处理（NLP）和复杂任务编排领域展现出独特优势。其通过节点（Node）和边（Edge）的组合，能够清晰地描述任务执行的逻辑路径，尤其适合需要动态调整流程的场景。本文将以一个简单的问答系统为例，从环境搭建到完整代码实现，逐步解析LangGraph的核心用法。

一、LangGraph的核心概念与适用场景

1.1 为什么选择LangGraph？

传统NLP任务通常依赖线性流程或条件分支，但在处理多轮对话、动态任务切换或复杂逻辑时，这种设计容易变得冗余且难以维护。LangGraph通过图结构将任务分解为可复用的节点，并通过边定义节点间的依赖关系，实现了流程的模块化和动态化。例如：

  • 多轮对话管理：根据用户输入动态跳转到不同处理节点；
  • 任务链编排：组合多个工具（如检索、计算、生成）完成复杂任务；
  • 异常处理：通过特定边跳转到错误恢复节点。

1.2 核心组件解析

  • 节点（Node）：代表一个独立的处理单元（如文本分类、信息抽取）；
  • 边（Edge）：定义节点间的执行顺序或条件跳转；
  • 状态（State）：存储流程执行过程中的中间数据；
  • 控制器（Controller）：管理流程的启动、暂停和恢复。

二、环境准备与基础依赖

2.1 安装必要库

  1. pip install langgraph langchain-core
  • langgraph：核心流程图框架；
  • langchain-core：提供与LLM（大语言模型）交互的通用接口。

2.2 示例场景设计

假设需要构建一个简单的问答系统，包含以下节点：

  1. 输入解析：检查用户问题类型（如事实型、观点型）；
  2. 知识检索：从数据库中获取相关信息；
  3. 答案生成：结合检索结果生成回答；
  4. 异常处理：当检索失败时返回友好提示。

三、完整代码实现与分步解析

3.1 定义节点类

每个节点需实现call方法，接收state参数并返回更新后的状态。

  1. from langgraph.predefined import State
  2. from langchain_core.prompts import PromptTemplate
  3. from langchain_core.outputs import ChatGeneration
  5. class InputParser:
  6.     def __call__(self, state: State) -> State:
  7.         question = state["input"]
  8.         if "是什么" in question:
  9.             state["question_type"] = "fact"
  10.         else:
  11.             state["question_type"] = "opinion"
  12.         return state
  14. class KnowledgeRetriever:
  15.     def __init__(self, knowledge_base):
  16.         self.knowledge_base = knowledge_base
  18.     def __call__(self, state: State) -> State:
  19.         question_type = state["question_type"]
  20.         if question_type == "fact":
  21.             # 模拟检索逻辑
  22.             state["retrieved_info"] = f"关于{question_type}问题的答案..."
  23.         else:
  24.             raise ValueError("无法处理该类型问题")
  25.         return state
  27. class AnswerGenerator:
  28.     def __call__(self, state: State) -> State:
  29.         info = state["retrieved_info"]
  30.         prompt = PromptTemplate(
  31.             input_variables=["info"],
  32.             template="根据以下信息生成回答：{info}"
  33.         )
  34.         # 模拟LLM生成
  35.         state["answer"] = ChatGeneration(text=f"生成的回答：{info}").text
  36.         return state
  38. class ErrorHandler:
  39.     def __call__(self, state: State) -> State:
  40.         state["answer"] = "抱歉，我暂时无法回答这个问题。"
  41.         return state

3.2 构建流程图

使用@node装饰器注册节点，并通过add_edges定义边关系。

  1. from langgraph.graph import Graph
  3. # 初始化图
  4. graph = Graph()
  6. # 注册节点
  7. graph.add_node("parser", InputParser())
  8. graph.add_node("retriever", KnowledgeRetriever(knowledge_base={}))
  9. graph.add_node("generator", AnswerGenerator())
  10. graph.add_node("error_handler", ErrorHandler())
  12. # 定义边关系
  13. graph.add_edges(
  14.     "parser" --> "retriever" if "question_type" in State,
  15.     "retriever" --> "generator" if no_exception,
  16.     "retriever" --> "error_handler" if exception_occurs
  17. )

3.3 执行流程

初始化状态并运行图：

  1. state = State(input="人工智能是什么？")
  2. result = graph.run(state)
  3. print(result["answer"])  # 输出：生成的回答：关于fact问题的答案...

四、关键注意事项与优化建议

4.1 状态管理最佳实践

  • 避免全局状态：所有数据应通过state对象传递，确保流程隔离性；
  • 类型提示：为state字段添加类型注解，提升代码可读性；
  • 默认值初始化：在节点中处理缺失字段，避免KeyError。

4.2 边条件设计原则

  • 显式优于隐式：明确边触发条件（如if state["score"] > 0.8）；
  • 避免循环依赖：确保图结构无环（DAG），否则需手动处理循环；
  • 优先级控制：通过权重或顺序定义多边竞争时的执行优先级。

4.3 性能优化方向

  • 节点并行化：对无依赖的节点使用异步执行；
  • 缓存中间结果：对频繁使用的检索结果进行缓存；
  • 动态图调整：根据运行时数据动态添加/删除节点。

五、扩展场景与进阶用法

5.1 动态图生成

通过外部逻辑动态修改图结构：

  1. def build_dynamic_graph(user_role):
  2.     graph = Graph()
  3.     if user_role == "admin":
  4.         graph.add_node("admin_panel", AdminPanel())
  5.     # 添加其他节点...
  6.     return graph

5.2 与LLM集成

结合LangChain的LLM链：

  1. from langchain_community.llms import OpenAI
  3. class LLMNode:
  4.     def __init__(self, llm: OpenAI):
  5.         self.llm = llm
  7.     def __call__(self, state: State) -> State:
  8.         prompt = state["prompt"]
  9.         state["llm_output"] = self.llm(prompt)
  10.         return state

5.3 调试与可视化

使用langgraph.visualizer生成流程图：

  1. from langgraph.visualizer import visualize
  3. graph = build_dynamic_graph("user")
  4. visualize(graph).show()  # 生成交互式流程图

六、总结与未来展望

通过本文的简单示例，我们掌握了LangGraph的核心用法：从节点定义、边关系到状态管理，再到动态执行。其模块化设计使得复杂流程的开发和维护更加高效。未来，随着LangGraph与LLM、Agent技术的深度融合，其在自动化任务编排、多模态交互等领域的应用潜力将进一步释放。开发者可结合具体场景，探索更复杂的图结构与动态调整策略。

最新文章