以下代码的开发环境：

[project]
name = "my-langgraph"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "dotenv>=0.9.9",
    "langchain>=0.3.27",
    "langchain-openai>=0.3.33",
    "langfuse>=3.5.0",
    "langgraph>=0.6.7",
    "langgraph-checkpoint-sqlite>=2.0.11",
    "numpy>=2.3.3",
]

一、 State 是什么

State = “全局上下文状态对象”，是 StateGraph 用来在节点之间传递和合并数据的统一容器。

• 它是一个 TypedDict 或 dataclass（推荐使用 TypedDict），描述了整张图要共享的字段。
• 每个节点接收到的输入 state 就是这个结构，节点返回的结果也会用来更新/合并这个 state。

二、 定义方式

1. 用 TypedDict

from typing import TypedDict

class MyState(TypedDict):
    question: str
    answer: str
    steps: list[str]

2 .用 dataclass (也可以)

from dataclasses import dataclass

@dataclass
class MyState:
    question: str
    answer: str = ""
    steps: list[str] = None

三、 State 在 StateGraph 中的工作机制

1.初始状态
你在运行图时传入一个初始 state（通常是一个字典），例如：

graph.invoke({"question": "什么是LangGraph？"})

2.节点执行
每个节点函数签名通常是：

def my_node(state: MyState) -> dict | MyState | None:
    ...

3.返回值合并规则

• 如果返回 dict 或 MyState，框架会用 state.update(...) 方式合并。
• 如果返回 None，状态保持不变。
• 如果返回其他类型（str、int、list…），不会更新状态。

def step1(state: MyState):
    # 产生新字段 answer
    return {"answer": f"回答：{state['question']}"}

def step2(state: MyState):
    # 添加一个步骤
    return {"steps": (state.get("steps") or []) + ["step2 done"]}

4.状态传递

• 每个节点执行后，更新后的 state 会传递给下游节点。
• 所以下游节点总是能拿到前面节点的最新结果。

在 StateGraph 里，节点的返回值会被合并进全局 state。合并规则很简单：

• 如果节点返回的是一个 dict 或你的 State 类型的实例，就会用这个结果去 update 现有的状态（类似 state.update(return_value)）。
• 如果节点返回的是其它类型（比如 str、int、list……），框架无法把它合并进状态，就会忽略更新，原有的 state 保持不变。

四、 与普通 Graph 的区别

特性	Graph	StateGraph
状态管理	手动传递返回值	自动合并全局状态
节点返回值要求	任意类型	推荐 dict / TypedDict 以便更新状态
流程复用	需要自己处理上下文	状态字段天然复用
适合场景	单纯流程编排	Agentic RAG、工作流自动化、上下文共享场景

五、 典型用法示例

from langgraph.graph import StateGraph
from typing import TypedDict

class QAState(TypedDict):
    question: str
    search_results: list[str]
    answer: str

def retrieve(state: QAState):
    # 假装检索
    return {"search_results": ["LangGraph 是一个构建 LLM 工作流的框架"]}

def generate_answer(state: QAState):
    context = " ".join(state.get("search_results", []))
    return {"answer": f"问题: {state['question']} 回答: {context}"}

graph = StateGraph(QAState)
graph.add_node("retriever", retrieve)
graph.add_node("generator", generate_answer)
graph.set_entry_point("retriever")
graph.add_edge("retriever", "generator")

result = graph.invoke({"question": "LangGraph 是什么？"})
print(result)

输出：

{
  'question': 'LangGraph 是什么？',
  'search_results': ['LangGraph 是一个构建 LLM 工作流的框架'],
  'answer': '问题: LangGraph 是什么？ 回答: LangGraph 是一个构建 LLM 工作流的框架'
}

六、 常见问题与坑

❌ 返回非字典值

def step(state):
    return "hello"  # ❌ 不会更新 state

✅ 正确写法

def step(state):
    return {"greeting": "hello"}

❌ 返回覆盖整个状态

如果你直接返回一个完整的 QAState 对象，会覆盖同名字段，所以通常只返回要更新的那部分。

✅ 建议

只返回增量更新：

def step(state):
    return {"answer": "新的答案"}

❌ 定义不完整

如果 TypedDict 没定义字段，后面更新时 IDE 可能无法提示。

✅ 建议

用 TypedDict 或 dataclass 显式声明所有需要共享的字段。

七、 实战建议

• 推荐用 TypedDict：IDE 类型提示友好，返回 dict 时兼容性高。
• 节点只返回增量字段：避免覆盖已有内容。
• 想完全控制状态合并：可以不用 StateGraph，直接用 Graph。
• 如果要流式输出：通常在状态中加入一个如 stream: list[str] 字段，逐步追加。