LangGraph 那点事

米落枫 · 发表于 5 小时前

作者：爱琴孩
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）：代表一个独立单元，可以是：

边（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: str
query_type: str
answer: 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有望在更多领域发挥重要作用。

原文地址：https://blog.csdn.net/qq_28165595/article/details/150703341

25位大厂高管转战生成式AI创业！吸金猛，不

LangGraph 那点事

LangGraph(一)——构建聊天机器人

关于我们

产品与服务

解决方案

产品与服务