创建与连接节点

节点（Node）是图里干活的函数。上一篇我们定义好了状态这块"黑板"，本篇就把一个个函数挂上去，并连成一条流水线。

一、节点是什么

节点就是一个普通 Python 函数，签名固定：

def my_node(state: State) -> dict:
    ...
    return {"some_field": new_value}

• 输入：当前完整状态（一个 dict 或你定义的 TypedDict/Pydantic 实例）。
• 输出：一个部分更新字典，只包含要改的字段。框架会自动合并回主状态。
• 节点之间不直接互调，只通过状态交换数据——这就是"解耦"。

如果你写过 LangGraph4j（Java），对应的就是 AppStateAwareNode，区别是 Python 里就是普通函数，没有继承。

二、节点的多种写法

1. 最简单的普通函数

def greet(state):
    return {"message": f"你好，{state['user_input']}！"}

2. 带 config 参数（拿到运行时配置）

第二个固定参数 config 能拿到 RunnableConfig，里面常放 thread_id、回调、自定义元数据：

from langchain_core.runnables import RunnableConfig

def greet(state, config: RunnableConfig):
    thread_id = config.get("configurable", {}).get("thread_id", "unknown")
    return {"message": f"[会话{thread_id}] 你好，{state['user_input']}！"}

调用时这样传：

app.invoke(
    {"user_input": "阿宝"},
    config={"configurable": {"thread_id": "t-1"}},
)

3. 异步节点 async

跑大模型、查数据库时用异步更高效。ainvoke / astream 会自动走异步路径：

async def agenerate(state):
    # 假设这是一个异步的模型调用
    answer = await fake_llm_call(state["user_input"])
    return {"answer": answer}

# 编译后用 ainvoke 运行
result = await app.ainvoke({"user_input": "你好"})

三、节点如何读写状态

读：直接 state["字段"]；写：返回值里带这个字段即可。

class State(TypedDict):
    user_input: str
    answer: str
    call_count: int

def step(state: State) -> dict:
    q = state["user_input"]              # 读
    n = state.get("call_count", 0) + 1   # 读（带默认值）
    return {"answer": f"第{n}次回答", "call_count": n}  # 写

关键：不要在节点里修改 state 对象本身（如 state["x"] = 1）。正确做法是 return {"x": 1}，让框架合并。直接改 state 在某些模式下不生效。

四、用 add_edge 串联节点

add_edge(A, B) 表示"执行完 A，下一步执行 B"：

from langgraph.graph import StateGraph, START, END

graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve")   # 起点 → retrieve
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)     # generate → 终点

flowchart LR
    START --> retrieve --> generate --> END

五、用 add_conditional_edges 分支

当你要"根据状态决定走哪条路"时，用条件边。详见 条件边，这里给最小例子：

def route(state) -> str:
    # 返回字符串，表示下一个节点名
    if state["score"] >= 0.8:
        return "generate"
    return "rewrite"

graph.add_conditional_edges(
    "grade",            # 从哪个节点出发
    route,              # 路由函数
    ["generate", "rewrite"],  # 可能的去向（可选，用于可视化/校验）
)

六、完整工作流：检索→评分→生成

下面这个例子串起三个节点，并用条件边在"评分够/不够"之间分流：

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    user_input: str
    docs: list[str]
    score: float
    answer: str

def retrieve(state):
    docs = [f"关于「{state['user_input']}」的片段"]
    return {"docs": docs, "score": 0.6}  # 假装检索完顺便给了个分

def rewrite(state):
    # 评分不够，改写查询再检索一次（这里简化为加一条片段）
    return {"docs": state["docs"] + ["补充片段"], "score": 0.9}

def generate(state):
    return {"answer": f"综合 {len(state['docs'])} 个片段生成答案"}

# 路由：分数够就去 generate，不够去 rewrite
def route_after_grade(state) -> str:
    return "generate" if state["score"] >= 0.8 else "rewrite"

graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("rewrite", rewrite)
graph.add_node("generate", generate)

graph.add_edge(START, "retrieve")
graph.add_conditional_edges("retrieve", route_after_grade, ["generate", "rewrite"])
graph.add_edge("rewrite", "generate")
graph.add_edge("generate", END)

app = graph.compile()
result = app.invoke({"user_input": "什么是状态机"})
print(result)

每步状态变化（用 stream_mode="updates" 观察）

for chunk in app.stream({"user_input": "什么是状态机"}, stream_mode="updates"):
    print(chunk)

输出大致是：

{'retrieve': {'docs': [...], 'score': 0.6}}
{'rewrite':   {'docs': [..., '补充片段'], 'score': 0.9}}
{'generate':  {'answer': '综合 2 个片段生成答案'}}

能清楚看到每个节点返回的增量更新，这正是调试利器。更多见 流式输出。

七、常见踩坑

踩坑 1：节点签名写错

• 节点必须是函数（或可调用对象），不能是类本身。
• 异步节点必须写 async def，否则 ainvoke 不会真正并发。
• config 参数名固定，且必须是第二个位置参数，不能写成 cfg。

踩坑 2：返回了 State 里不存在的 key

def bad(state):
    return {"summarry": "..."}  # 拼错成 summarry，State 里没有，更新被忽略

养成习惯：返回前对照 State 定义。开 Pydantic 状态模式可更早暴露这类错误。

踩坑 3：条件边的路由函数返回了不存在的节点名

route 返回 "generate" 但图里没有 generate 节点，运行时会抛 KeyError。建议在 add_conditional_edges 第三参数列出所有可能去向，框架会校验。

踩坑 4：在节点里直接改 state

def bad(state):
    state["x"] = 1     # ❌ 不要直接改
    return state       # 也不要返回整个 state

正确做法只返回更新字段。直接改 state 在带检查点/并行场景下行为不可预期。

踩坑 5：循环没有终止条件

retrieve → rewrite → retrieve 这种环如果没有出口（条件边永远不导向 END），会无限循环或耗尽步数报错。一定要有计数器或评分出口。

八、小结

• 节点是普通函数，签名 (state) 或 (state, config)，返回部分更新字典。
• add_edge 串联，add_conditional_edges 做分支。
• 节点间通过状态解耦，不要直接互调。
• 调试用 stream(stream_mode="updates") 逐步看每个节点的增量。
• 下一篇 构建完整工作流 会把所有概念整合成一个完整例子。