Руководство по началу работы с LangGraph

LangGraph — этонизкоуровневый фреймворк оркестрации агентов, разработанный командой LangChain, специально предназначенный для создания рабочих процессов ИИ с сохранением состояния и длительным выполнением.

В отличие от традиционных линейных цепочек вызовов LLM, LangGraph моделирует рабочие процессы какориентированный граф：

• Узел: Функции, выполняющие определённые действия (например, вызов LLM, выполнение инструментов, обработка данных)
• Ребро: Определяет пути потока между узлами, поддерживая условное ветвление
• Состояние: Данные, совместно используемые и передаваемые по всему рабочему процессу

Репозиторий с открытым исходным кодом:https://github.com/langchain-ai/langgraph。

Представьте, что вы дирижируете симфонией: традиционная цепочка LLM похожа на исполнение произведения от начала до конца, только последовательно; тогда как LangGraph похож на дирижёра, который может в любой момент изменить порядок исполнения в зависимости от реакции живой аудитории, повторить определённую часть или перейти к конкретному фрагменту. Это даёт AI-рабочим процессам мудрость «дирижёра» — способность к циклам, ветвлению и возврату назад, что действительно позволяет реализовать сложное автономное принятие решений.

Почему стоит выбрать LangGraph?

В таблице ниже сравниваются основные различия между традиционными цепочками LLM и LangGraph:

Применимые сценарии

• Чат-боты: Должны помнить контекст многоходового диалога
• Автономный агент: Может планировать, использовать инструменты и итеративно размышлять
• Мультиагентная система: Несколько ИИ совместно выполняют сложные задачи
• Рабочий процесс утверждения: Автоматизированный процесс, требующий проверки человеком
• Ассистент-исследователь: Требует многошаговых рассуждений и поиска информации

Основные концепции

Прежде чем начать писать код, сначала разберитесь с тремя ключевыми концепциями LangGraph.

Граф

Граф — это схема всего рабочего процесса, определяющая полную логическую структуру агента. Он состоит из узлов и рёбер:

StateGraph
   |-- Nodes（节点）
   |     |-- node_a
   |     |-- node_b
   |     +-- node_c
   +-- Edges（边）
         |-- START -> node_a
         |-- node_a -> node_b（条件边）
         |-- node_a -> node_c（条件边）
         +-- node_b -> END

State

Состояние — это то, что проходит через весь графобщая структура данных. Каждый узел может читать и обновлять состояние, а обновлённое состояние передаётся следующему узлу.

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class MyState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表（自动追加）
    user_name: str                            # 用户名称
    step_count: int                           # 步骤计数

Узлы

Узлы — это обычные функции Python, которые получают текущее State и возвращают обновлённое State (частичные поля).

def my_node(state: MyState) -> dict:
    # 读取状态
    messages = state["messages"]
    
    # 执行操作...
    result = "处理结果"
    
    # 返回更新的字段（不需要返回所有字段）
    return {"messages": [{"role": "ai", "content": result}]}

Рёбра

Рёбра определяют поток между узлами:

• Обычное ребро: фиксированный путь,node_a -> node_b
• Условное ребро: динамически определяет маршрут на основе State,node_a -> node_b или node_c
• Начальное ребро：START -> 第一个节点
• Конечное ребро：某节点 -> END

Настройка окружения

Установка зависимостей

Установите LangGraph и связанные зависимости, используя отечественное зеркало:

# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate      # macOS/Linux
# venv\Scripts\activate       # Windows

# 安装 LangGraph 和 LangChain
pip install langgraph langchain langchain-openai python-dotenv -i https://mirrors.aliyun.com/pypi/simple/

# 可选：安装开发工具
pip install langgraph-cli jupyter -i https://mirrors.aliyun.com/pypi/simple/

Настройка API-ключа

В корневом каталоге проекта создайте.envфайл и настройте OpenAI и DeepSeek:

# .env 文件内容

# OpenAI 配置（国外用户）
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://api.openai.com/v1

# DeepSeek 配置（国内用户推荐）
DEEPSEEK_API_KEY=sk-xxx
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-pro

Проверка установки

import langgraph
print(f"LangGraph 版本: {langgraph.__version__}")

