LangGraph-Tutorial für den Einstieg

LangGraph ist einLow-Level-Agent-Orchestrierungsframework, das vom LangChain-Team entwickelt wurdeund speziell für den Aufbau zustandsbehafteter, lang laufender KI-Workflows konzipiert wurde.

Im Gegensatz zu traditionellen linearen LLM-Aufrufketten modelliert LangGraph Workflows alsgerichteten Graphen：

• Knoten: Funktionen, die bestimmte Aktionen ausführen (z. B. Aufrufen von LLMs, Ausführen von Tools, Verarbeiten von Daten)
• Kante: Definiert Flusspfade zwischen Knoten und unterstützt bedingte Verzweigungen
• Zustand: Daten, die im gesamten Workflow gemeinsam genutzt und weitergegeben werden

Open-Source-Repository:https://github.com/langchain-ai/langgraph。

Stellen Sie sich vor, Sie dirigieren eine Symphonie: Eine traditionelle LLM-Kette ist wie das Abspielen eines Stücks von Anfang bis Ende, nur sequenziell; LangGraph hingegen ist wie ein Dirigent, der die Aufführungsreihenfolge jederzeit anhand der Reaktionen des Live-Publikums anpassen, einen bestimmten Satz wiederholen oder zu einer bestimmten Passage springen kann. Es verleiht KI-Workflows die Weisheit eines „Dirigenten“ – fähig zu Schleifen, Verzweigungen und Rückverfolgung, und ermöglicht so wirklich komplexe autonome Entscheidungsfindung.

Warum LangGraph wählen?

Die folgende Tabelle vergleicht die wichtigsten Unterschiede zwischen traditionellen LLM-Ketten und LangGraph:

Geeignete Szenarien

• Chatbots: Müssen sich den Kontext mehrrundiger Dialoge merken
• Autonomer Agent: Kann planen, Tools verwenden und Denkprozesse iterieren
• Multi-Agent-System: Mehrere KIs arbeiten zusammen, um komplexe Aufgaben zu erledigen
• Genehmigungsworkflow: Ein automatisierter Prozess, der eine menschliche Überprüfung erfordert
• Recherche-Assistent: Erfordert mehrstufiges Schlussfolgern und Informationsabruf

Kernkonzepte

Bevor Sie mit dem Schreiben von Code beginnen, sollten Sie zunächst die drei Kernkonzepte von LangGraph verstehen.

Graph

Der Graph ist der Bauplan des gesamten Workflows und definiert die vollständige logische Struktur des Agenten. Er besteht aus Knoten und Kanten:

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

State

Der Zustand ist diegemeinsam genutzte Datenstruktur, die durch den gesamten Graphen läuft. Jeder Knoten kann den Zustand lesen und aktualisieren, und der aktualisierte Zustand wird an den nächsten Knoten weitergegeben.

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                           # 步骤计数

Knoten

Knoten sind gewöhnliche Python-Funktionen, die den aktuellen Zustand empfangen und den aktualisierten Zustand (Teilfelder) zurückgeben.

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

Kanten

Kanten definieren den Fluss zwischen Knoten:

• Normale Kante: fester Pfad,node_a -> node_b
• Bedingte Kante: leitet dynamisch basierend auf dem Zustand weiter,node_a -> node_b oder node_c
• Startkante：START -> 第一个节点
• Endkante：某节点 -> END

Umgebungs-Setup

Abhängigkeiten installieren

LangGraph und zugehörige Abhängigkeiten mit einem inländischen Mirror installieren:

# 创建虚拟环境（推荐）
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-Schlüssel konfigurieren

Im Projektstammverzeichnis, erstellen Sie.enveine Datei, und konfigurieren Sie OpenAI und 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

Installation überprüfen

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

Ihr erstes LangGraph-Programm

Beginnen wir mit dem einfachsten Beispiel – einem linearen Workflow mit nur zwei Knoten.

Ausführungsergebnis:

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

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

Graphstruktur visualisieren

In Jupyter Notebook können Sie die Graphstruktur direkt visualisieren:

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

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

Zustandsverwaltung

Zustand mit TypedDict definieren

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]

Zustand mit Pydantic definieren (für Produktion empfohlen)

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 (integrierter Komfort-Zustand)

LangGraph bietet integriertenMessagesState, speziell für Konversationsszenarien entwickelt:

from langgraph.graph import MessagesState

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

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

Knoten

Reguläre Funktionsknoten

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

LLM-Aufrufknoten

Asynchrone Knoten

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": [...]})

Klassen als Knoten verwenden

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)

Kanten und bedingtes Routing

Reguläre Kanten

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

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

Bedingte Kanten

Bedingte Kanten sind ein Kernmerkmal von LangGraph und entscheiden dynamisch über den nächsten Schritt basierend auf dem aktuellen State.

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 时 -> 结束
    }
)

Parallele Ausführung (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")

Vollständiges Beispiel für bedingtes Routing

Beispiel für ein Ausführungsergebnis:

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

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

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

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

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

Einen Chatbot erstellen

Nutze nun das Gelernte, um einen Roboter zu bauen, der Multi-Turn-Konversationen unterstützt. Dieser Roboter kann den Gesprächskontext behalten und ein kontinuierliches interaktives Erlebnis bieten.

Beispiel für ein Ausführungsergebnis:

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

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

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

You: 退出
再见！

Tool-Aufruf - ReAct-Agent

Dem Agenten zu ermöglichen, externe Tools zu verwenden, ist eine der leistungsstärksten Funktionen von LangGraph.ReAct(Reason + Act) ist das gebräuchlichste Agentenmuster: LLM denkt -> wählt ein Tool aus -> führt das Tool aus -> beobachtet das Ergebnis -> denkt weiter.

Tools definieren

Definiere zunächst die Tools, die der Agent verwenden kann:

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' oder  '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]

