LangGraph框架解析：构建复杂LLM应用的工作流引擎

1. LangGraph框架深度解析

LangGraph是由LangChain团队开发的开源框架，专为构建基于大型语言模型(LLM)的复杂应用而设计。作为一名长期从事AI应用开发的工程师，我发现LangGraph最吸引人的特点是它采用图结构(graph)来表示工作流，这为处理循环逻辑、状态管理和多主体协作等复杂场景提供了优雅的解决方案。

1.1 核心架构设计

LangGraph的核心架构基于有向图模型，这种设计选择并非偶然。在传统的工作流引擎中，线性流程(DAG)是主流，但对于需要反复迭代的AI应用来说，这种结构就显得力不从心。LangGraph的创新之处在于：

• 节点(Node)：代表具体的操作单元，可以是LLM调用、工具使用或自定义函数
• 边(Edge)：定义执行路径，支持条件分支(conditional edges)
• 状态(State)：贯穿整个图的共享数据结构，自动维护上下文

这种架构特别适合需要"思考-行动-观察"循环的Agent应用。比如，当构建一个研究助手时，Agent可能需要：1)分析问题 2)决定搜索策略 3)评估结果 4)根据结果决定继续搜索或生成最终答案——这正是典型的循环工作流。

1.2 与LangChain的关系

虽然同属一个生态，但LangGraph和LangChain有明确的分工：

实际项目中，我经常组合使用两者——用LangChain处理标准化组件(如提示模板、文档加载器)，用LangGraph编排复杂逻辑。

1.3 关键技术特性

持久化检查点(Checkpointing)
这是生产环境中不可或缺的功能。LangGraph会在每个"超级步骤"(super-step)自动保存状态到检查点，支持：

• 错误恢复：从最后成功步骤继续执行
• 人工介入：在关键决策点引入人工审核
• 时间旅行：回溯历史状态进行分析

多级记忆系统

• 短期记忆：会话线程内的上下文保持
• 长期记忆：跨会话的知识持久化
• 向量记忆：通过嵌入实现语义检索

2. 实战：构建ReAct Agent

ReAct(Reasoning+Acting)是当前最先进的Agent架构之一。下面我将展示如何用LangGraph实现一个完整的天气查询Agent。

2.1 环境准备

首先安装必要依赖：

bash复制pip install langgraph langchain-community dashscope

2.2 工具定义

Agent的能力很大程度上取决于其工具集。我们先定义一个模拟天气查询工具：

python复制from langchain_core.tools import Tool

def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 实际项目中这里会调用天气API
    return f"{city}当前天气：晴，25℃"

weather_tool = Tool(
    name="get_weather",
    func=get_weather,
    description="查询城市天气信息"
)

2.3 Agent组装

LangGraph提供了预构建的create_react_agent函数，大大简化了开发：

python复制from langgraph.prebuilt import create_react_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.checkpoint.memory import InMemorySaver

# 初始化大模型
model = ChatTongyi(model="qwen-max", temperature=0)

# 创建带记忆的Agent
agent = create_react_agent(
    model=model,
    tools=[weather_tool],
    checkpointer=InMemorySaver()  # 内存检查点
)

2.4 运行测试

python复制config = {"configurable": {"thread_id": "test_1"}}

# 第一轮查询
response = agent.invoke(
    {"messages": [{"role": "user", "content": "上海天气怎么样？"}]},
    config
)
print(response["messages"][-1].content)

# 第二轮查询（测试记忆）
response = agent.invoke(
    {"messages": [{"role": "user", "content": "那北京呢？"}]},
    config
)
print(response["messages"][-1].content)

关键细节说明：

1. configurable.thread_id是会话标识符，相同的thread_id会共享记忆
2. 消息格式必须遵循OpenAI的message协议
3. 检查点(Checkpointer)可以替换为Redis等持久化后端

3. 自定义工作流开发

预构建的Agent虽然方便，但实际业务往往需要定制化工作流。下面我们构建一个带搜索功能的问答系统。

3.1 搜索工具集成

使用Tavily搜索引擎（需先注册获取API key）：

python复制from langchain_tavily import TavilySearch
import os

# 配置搜索工具
tavily_search = TavilySearch(max_results=2)
tools = [tavily_search]

3.2 状态图定义

python复制from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

# 初始化图构建器
graph_builder = StateGraph(State)