Ваша первая программа на LangGraph

Начнём с самого простого примера — линейного рабочего процесса всего с двумя узлами.

Результат выполнения:

[greet_node] 收到消息: 世界
[process_node] 处理消息: 你好！世界

最终结果: {'message': '你好！世界', 'processed': True}

Визуализация структуры графа

В Jupyter Notebook можно напрямую визуализировать структуру графа:

# 在 Jupyter Notebook 中可视化
from IPython.display import Image
Image(graph.get_graph().draw_mermaid_png())

# 或者打印 Mermaid 格式
print(graph.get_graph().draw_mermaid())

Управление состоянием

Определение состояния с помощью TypedDict

from typing import TypedDict, Annotated, Optional
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 消息历史（add_messages reducer 自动追加而非覆盖）
    messages: Annotated[list, add_messages]
    
    # 普通字段（直接覆盖）
    user_id: str
    session_id: str
    
    # 可选字段
    error: Optional[str]
    
    # 计数器（使用 operator.add 作为 reducer）
    retry_count: Annotated[int, lambda x, y: x + y]

Определение состояния с помощью Pydantic (рекомендуется для продакшена)

from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class ProductionState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    user_id: str = ""
    confidence_score: float = 0.0
    
    class Config:
        arbitrary_types_allowed = True

MessagesState (встроенное удобное состояние)

LangGraph предоставляет встроенноеMessagesState, разработанное специально для сценариев диалога:

from langgraph.graph import MessagesState

# MessagesState 等价于:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], add_messages]

# 直接使用，无需自定义
builder = StateGraph(MessagesState)

Узлы

Обычные функциональные узлы

def simple_node(state: AgentState) -> dict:
    # 读取状态
    last_message = state["messages"][-1]
    
    # 执行操作
    response = f"收到: {last_message.content}"
    
    # 返回部分状态更新
    return {
        "messages": [{"role": "assistant", "content": response}]
    }

Узлы вызовов LLM

Асинхронные узлы

import asyncio

async def async_node(state: AgentState) -> dict:
    """异步节点，适合 I/O 密集型操作"""
    # 模拟异步操作（如 API 调用、数据库查询）
    await asyncio.sleep(0.1)
    
    result = await some_async_api_call(state["messages"][-1].content)
    return {"messages": [{"role": "assistant", "content": result}]}

# 使用异步图
result = await graph.ainvoke({"messages": [...]})

Использование классов как узлов

class RouterNode:
    def __init__(self, llm, system_prompt: str):
        self.llm = llm
        self.system_prompt = system_prompt
    
    def __call__(self, state: AgentState) -> dict:
        """类实例可以作为节点使用"""
        messages = [
            SystemMessage(content=self.system_prompt),
            *state["messages"]
        ]
        response = self.llm.invoke(messages)
        return {"messages": [response]}

# 添加类节点
router = RouterNode(llm, "你是一个专业的路由助手。")
builder.add_node("router", router)

Рёбра и условная маршрутизация

Обычные рёбра

# 固定路径：node_a 完成后始终执行 node_b
builder.add_edge("node_a", "node_b")

# 结束：node_a 完成后图结束
builder.add_edge("node_a", END)

Условные рёбра

Условные рёбра — ключевая возможность LangGraph, динамически определяющая следующий шаг на основе текущего состояния.

def route_after_llm(state: AgentState) -> str:
    """
    路由函数：根据 LLM 的最新输出决定走哪条路径
    返回值必须是已注册节点名称或 END
    """
    last_message = state["messages"][-1]
    
    # 如果 LLM 请求使用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 否则结束
    return END

# 添加条件边
builder.add_conditional_edges(
    "llm",              # 源节点
    route_after_llm,    # 路由函数
    {
        "tools": "tool_executor",   # 返回 "tools" 时 -> tool_executor 节点
        END: END                    # 返回 END 时 -> 结束
    }
)

Параллельное выполнение (Fan-out)

# 从一个节点并行分叉到多个节点
builder.add_edge("start_node", "branch_a")
builder.add_edge("start_node", "branch_b")
builder.add_edge("start_node", "branch_c")

