LangGraph 那点事

1. 概述

1.1 什么是LangGraph

LangGraph是由LangChain公司开发的一个库，用于构建基于大型语言模型(LLM)的有状态多主体应用程序，特别适用于创建智能体(Agent)和多智能体工作流。LangGraph能够让开发者构建具有循环功能的计算图，这在传统的链式结构(Chain)或有向无环图(DAG)中是无法实现的。通过LangGraph，开发者可以创建更加灵活、动态的智能体系统，能够处理更复杂的任务流程。

1.2 核心概念

1.2.1 状态图(StateGraph)

LangGraph的核心是StateGraph类，用于表示整个图结构。初始化时需要定义一个状态对象，这个状态对象会随着图的执行而更新。

1.2.2 节点(Nodes)

节点是图中的处理单元，可以是函数或LangChain的可运行组件。节点接收状态对象作为输入，并返回要更新的状态属性。

1.2.3 边(Edges)

边定义了节点之间的执行流程和数据流向。有两种类型的边：

• 普通边：固定的流向

• 条件边：基于节点输出的条件决定下一步去向

1.2.4 状态(State)

状态是在图执行过程中维护的中央数据对象。状态的属性可以通过两种方式更新：

• 完全覆盖：节点返回一个新值来替换原属性

• 添加：节点返回值会被添加到现有属性(适用于列表等)

1.2.5 循环(Cycles)

LangGraph的一个重要特性是能够创建循环流程，允许LLM在循环中进行推理决策，这对于构建智能体系统至关重要。

1.3 为什么需要LangGraph？

随着企业逐渐向更智能、数据驱动的组织发展，对于不仅仅能回答简单问题的系统需求正在迅速增加。企业不再满足于只能响应提示的AI系统，他们需要能够思考、计划和行动的系统。这些下一代系统必须能够：

• 在多步骤过程中进行协调
• 选择最合适的技术或数据源
• 检索和推理上下文信息
• 在无需持续人工干预的情况下自主执行决策

2. 技术架构

2.1 工作流类型

• 顺序工作流
• 条件分支工作流
• 循环工作流
• 并行工作流

2.2 LangGraph 与 LangChain 对比

特性	LangChain（Chain）	LangGraph（Graph）
控制流程	线性	可分支、循环、并行
复杂度管理	较难	模块化、清晰可控
多 Agent 协作	较难实现	易于设计多个节点交互
适合任务类型	单步/固定任务	多步骤、有状态流程
智能体开发	不适合 ❌	适合 ✅

2.3 图结构的三大要素

LangGraph的核心构建在三个关键要素之上：

节点（Node）：代表一个独立单元，可以是：

• Agent节点：封装独立Agent能力
• Tool节点：调用具体工具
• END节点：流程结束的标识

边（Edge）：标注状态流转的决策路径，决定下一步跳转到哪个节点：

• 顺序执行（线性流程）
• 条件跳转（基于状态的动态路由）

状态（State）：贯穿整个流程，记录数据或交互状态，驱动节点间的流转

这种设计使得LangGraph高度模块化和直观，特别适合需要遵循程序逻辑和条件分支的企业系统。

3.LangGraph的状态管理与工作流构建

状态管理是LangGraph的核心特性之一，它通过Pydantic模型来定义和维护整个工作流的共享状态。

3.1状态定义与管理

LangGraph使用TypedDict或Pydantic来定义状态结构，每个节点都可以访问和修改这个共享状态：

class AgentState(TypedDict):messages: List[BaseMessage]  # 对话历史agent_outcome: str    # 下一步决策      tool_response: str    # 工具调用结果   

   

3.2节点的实现

节点是执行具体功能的处理单元，它们接收当前状态并返回更新后的状态：

def agent_decision_node(state: AgentState) -> dict:# 分析用户意图并做出决策last_message = state["messages"][-1].content# 基于意图返回下一步行动return {"agent_outcome": decision}

3.3工作流的组装

通过StateGraph类，我们可以将节点和边组装成完整的工作流：

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_decision_node)
workflow.add_edge("agent", "tool")
workflow.set_entry_point("agent")

4. LangGraph的高级特性与最佳实践

LangGraph提供了许多高级特性，使其能够处理更复杂的场景。

4.1条件路由与动态决策

LangGraph支持条件边，可以根据状态动态决定执行路径：