# 添加LLM节点
def chatbot(state: State):
    return {"messages": [model.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)

# 添加工具节点
from langgraph.prebuilt import ToolNode
tool_node = ToolNode([tavily_search])
graph_builder.add_node("tools", tool_node)

3.3 条件路由

关键是要定义何时使用工具：

python复制def route_tools(state: State):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return END

graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", END: END}
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

3.4 持久化集成

python复制from langgraph.checkpoint.memory import MemorySaver

graph = graph_builder.compile(
    checkpointer=MemorySaver()  # 可替换为生产级存储
)

4. Agentic RAG实现

检索增强生成(RAG)是当前最热门的LLM应用模式之一。结合Agent能力后，系统可以主动决定何时检索、如何利用检索结果。

4.1 知识库构建

python复制from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import DashScopeEmbeddings

# 加载PDF文档
loader = PyMuPDFLoader("technical_doc.pdf")
pages = loader.load_and_split()

# 文档分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=200
)
texts = text_splitter.split_documents(pages[:10])  # 示例只用前10页

# 创建向量库
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
db = FAISS.from_documents(texts, embeddings)
retriever = db.as_retriever(search_kwargs={"k": 3})

4.2 检索节点实现

python复制from langchain_core.prompts import ChatPromptTemplate

template = """基于以下上下文回答问题：
{context}

问题：{question}
"""
prompt = ChatPromptTemplate.from_template(template)

def retrieval(state: State):
    question = state["messages"][-1].content
    docs = retriever.invoke(question)
    context = "\n".join(d.page_content for d in docs)
    return {"messages": [{"role": "system", "content": prompt.format(
        context=context, question=question
    )}]}

4.3 工作流编排

python复制graph_builder = StateGraph(State)

graph_builder.add_node("retrieval", retrieval)
graph_builder.add_node("llm", chatbot)

graph_builder.add_edge(START, "retrieval")
graph_builder.add_edge("retrieval", "llm")
graph_builder.add_edge("llm", END)

# 添加人工审核分支
def should_escalate(state: State) -> str:
    last_msg = state["messages"][-1].content
    if "不确定" in last_msg or "无法回答" in last_msg:
        return "human"
    return END

graph_builder.add_conditional_edges(
    "llm",
    should_escalate,
    {"human": "human", END: END}
)

def human_review(state: State):
    question = state["messages"][-2].content
    # 实际项目中这里会连接工单系统
    human_input = input(f"需要人工回答：{question}\n请输入回复：")
    return {"messages": [{"role": "assistant", "content": human_input}]}

graph_builder.add_node("human", human_review)
graph_builder.add_edge("human", END)

5. 生产环境实践建议

经过多个项目的实战，我总结出以下关键经验：

5.1 性能优化

1. 批量处理：对多个查询进行批量化处理

   python复制# 好于多次单独调用
   graph.batch([input1, input2])
   

2. 缓存策略：

   • 对LLM响应使用Redis缓存
   • 对检索结果建立本地缓存

3. 异步处理：

   python复制async for event in graph.astream(input):
       handle_event(event)
   

5.2 监控指标

必须监控的关键指标包括：

• 每个节点的执行耗时
• LLM调用的token消耗
• 工具调用的成功率
• 人工介入比例

5.3 常见陷阱

1. 状态污染：确保每个节点只修改它应该修改的状态部分
2. 循环失控：设置最大循环次数避免无限循环

   python复制graph_builder.compile(max_loops=10)
   

3. 工具选择冲突：当多个工具匹配时，明确设置优先级

6. 扩展应用场景

LangGraph的灵活性使其适用于多种复杂场景：

6.1 多Agent协作系统

python复制def researcher(state):
    # 负责信息收集
    pass

def analyst(state):
    # 负责数据分析
    pass

def writer(state):
    # 负责报告生成
    pass

graph_builder.add_node("researcher", researcher)
graph_builder.add_node("analyst", analyst)
graph_builder.add_node("writer", writer)

# 定义协作流程
graph_builder.add_edge(START, "researcher")
graph_builder.add_edge("researcher", "analyst")
graph_builder.add_edge("analyst", "writer")

6.2 动态工作流

根据运行时条件动态调整工作流：

python复制def dynamic_router(state):
    if state["task_type"] == "simple":
        return "fast_path"
    return "full_analysis"

graph_builder.add_conditional_edges(
    START,
    dynamic_router,
    {"fast_path": "quick_process", "full_analysis": "deep_analysis"}
)

6.3 混合主动系统

结合预定规则和LLM决策：

python复制def rule_engine(state):
    # 基于业务规则处理
    if meets_condition(state):
        return "rule_based_action"
    return "llm_decision"

def llm_decider(state):
    # 让LLM决定下一步
    pass

这些扩展模式展示了LangGraph如何支撑企业级AI应用的复杂需求。在实际项目中，我们通常需要根据具体业务场景调整和组合这些模式。