第一个LangGraph程序

环境就绪，开始写代码。本篇从一个最小的 Hello World 出发，逐行讲解 LangGraph 的核心 API，再升级到接入真实 LLM，让你直观感受"状态在节点间流转"是怎么一回事。

一、要做什么

我们要构建一个最简单的图：

graph LR
    S([START]) --> A[节点A 输出问候]
    A --> B[节点B 加工问候]
    B --> E([END])

• 节点 A：根据用户名字生成一句问候。
• 节点 B：把问候加工一下（比如加上 emoji）。
• 状态：在节点间共享的数据，含问候文本和一个步数计数器。

最终 invoke 一次拿到完整结果。

二、定义 State

LangGraph 的状态用 TypedDict 定义（也支持 Pydantic，详见 状态）：

from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    # 用户名（输入时给定）
    name: str
    # 问候文本（节点间传递）
    greeting: str
    # 步数计数器，用 reducer 累加（每节点返回的值会与旧值相加）
    step: Annotated[int, add]

要点：

• TypedDict 让 Python 知道"这个字典应该有哪些键"，写代码时有类型提示。
• Annotated[int, add] 表示 step 字段的更新方式不是"覆盖"，而是"用 operator.add 合并"——也就是累加。
• 字段没标 reducer 的，默认覆盖。

三、写节点函数

节点就是普通 Python 函数，签名 (state: State) -> dict，返回值是要更新的字段子集：

def greet(state: State) -> dict:
    """节点A：根据用户名生成问候。"""
    name = state["name"]
    greeting = f"你好，{name}！"
    # 返回要更新的字段。只更新 greeting 和 step。
    return {"greeting": greeting, "step": 1}

def decorate(state: State) -> dict:
    """节点B：给问候加装饰。"""
    text = state["greeting"]
    decorated = f"🎉 {text} 🎉"
    return {"greeting": decorated, "step": 1}

注意两个节点都返回 {"step": 1}——因为有 reducer add，这两个 1 会被累加成 2。如果没有 reducer，第二个节点的返回会覆盖第一个，最终 step=1。这就是 reducer 的意义。

四、组装图

from langgraph.graph import StateGraph, START, END

# 创建图，传入状态类型
graph = StateGraph(State)

# 注册节点：第一个参数是节点名（字符串），第二个是函数
graph.add_node("greet", greet)
graph.add_node("decorate", decorate)

# 连边
graph.add_edge(START, "greet")      # 入口：从 START 走到 greet
graph.add_edge("greet", "decorate") # greet 完了走 decorate
graph.add_edge("decorate", END)     # decorate 完了结束

# 编译：得到一个可执行的 CompiledGraph
app = graph.compile()

逐行解释：

• StateGraph(State)：创建一个状态图，状态类型是 State。
• add_node(name, fn)：注册一个节点。name 是字符串标识，后面连边时用它。
• add_edge(source, target)：连一条从 source 到 target 的边。
• START 和 END 是 LangGraph 内置的两个特殊节点，表示图的入口和出口。
• compile()：把图编译成可执行版本，会校验结构（比如是否所有节点都接到了边）。

五、运行

# 初始输入：必须包含 State 里定义的字段（或它们的子集）
initial_state = {"name": "麻瓜", "step": 0}

# invoke：一次性同步执行
result = app.invoke(initial_state)

print("最终状态：", result)
print(f"问候语：{result['greeting']}")
print(f"总步数：{result['step']}")

六、完整代码与输出

把上面拼起来：

# hello.py
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    name: str
    greeting: str
    step: Annotated[int, add]


def greet(state: State) -> dict:
    name = state["name"]
    return {"greeting": f"你好，{name}！", "step": 1}


def decorate(state: State) -> dict:
    text = state["greeting"]
    return {"greeting": f"🎉 {text} 🎉", "step": 1}


graph = StateGraph(State)
graph.add_node("greet", greet)
graph.add_node("decorate", decorate)
graph.add_edge(START, "greet")
graph.add_edge("greet", "decorate")
graph.add_edge("decorate", END)
app = graph.compile()

result = app.invoke({"name": "麻瓜", "step": 0})
print(result)

运行：

uv run python hello.py

输出：

{'name': '麻瓜', 'greeting': '🎉 你好，麻瓜！ 🎉', 'step': 2}

可以看到 step 是 2，正是两个节点各加 1 累加的结果。greeting 经过了两次更新。

