一些容易混淆的点（个人记录）

config 和 configurable:

config

是框架的标准流转载体，必须是请求内的局部变量，并发处理

configurable

configurable是config字典中的特殊字段，专门用来装载动态、因人而异、随时可配置的自定义参数。
存储着thread_id 和 checkpointer_id，一个是线程id,一个是节点id，精准定位。

config = {
    # ------  config 层的“系统标准参数” ------
    "tags": ["v1.0", "production"],  # 给这次运行打标签，方便在 LangSmith 里搜索
    "metadata": {"source": "web"},   # 附加元数据
    "recursion_limit": 50,           # 【系统参数】：防止图陷入死循环，最多跑50步
    "callbacks": [my_custom_handler],# 【系统参数】：埋点器
    
    # ------  configurable (定制表单) ------
    "configurable": {
        # 【LangGraph 专属字段】
        "thread_id": "user_123",       # 抽屉钥匙
        "checkpoint_id": "1ef8...",    # 时光机钥匙
        
        # 【塞自己的自定义字段！】
        "user_vip_level": "gold",      # 比如传一个用户的 VIP 等级进去
        "llm_temperature": 0.7         # 动态决定这次大模型的温度
    }
}

Query 和 State：

Query：大门保安

from pydantic import BaseModel
# 这是 FastAPI 的专属数据类型
class Query(BaseModel):
    question: str
    thread_id: str

State：厨房工单

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

# 这是 LangGraph 的专属数据类型
class AgentState(TypedDict):
    # messages 字段专门用来存聊天记录
    # Annotated[..., add_messages] 是魔法，代表新消息会自动追加，而不是覆盖
    messages: Annotated[list, add_messages] 
    
    # 你还可以随意加别的字段，比如大模型如果查了天气，就存在这里
    # current_weather: str

Query在FastAPI的路由函数中传给State

@app.post("/ch@t/stream")
async def ch@t_stream(query: Query): # <--- 拿到 FastAPI 的 Query 对象
    
    # 1. 提取钥匙
    config = {"configurable": {"thread_id": query.thread_id}}
    
    # 2. 提取文本
    #    把 Query 里的纯文本 question，包装成 LangChain 标准的 HumanMessage，并塞进一个字典，这个字典的格式必须跟 AgentState 一模一样！
    initial_state_data = {
        "messages": [HumanMessage(content=query.question)]
    }
    
    # 3. 把初始的 state 数据交给 LangGraph 开始跑
    async for event in graph.astream_events(initial_state_data, config):
       

数据流通

config
thread_id
checkpointer_id
State[xx message]
config 中的 thread_id → checkpointer_id → 查历史checkpoints.db 中的 State → Reducer 合并 → 执行流转 → 存回 checkpointer"这一完整链路。

前后端流转

编排层流转

数据流转