def router(state):if state["intent"] == "technical":return "expert_agent"else:return "review_agent"
​
workflow.add_conditional_edges("intent_agent", router)

4.2 循环控制与迭代优化

与传统的DAG（有向无环图）不同，LangGraph支持循环结构，这使得实现多轮决策和重试机制成为可能：

• 验证-修正循环：不断改进输出直到满足质量标准

• 自我反思模式：通过反馈循环持续优化结果

• 树搜索决策：支持深度优先或广度优先的搜索策略

4.3可视化与调试

LangGraph提供了强大的可视化功能，可以直观地展示工作流结构：

# 生成Mermaid图表
mermaid_code = graph.get_graph().draw_mermaid()
# 保存为PNG图片
app.get_graph().draw_mermaid_png(output_file_path="workflow_graph.png")

6. demo代码

import os
from typing import TypedDict
​
from langchain_community.chat_models import ChatOpenAI
from langgraph.constants import START
from langgraph.graph import StateGraph, END
​
# 设置OpenAI API密钥
os.environ["OPENAI_API_KEY"] = "mykey"
​
# 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo")
​
​
# 定义状态类型
class State(TypedDict):user_input: strquery_type: stranswer: str
​
​
# 定义节点函数
def classify_query(state: State) -> dict:"""分类节点：判断用户输入的问题类型"""prompt = f"""你是一个智能客服助手，需要判断用户的问题属于以下哪种类型：1. 产品咨询2. 技术支持3. 售后服务4. 其他问题用户输入：{state['user_input']}请只返回类型名称，不要其他内容。"""res = llm.invoke(prompt)
​if "产品咨询" in res.content:return {"query_type": "product"}elif "技术支持" in res.content:return {"query_type": "technical"}elif "售后服务" in res.content:return {"query_type": "service"}else:return {"query_type": "other"}
​
​
def handle_product_query(state: State) -> dict:"""处理产品咨询"""prompt = f"""你是一个专业的产品顾问，请回答用户关于产品的咨询。用户问题：{state['user_input']}"""res = llm.invoke(prompt)return {"answer": res.content}
​
​
def handle_technical_query(state: State) -> dict:"""处理技术支持"""prompt = f"""你是一个专业的技术支持工程师，请回答用户的技术问题。用户问题：{state['user_input']}"""res = llm.invoke(prompt)return {"answer": res.content}
​
​
def handle_service_query(state: State) -> dict:"""处理售后服务"""prompt = f"""你是一个专业的售后服务顾问，请回答用户关于售后服务的问题。用户问题：{state['user_input']}"""res = llm.invoke(prompt)return {"answer": res.content}
​
​
def handle_other_query(state: State) -> dict:"""处理其他问题"""return {"answer": "抱歉，您的问题超出了我的服务范围，请稍后联系人工客服。"}
​
​
# 创建图
graph = StateGraph(State)
​
# 添加节点
graph.add_node("classify", classify_query)
graph.add_node("product", handle_product_query)
graph.add_node("technical", handle_technical_query)
graph.add_node("service", handle_service_query)
graph.add_node("other", handle_other_query)
​
# 添加边
graph.add_edge(START, "classify")
​
​
# 定义路由函数
def router(state: State):if state["query_type"] == "product":return "product"elif state["query_type"] == "technical":return "technical"elif state["query_type"] == "service":return "service"else:return "other"
​
​
# 添加条件边
graph.add_conditional_edges("classify",router,{"product": "product","technical": "technical","service": "service","other": "other"}
)
​
# 添加结束边
graph.add_edge("product", END)
graph.add_edge("technical", END)
graph.add_edge("service", END)
graph.add_edge("other", END)
​
# 编译图
app = graph.compile()
​
if __name__ == "__main__":# 测试不同的问题test_questions = ["这个产品的保修期是多久？","我的电脑无法开机怎么办？","如何申请退货？","今天天气怎么样？"]
​for question in test_questions:print(f"\n用户问题: {question}")result = app.invoke({"user_input": question})print(f"回答: {result['answer']}")

6. 总结

LangGraph作为一个新兴的AI应用开发框架，为构建复杂的AI工作流提供了强大的支持。它的出现大大简化了AI应用的开发过程，特别适合需要复杂逻辑处理的场景。随着AI技术的不断发展，LangGraph有望在更多领域发挥重要作用。