七、用 stream 看每一步

invoke 只给最终结果。想看每步状态变化，用 stream：

for event in app.stream({"name": "麻瓜", "step": 0}):
    print(event)

输出：

{'greet': {'greeting': '你好，麻瓜！', 'step': 1}}
{'decorate': {'greeting': '🎉 你好，麻瓜！ 🎉', 'step': 2}}

每个事件是某个节点的更新。第一次事件是 greet 节点的输出，第二次是 decorate 节点的输出。流式详见 编译与运行。

八、升级版：接入真实 LLM

把 greet 节点换成调用 LLM 生成问候，演示状态在节点间正常流转：

# hello_llm.py
import os
from typing import TypedDict, Annotated
from operator import add
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

load_dotenv()


class State(TypedDict):
    name: str
    greeting: str
    step: Annotated[int, add]


# 初始化模型（兼容 OpenAI 接口的任意服务）
llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


def greet(state: State) -> dict:
    """节点A：调用 LLM 生成一句有创意的问候。"""
    name = state["name"]
    msg = llm.invoke([
        HumanMessage(content=f"用一句话热情地问候 {name}，要求 15 字以内。"),
    ])
    return {"greeting": msg.content, "step": 1}


def decorate(state: State) -> dict:
    """节点B：加装饰。"""
    text = state["greeting"]
    return {"greeting": f"🎉 {text} 🎉", "step": 1}


graph = StateGraph(State)
graph.add_node("greet", greet)
graph.add_node("decorate", decorate)
graph.add_edge(START, "greet")
graph.add_edge("greet", "decorate")
graph.add_edge("decorate", END)
app = graph.compile()


result = app.invoke({"name": "麻瓜", "step": 0})
print("问候语：", result["greeting"])
print("总步数：", result["step"])

输出可能像：

问候语：  🎉 嘿，麻瓜！欢迎来到 LangGraph 的世界！ 🎉
总步数： 2

可以看到，节点不关心"上一步是谁"，只看当前 state。换成 LLM 后整个图结构完全没变，只改了节点内部实现——这就是状态解耦的好处。

九、画图

想直观看看图长什么样，可以打印 mermaid：

print(app.get_graph().draw_mermaid())

输出大致：

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD
    __start__([<p>__start__</p>]):::first
    greet([greet])
    decorate([decorate])
    __end__([<p>__end__</p>]):::last
    __start__ --> greet
    greet --> decorate
    decorate --> __end__

把这段贴到任何支持 mermaid 的地方（如 GitHub README、VitePress 文档）就能渲染成流程图。

十、常见踩坑

1. 忘了 add_edge(START, ...)：报错 ValueError: Graph must have an entrypoint。START 是图的唯一入口，必须有一条从 START 出发的边。
2. 节点返回了 State 里没定义的 key：这个 key 会被静默忽略，不会进 state。如果发现状态里没你期望的字段，先检查 State 定义。
3. 节点返回 None 或不返回 dict：节点应返回 dict（哪怕空 {}）。返回 None 在某些版本会被当成"无更新"，新手最好显式 return \{\}。
4. 想要累加却没用 reducer：忘记 Annotated[int, add]，每个节点的 step 都覆盖前一个，最终只看到最后一个节点的值。
5. 传给 invoke 的初始 state 字段不完整：可选字段可以缺，但节点里要读的字段必须有。否则 KeyError。
6. 以为 invoke 返回的是最后一步的输出：错——invoke 返回的是最终完整 state，所有字段都在。这是新手常误解的点。
7. 图里没有 END：图必须有出口，否则要么编译失败，要么无限循环。
8. 节点顺序依赖：add_node 的调用顺序不影响执行顺序——执行顺序由边决定。把图想象成接线，不是排队。

十一、小结

• 一个最小 LangGraph 程序：定义 State → 写节点函数 → StateGraph 加节点加边 → compile → invoke。
• 节点是普通 Python 函数，返回要更新的字段子集。
• Annotated[T, reducer] 控制字段更新方式，默认覆盖。
• START 是入口、END 是出口，都必须显式连边。
• invoke 返回最终完整 state，不是最后一步的输出。
• 用 stream 看每一步状态变化。
• 把节点换成 LLM 调用，图结构不用动——状态在节点间解耦流转。

下一篇我们把所有核心概念串成一张总览图。核心概念总览 →