LangGraph解析

随着大语言模型（LLM）技术的快速发展，AI应用的需求已经从简单的问答扩展到了复杂的多步骤任务处理、多智能体协作等场景。传统的线性链式开发方式在面对这些复杂需求时显得力不从心。LangGraph应运而生，它是一个用于构建有状态、多智能体AI应用的开源框架，采用图（Graph）结构来编排复杂工作流，为开发者提供了更强大和灵活的工具。

本文将基于LangGraph的官方文档和实践经验，从基础概念到核心机制，再到实际应用案例，全面深入地解析LangGraph的技术原理和应用方法，帮助开发者掌握这一强大的AI应用开发框架。

一、LangGraph基础概览

1.1 核心理念：图驱动的工作流

LangGraph将AI应用的逻辑抽象为一张由"节点"（Nodes）和"边"（Edges）构成的状态图。下图清晰展示了这种结构与执行方式：

在LangGraph的图结构中，每个组件都有明确的职责：

1.2 主要特点与优势

LangGraph相比传统的AI应用开发框架，具有以下显著特点：

1.3 生态系统与使用方式

LangGraph提供了灵活的集成路径，开发者可以根据需求选择不同的使用方式：

1.4 LangGraph与LangChain的区别

二、LangGraph图的关键概念

2.1 智能体（Agent）

智能体是LangGraph中最基础也是最重要的概念之一。

工作原理

2.2 工作流（Workflow）

2.2.1 并行化（Parallelization）

并行化是指让多个LLMs同时处理一个任务，并通过编程方式对它们的输出结果进行聚合。

两种主要变体：

并行化的适用场景：

2.2.2 路由（Router）

路由的工作是将输入进行分类，并将其导向到相应的后续任务。

2.2.3 协调者（Orchestrator）

在协调者-工作者模式中，一个协调者将任务分解并将每个子任务分配给工作者。

与并行化模式的区别：

2.2.4 评估者-优化器（Evaluator-optimizer）

三、深度理解LangGraph核心：Graph

3.1 Graph的基本组成

Graph是LangGraph的基本构建模块，它是一个有向无环图（DAG），用于描述任务之间的依赖关系。

3.2 State状态

3.2.1 State的定义方式

使用TypedDict定义：

from typing import TypedDict ​ class OverallState(TypedDict): foo: str user_input: str graph_output: str

使用Pydantic BaseModel定义：

from pydantic import BaseModel ​ class OverallState(BaseModel): a: str

3.2.2 State的更新机制

MessagesState便捷使用：

from langgraph.graph import MessagesState ​ # 直接使用，无需手动定义 state = MessagesState()

3.3 Node节点

3.3.1 Node的基本定义

def node_1(state: InputState) -> OverallState: return {"foo": state["user_input"] + "> 长沙市"}

3.3.2 Node的高级特性

缓存示例：

from langgraph.types import CachePolicy from langgraph.cache.memory import InMemoryCache ​ builder.add_node("node1", node_1, cache_policy=CachePolicy(ttl=5)) graph = builder.compile(cache=InMemoryCache()) ​ # 第一次调用：执行节点逻辑 graph.invoke({"number": 5}, config={"configurable": {"user_id": "123"}}) ​ # 第二次调用：从缓存获取 graph.invoke({"number": 5}, config={"configurable": {"user_id": "456"}})

3.4 Edge边

3.4.1 边的类型

3.4.2 普通边和EntryPoint

from langgraph.constants import START, END ​ builder.add_edge(START, "node_1") builder.add_edge("node_1", "node_2") builder.add_edge("node_2", END)

3.4.3 条件边

def routing_func(state: State) -> str: if state["number"] > 5: return "node1" else: return END ​ builder.add_conditional_edges(START, routing_func)

使用映射的方式：

def routing_func(state: State) -> bool: return state["number"] > 5 ​ builder.add_conditional_edges( START, routing_func, {True: "node_a", False: "node_b"} )

3.4.4 Send动态路由

from langgraph.types import Send ​ def routing_func(state: State): result = [] for message in state["messages"]: result.append(Send("node1", {"msg": message})) return result ​ builder.add_conditional_edges(START, routing_func)

3.5 子图的使用

创建子图：

# 子图构建 subgraph_builder = StateGraph(State) subgraph_builder.add_node("sub_node", sub_node_function) subgraph_builder.add_edge(START, "sub_node") subgraph_builder.add_edge("sub_node", END) subgraph = subgraph_builder.compile() ​ # 父图构建 builder = StateGraph(State) builder.add_node("subgraph_node", subgraph) builder.add_edge(START, "subgraph_node") builder.add_edge("subgraph_node", END) graph = builder.compile()

3.6 图的Stream支持

Custom Stream示例：

from langgraph.config import get_stream_writer ​ def node(state: State): writer = get_stream_writer() writer({"自定义key": "在节点内返回自定义信息"}) return {"answer": "some data"} ​ for chunk in graph.stream(inputs, stream_mode="custom"): print(chunk)

四、LangGraph实战：SQLAgent实现

4.1 SQL应用场景

4.2 langchain_community的使用

4.2.1 安装

pip install langchain-community pip install langchain-community[sql]

4.2.2 基本使用

from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit ​ db = SQLDatabase.from_uri("sqlite:///Chinook.db") toolkit = SQLDatabaseToolkit(db=db, llm=llm) tools = toolkit.get_tools()

4.2.3 主要功能模块

4.3 SQL智能体的创建

完整示例：

import asyncio from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit ​ llm = init_chat_model("deepseek:deepseek-chat") db = SQLDatabase.from_uri("sqlite:///Chinook.db") toolkit = SQLDatabaseToolkit(db=db, llm=llm) tools = toolkit.get_tools() ​ system_prompt = """ 你是SQL数据库智能体，根据问题生成正确的{dialect}查询语句。 限制查询结果最多{top_k}条，只查询相关列。 严禁执行INSERT、UPDATE、DELETE等操作。 开始时必须查看表结构。 """.format(dialect=db.dialect, top_k=10) ​ agent = create_agent(llm, tools, system_prompt=system_prompt) ​ async def main(): async for step in agent.astream( {"messages": [{"role": "user", "content": "哪种音乐类型的曲目平均时长最长？"}]}, stream_mode="values" ): step["messages"][-1].pretty_print() ​ asyncio.run(main())

4.4 SQL工具介绍

4.5 MCP Server Chart集成

集成示例：

import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient ​ mcp_client = MultiServerMCPClient({ "mcp-server-chart": { "command": "npx", "args": ["-y", "@antv/mcp-server-chart"], "transport": "stdio" } }) ​ mcp_tools = asyncio.run(mcp_client.get_tools()) agent = create_agent(llm, tools + mcp_tools, system_prompt=system_prompt)

MCP特点：

五、LangGraph本地服务器与UI界面

5.1 配置langgraph.json

{ "$schema": "https://langgra.ph/schema.json", "dependencies": ["."], "graphs": { "agent": "./main.py:agent", "sql_agent": "./example/pro2/sql_agent.py:agent" }, "env": ".env", "image_distro": "wolfi" }

5.2 启动服务器

langgraph dev

5.3 UI界面功能