Einen ReAct-Agenten erstellen

Streaming des Tool-Aufrufprozesses

# 流式观察 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-Zusammenarbeit

LangGraph unterstützt nativ das Pausieren während der Workflow-Ausführung, um auf menschliche Überprüfung oder Eingabe zu warten. Dies ist sehr nützlich für sensible Vorgänge, die eine menschliche Bestätigung erfordern.

interrupt verwenden, um die Ausführung zu pausieren

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)

Vollständiger Genehmigungs-Workflow

Breakpoints auf Kanten setzen

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

Persistenter Speicher

LangGraph bietet einen integrierten Mechanismus zur Zustandspersistenz, der es Agenten ermöglicht, den Gesprächsverlauf über Sitzungen hinweg zu behalten.

In-Memory-Speicherung (geeignet für Entwicklung und Tests)

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）

Persistente SQLite-Speicherung (geeignet für lokale Projekte)

# 安装依赖
# 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
    )

Gesprächsverlauf anzeigen

# 获取某个 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'])}")

Multi-Agent-Systeme

LangGraph zeichnet sich durch die Koordination mehrerer spezialisierter Agenten aus, die zusammenarbeiten. Durch die Zerlegung komplexer Aufgaben und deren Zuweisung an verschiedene Experten-Agenten können leistungsfähigere Problemlösungsfähigkeiten erreicht werden.

Supervisor-Muster

Teilgraph

Komplexe Teilprozesse als Teilgraphen kapseln und im Hauptgraphen wiederverwenden:

# 将复杂子流程封装为子图，在主图中复用
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()

Visuelles Debugging mit LangGraph Studio

LangGraph Studio ist die offizielle visuelle Entwicklungsumgebung, mit der Sie den Ausführungsprozess des Agenten in Echtzeit verfolgen können, was die Entwicklungs- und Debugging-Effizienz erheblich verbessert.

LangGraph CLI installieren

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

Projektkonfigurationsdatei erstellen

Im Projektstammverzeichnis erstellenlanggraph.jsonKonfigurationsdatei:

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

Entwicklungsserver starten

langgraph dev

Nach dem Start besuchen Siehttp://localhost:8123Anschließend können Sie LangGraph Studio in Ihrem Browser verwenden.

Hauptfunktionen von Studio

• Echtzeit-Visualisierung: Zeigt den Ausführungsprozess der Knoten grafisch an, um den Workflow-Status intuitiv zu verstehen
• State-Inspektion: An jedem Knoten anhalten, um den aktuellen State anzuzeigen, was die Fehlersuche erleichtert
• Zeitreise: Historische Ausführungsschritte erneut abspielen, um Statusänderungen bei jedem Schritt zu verfolgen
• Hot Reload: Aktualisiert die Graphstruktur nach Codeänderungen automatisch, kein Neustart des Dienstes erforderlich

Best Practices und FAQs

Best Practices

Wichtige Punkte beim State-Design

• State knapp halten und nur notwendige Felder enthalten
• Explizite Reducer für komplexe Felder definieren (z. B.add_messages）
• Pydantic-Modelle verwenden, um State-Typen in Produktionsumgebungen zu validieren
• Vermeiden Sie das Speichern übergroßer Objekte im State; ziehen Sie externen Speicher in Betracht

Wichtige Punkte beim Knoten-Design

• Jeder Knoten hat eine einzige Verantwortung, was Tests und Wiederverwendung erleichtert
• Knotenfunktionen sollten idempotent sein (gleiche Eingabe erzeugt gleiche Ausgabe)
• Vermeiden Sie das direkte Ändern des übergebenen States in Knoten; geben Sie stattdessen neue Werte zurück
• Async-Knoten sinnvoll für I/O-intensive Operationen einsetzen

Fehlerbehandlung

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}"}]
        }

Endlosschleifen vermeiden

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

Häufig gestellte Fragen (FAQ)

F1: Was ist, wenn das Format des Rückgabewerts eines Knotens falsch ist?

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

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

F2: Wie übergebe ich temporäre Daten zwischen Knoten?

Fügen Sie temporäre Daten zur State-Definition hinzu oder verwenden Sie eine Unterstrich-Präfix-Konvention für interne Felder:

from typing import TypedDict

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

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

F3: Wie debuggt man den Knotenausführungsprozess?

# 使用 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}")

F4: Was ist der Unterschied zwischen StateGraph und MessageGraph?

MessageGraphEs ist eine frühere Version der API mit relativ begrenzten Funktionen. Jetzt wird empfohlen, einheitlich zu verwendenStateGraph, was flexibler und funktionsreicher ist. Für die Verarbeitung von Nachrichten verwenden SieStateGraph(MessagesState)oder passen Sie eine an, diemessagesFelder im State enthält.

Zusammenfassung

Dieser Artikel führt systematisch in die Kernkonzepte und praktischen Anwendungen des LangGraph-Frameworks ein. Durch graphbasierte Workflow-Orchestrierung ermöglicht LangGraph Entwicklern, komplexe KI-Agenten mit Fähigkeiten wie Schleifen, Verzweigungen und Zustandspersistenz zu erstellen. Im Vergleich zu traditionellen linearen LLM Chains bietet LangGraph erhebliche Vorteile in Szenarien, die mehrstufiges Reasoning, Mensch-Maschine-Zusammenarbeit und Multi-Agenten-Koordination erfordern.

Empfohlener Lernpfad