# 多个节点汇聚到一个节点（Fan-in）
builder.add_edge("branch_a", "merge_node")
builder.add_edge("branch_b", "merge_node")
builder.add_edge("branch_c", "merge_node")

Полный пример условной маршрутизации

Пример результата выполнения:

User: 北京今天天气怎么样？
Assistant: 很抱歉，我没有实时天气数据。不过北京现在是春季，建议您出门时关注天气预报...
--------------------------------------------------

User: 帮我写一个 Python 快速排序
Assistant: 好的，下面是一个 Python 实现的快速排序算法：

```python
def quicksort(arr):
    if len(arr) <= 1:
        return arr
    ...
--------------------------------------------------

User: 你好，介绍一下你自己
Assistant: 你好！我是一个 AI 助手，很高兴为你服务。我可以帮你回答各种问题...
--------------------------------------------------

User: 再见啦！
Assistant: 再见！期待下次与你交流。
--------------------------------------------------

Создание чат-бота

Теперь используйте то, что вы узнали, чтобы построить робота, поддерживающего многоходовые диалоги. Этот робот может запоминать контекст разговора и обеспечивать непрерывный интерактивный опыт.

Пример результата выполнения:

You: 你好，我叫小明
Assistant: 你好，小明！很高兴认识你。有什么我可以帮助你的吗？

You: 我想学习 Python 编程
Assistant: 太好了！Python 是一门很适合初学者的编程语言。我可以帮你从基础开始：
1. 首先了解变量和数据类型
2. 学习条件语句和循环
3. 掌握函数的定义和使用
你想从哪个部分开始呢？

You: 你还记得我叫什么吗？
Assistant: 当然记得，你叫小明！你刚才说想学习 Python 编程，我们可以继续这个话题。

You: 退出
再见！

Вызов инструментов - ReAct агент

Предоставление агенту возможности использовать внешние инструменты — одна из самых мощных функций LangGraph.ReAct(Reason + Act) — самый распространённый паттерн агента: LLM думает -> выбирает инструмент -> выполняет инструмент -> наблюдает результат -> продолжает думать.

Определение инструментов

Сначала определите инструменты, которые может использовать Агент:

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息。
    
    Args:
        query: 搜索关键词
    
    Returns:
        搜索结果摘要
    """
    # 实际项目中替换为真实搜索 API
    return f"关于 '{query}' 的搜索结果：这是模拟的搜索结果..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。
    
    Args:
        expression: 数学表达式，如 '2 + 2' или  '100 * 0.8'
    
    Returns:
        计算结果
    """
    import ast
    import operator
    
    # 安全的运算符映射
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }
    
    def safe_eval(node):
        if isinstance(node, ast.Expression):
            return safe_eval(node.body)
        elif isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = safe_eval(node.left)
            right = safe_eval(node.right)
            return ops[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = safe_eval(node.operand)
            return ops[type(node.op)](operand)
        else:
            raise ValueError(f"不支持的表达式类型: {type(node)}")
    
    try:
        tree = ast.parse(expression, mode='eval')
        result = safe_eval(tree)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。
    
    Args:
        city: 城市名称
    
    Returns:
        天气信息
    """
    # 实际项目中替换为真实天气 API
    return f"{city} 今日天气：晴，温度 22C，湿度 60%"

tools = [search_web, calculate, get_weather]

Создайте ReAct-агента

Потоковый процесс вызова инструментов

# 流式观察 Agent 的每一步
for chunk in graph.stream(
    {"messages": [HumanMessage(content="搜索 LangGraph 的最新特性")]},
    stream_mode="updates"
):
    for node_name, updates in chunk.items():
        print(f"\n=== 节点: {node_name} ===")
        for msg in updates.get("messages", []):
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                for tc in msg.tool_calls:
                    print(f"  -> 调用工具: {tc['name']}({tc['args']})")
            else:
                print(f"  -> 输出: {msg.content[:200] if msg.content else '(无内容)'}")

Совместная работа с участием человека (Human-in-the-Loop)

LangGraph нативно поддерживает приостановку во время выполнения рабочего процесса, чтобы дождаться проверки или ввода человеком. Это очень полезно для чувствительных операций, требующих подтверждения человеком.

Используйте interrupt для приостановки выполнения

from langgraph.types import interrupt
from langgraph.checkpoint.memory import MemorySaver

def sensitive_action_node(state: MessagesState) -> dict:
    """执行敏感操作前请求人工审批"""
    last_msg = state["messages"][-1].content
    
    # 暂停图的执行，等待人工决策
    human_decision = interrupt({
        "question": "是否批准执行以下操作？",
        "action": last_msg,
        "risk_level": "中等"
    })
    
    if human_decision == "approve":
        return {"messages": [{"role": "assistant", "content": "操作已批准并执行完毕。"}]}
    else:
        return {"messages": [{"role": "assistant", "content": "操作已取消。"}]}

# 必须使用 checkpointer 才能支持 interrupt
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Полный рабочий процесс утверждения

Установка точек останова на рёбрах

# 另一种方式：在编译时指定断点
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_node"],   # 执行该节点前暂停
    # interrupt_after=["review_node"],     # 执行该节点后暂停
)

Персистентная память

LangGraph предоставляет встроенный механизм сохранения состояния, который позволяет агентам запоминать историю диалога между сессиями.

Хранилище в памяти (подходит для разработки и тестирования)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 使用 thread_id 区分不同会话
config_user_a = {"configurable": {"thread_id": "user-alice"}}
config_user_b = {"configurable": {"thread_id": "user-bob"}}

# Alice 的对话
graph.invoke({"messages": [HumanMessage(content="我叫 Alice")]}, config=config_user_a)
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config_user_a)
# Agent 能记住：你叫 Alice

# Bob 的对话完全独立
graph.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config_user_b)
# Agent 不知道 Bob 的名字（不同 thread_id）

Постоянное хранилище SQLite (подходит для локальных проектов)

# 安装依赖
# pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

# 数据持久化到文件，程序重启后对话历史仍存在
with SqliteSaver.from_conn_string("./chat_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "persistent-chat"}}
    
    # 第一次运行
    graph.invoke({"messages": [HumanMessage(content="我叫张三")]}, config=config)
    
    # 程序重启后再次运行，记忆仍然存在
    result = graph.invoke(
        {"messages": [HumanMessage(content="你还记得我叫什么吗？")]},
        config=config
    )

Просмотр истории диалога

# 获取某个 thread 的完整状态历史
history = list(graph.get_state_history(config))

for snapshot in history:
    print(f"时间: {snapshot.created_at}")
    print(f"消息数: {len(snapshot.values['messages'])}")
    print("---")

# 获取当前状态
current_state = graph.get_state(config)
print(f"当前消息数: {len(current_state.values['messages'])}")

Многоагентные системы

LangGraph отлично справляется с координацией совместной работы нескольких специализированных агентов. Разбивая сложные задачи и распределяя их между разными экспертными агентами, можно добиться более мощных возможностей решения задач.

Паттерн супервизора

Подграф

Инкапсулируйте сложные подпроцессы в виде подграфов и переиспользуйте их в основном графе:

# 将复杂子流程封装为子图，在主图中复用
sub_builder = StateGraph(MessagesState)
sub_builder.add_node("step1", step1_node)
sub_builder.add_node("step2", step2_node)
sub_builder.add_edge(START, "step1")
sub_builder.add_edge("step1", "step2")
sub_builder.add_edge("step2", END)
sub_graph = sub_builder.compile()

# 在主图中使用子图
main_builder = StateGraph(MessagesState)
main_builder.add_node("preprocessing", preprocess_node)
main_builder.add_node("sub_workflow", sub_graph)  # 直接使用编译好的子图
main_builder.add_node("postprocessing", postprocess_node)

main_builder.add_edge(START, "preprocessing")
main_builder.add_edge("preprocessing", "sub_workflow")
main_builder.add_edge("sub_workflow", "postprocessing")
main_builder.add_edge("postprocessing", END)

main_graph = main_builder.compile()

Визуальная отладка в LangGraph Studio

LangGraph Studio — это официальная среда визуальной разработки, которая позволяет просматривать процесс выполнения агента в реальном времени, значительно повышая эффективность разработки и отладки.

Установите LangGraph CLI

pip install langgraph-cli -i https://mirrors.aliyun.com/pypi/simple/

Создайте файл конфигурации проекта

Создайте в корневом каталоге проектаlanggraph.jsonФайл конфигурации:

{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./my_agent.py:graph"
  },
  "env": ".env"
}

Запустите сервер разработки

langgraph dev

После запуска посетитеhttp://localhost:8123Затем вы можете использовать LangGraph Studio в браузере.

Основные возможности Studio

• Визуализация в реальном времени: Графически отображает процесс выполнения узлов, позволяя интуитивно понимать состояние рабочего процесса
• Проверка состояния: Приостановите выполнение на любом узле, чтобы просмотреть текущее состояние, что упрощает диагностику проблем
• Путешествие во времени: Воспроизведите исторические шаги выполнения, чтобы отслеживать изменения состояния на каждом шаге
• Горячая перезагрузка: Автоматически обновляет структуру графа после изменений кода, перезапуск сервиса не требуется

Лучшие практики и часто задаваемые вопросы

Лучшие практики

Ключевые моменты проектирования State

• Делайте State компактным, содержащим только необходимые поля
• Определяйте явные редьюсеры для сложных полей (например,add_messages）
• Используйте модели Pydantic для проверки типов state в производственных средах
• Избегайте хранения слишком больших объектов в State; рассмотрите использование внешнего хранилища

Ключевые моменты проектирования узлов

• Каждый узел имеет одну ответственность, что упрощает тестирование и повторное использование
• Функции узлов должны быть идемпотентными (одинаковый вход даёт одинаковый выход)
• Избегайте прямого изменения переданного состояния в узлах; вместо этого возвращайте новые значения
• Разумно используйте async-узлы для операций, интенсивно использующих ввод-вывод

Обработка ошибок

def robust_node(state: AgentState) -> dict:
    try:
        result = risky_operation(state)
        return {"messages": [result], "error": None}
    except Exception as e:
        return {
            "error": str(e),
            "messages": [{"role": "assistant", "content": f"操作失败: {e}"}]
        }

Избегайте бесконечных циклов

def route_with_limit(state: AgentState) -> str:
    # 设置最大重试次数，防止无限循环
    if state.get("retry_count", 0) >= 3:
        return END
    
    if needs_retry(state):
        return "retry_node"
    return END

Часто задаваемые вопросы (FAQ)

Вопрос 1: Что делать, если формат возвращаемого значения узла неверен?

# 错误：直接修改 state 对象
def bad_node(state):
    state["messages"].append(...)  # 不要直接修改
    return state

# 正确：返回需要更新的字段
def good_node(state):
    return {"messages": [new_message]}  # 只返回变更字段

Вопрос 2: Как передавать временные данные между узлами?

Добавьте временные данные в определение State или используйте соглашение о префиксе с подчёркиванием для внутренних полей:

from typing import TypedDict

class PublicState(TypedDict):
    messages: list  # 对外暴露

class PrivateState(TypedDict):
    messages: list
    _internal_cache: dict  # 以下划线开头约定为内部使用

Вопрос 3: Как отлаживать процесс выполнения узла?

# 使用 stream 模式观察每个节点的输出
for event in graph.stream(initial_state, stream_mode="updates"):
    for node_name, state_update in event.items():
        print(f"\n[节点: {node_name}]")
        print(f"更新: {state_update}")

Вопрос 4: В чём разница между StateGraph и MessageGraph?

MessageGraphЭто более ранняя версия API с относительно ограниченной функциональностью. Теперь рекомендуется единообразно использоватьStateGraph, который более гибкий и многофункциональный. Для обработки сообщений используйтеStateGraph(MessagesState)или настройте свой, который содержитmessagesполя в State.

Итог

В этой статье систематически представлены основные концепции и практические применения фреймворка LangGraph. Благодаря оркестрации рабочих процессов на основе графов, LangGraph позволяет разработчикам создавать сложных AI-агентов с такими возможностями, как циклы, ветвление и сохранение состояния. По сравнению с традиционными линейными LLM Chains, LangGraph имеет значительные преимущества в сценариях, требующих многошаговых рассуждений, взаимодействия человека и машины и координации нескольких агентов.

Рекомендуемый путь обучения