1
ChatModel
2
Prompt
3
Message
4
Tool
5
OutputParser
6
Runnable
7
Retriever
8
Middleware
9
create_agent

1
LangChain 不只是 Chain；
2
LangChain Core 的关键是 Runnable 抽象；
3
LangChain Agent 的关键是基于 LangGraph 的 agent runtime；
4
create_agent 成为构建 Agent 的标准入口之一；
5
middleware 成为模型调用、工具调用、上下文处理、安全控制的重要扩展点。

1
StateGraph
2
CompiledStateGraph
3
Node
4
Edge
5
Conditional Edge
6
Reducer
7
Checkpoint
8
Store
9
Thread
10
Interrupt
11
Command
12
Subgraph
13
Functional API
14
Time Travel
15
Durable Execution
16
Pregel Runtime

1
LangGraph 不是普通流程图；
2
它是有状态、可恢复、可中断、可回放的 Agent / Workflow 执行框架；
3
生产级 Agent 的很多能力，如 memory、HITL、time travel、durable execution，都建立在 checkpoint 和 runtime 机制之上。

1
1. Runnable
2
2. ChatPromptTemplate / Messages
3
3. BaseChatModel / AIMessage / ToolMessage
4
4. @tool / StructuredTool
5
5. create_agent
6
6. StateGraph
7
7. add_node / add_edge / add_conditional_edges
8
8. Reducer / Annotated state
9
9. Checkpointer / thread_id
10
10. interrupt / Command
11
11. stream / events / tracing
12
12. subgraph / multi-agent graph

1
这个功能解决哪个 Agent 范式问题？

1
add_conditional_edges 解决 Router / Workflow 分支问题；
2
checkpoint 解决长期状态、恢复执行和 HITL 问题；
3
@tool 解决模型连接外部能力的问题；
4
create_agent 解决 ReAct 式模型-工具循环问题。

1
这个范式在 LangChain / LangGraph 中用哪个 API 表达？

1
Tool-use → @tool / StructuredTool
2
Workflow → StateGraph
3
Router → add_conditional_edges
4
HITL → interrupt + Command
5
Memory → checkpointer + thread_id

1
运行时发生了什么？

1
graph.invoke(input)
2
→ 初始化 state
3
→ 执行 START 指向的 node
4
→ node 返回 partial update
5
→ reducer 合并 state
6
→ edge 决定下一个节点
7
→ checkpoint 保存状态
8
→ 直到 END

1
源码为什么这样设计？

1
为什么 Runnable 要统一 invoke / stream / batch？
2
为什么 StateGraph node 返回 partial state？
3
为什么 checkpoint 要绑定 thread_id？
4
为什么 interrupt 必须依赖 checkpointer？
5
为什么 ToolMessage 要重新放回 messages？

1
为什么 prompt | model | parser 可以组合；
2
为什么每个组件都能 invoke / stream / batch；
3
为什么普通函数可以变成 RunnableLambda；
4
为什么 LangChain 能把 Prompt、Model、Parser、Retriever、Tool 放在同一套执行协议下。

1
普通 LLM Chain
2
→ LangChain RunnableSequence

1
User Input
2
→ Prompt
3
→ Model
4
→ Parser
5
→ Output

1
chain = prompt | model | parser

1
from langchain_core.prompts import ChatPromptTemplate
2
from langchain_core.output_parsers import StrOutputParser
3
from langchain.chat_models import init_chat_model
4

5
model = init_chat_model("openai:gpt-4.1-mini")
6

7
prompt = ChatPromptTemplate.from_messages([
8
    ("system", "你是一个旅行规划助手。"),
9
    ("user", "请为 {destination} 规划 {days} 天行程。")
10
])
11

12
chain = prompt | model | StrOutputParser()
13

14
result = chain.invoke({
15
    "destination": "东京",
16
    "days": 5
17
})
18

19
print(result)

1
chain.invoke(input)
2
  ↓
3
ChatPromptTemplate.invoke(input)
4
  ↓
5
生成 PromptValue / messages
6
  ↓
7
ChatModel.invoke(messages)
8
  ↓
9
返回 AIMessage
10
  ↓
11
StrOutputParser.invoke(AIMessage)
12
  ↓
13
返回字符串

1
langchain_core/runnables/base.py
2
langchain_core/runnables/config.py
3
langchain_core/runnables/utils.py
4
langchain_core/runnables/passthrough.py
5
langchain_core/runnables/branch.py

1
Runnable
2
RunnableSequence
3
RunnableParallel
4
RunnableLambda
5
RunnableConfig

1
1. Runnable 的核心抽象方法有哪些？
2
2. invoke、ainvoke、stream、batch 的默认实现关系是什么？
3
3. `|` 运算符如何构造 RunnableSequence？
4
4. RunnableConfig 如何向下传递？
5
5. callbacks、tags、metadata 如何进入执行链？
6
6. RunnableSequence 如何逐个调用子 Runnable？
7
7. RunnableParallel 如何并行调用多个子 Runnable？

1
prompt | model
2
prompt | model | parser

1
这是 RunnableSequence；
2
输入是什么；
3
中间输出是什么；
4
最终输出是什么；
5
是否支持 stream；
6
是否能加 with_retry / with_config。

1
01-langchain-runnable-source-deep-dive.md

1
PromptTemplate
2
ChatPromptTemplate
3
MessagesPlaceholder
4
SystemMessage
5
HumanMessage
6
AIMessage
7
ToolMessage
8
BaseChatModel

1
Context Engineering / Prompt Engineering
2
→ ChatPromptTemplate + Messages + ChatModel

1
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
2

3
prompt = ChatPromptTemplate.from_messages([
4
    ("system", "你是旅行规划助手，需要根据用户偏好规划行程。"),
5
    MessagesPlaceholder("history"),
6
    ("user", "{input}")
7
])
8

9
messages = prompt.invoke({
10
    "history": [],
11
    "input": "我想去大阪玩 4 天，喜欢美食和城市漫步"
12
})
13

14
print(messages)

1
输入变量
2
  ↓
3
ChatPromptTemplate 格式化
4
  ↓
5
PromptValue
6
  ↓
7
messages list
8
  ↓
9
ChatModel 接收 messages
10
  ↓
11
返回 AIMessage

1
langchain_core/prompts/
2
langchain_core/messages/
3
langchain_core/language_models/chat_models.py

1
BaseMessage
2
HumanMessage
3
SystemMessage
4
AIMessage
5
ToolMessage
6
BaseChatModel
7
ChatPromptTemplate
8
MessagesPlaceholder

1
1. ChatPromptTemplate.invoke 返回的对象是什么？
2
2. PromptValue 如何转换为 messages？
3
3. AIMessage 为什么能携带 tool_calls？
4
4. response_metadata 和 usage_metadata 在哪里保存？
5
5. BaseChatModel.invoke 和 _generate 的关系是什么？
6
6. 不同模型供应商的返回值如何统一为 AIMessage？

1
02_langchain_core_prompt_message_chatmodel_source_deep_dive.md

1
StrOutputParser
2
JsonOutputParser
3
PydanticOutputParser
4
with_structured_output
5
schema validation
6
parser failure
7
retry / repair

1
意图识别 / 槽位抽取 / 计划生成 / 责任判断
2
→ 结构化输出

1
from pydantic import BaseModel, Field
2
from langchain.chat_models import init_chat_model
3

4
class TravelIntent(BaseModel):
5
    destination: str | None = Field(description="目的地")
6
    days: int | None = Field(description="旅行天数")
7
    preferences: list[str] = Field(description="用户偏好")
8
    need_clarification: bool = Field(description="是否需要追问")
9

10
model = init_chat_model("openai:gpt-4.1-mini")
11
structured_model = model.with_structured_output(TravelIntent)
12

13
result = structured_model.invoke(
14
    "我想去东京玩 5 天，喜欢美食和城市漫步"
15
)
16

17
print(result)

1
Pydantic schema
2
  ↓
3
转换为模型可理解的结构化输出约束
4
  ↓
5
模型生成结构化结果
6
  ↓
7
框架解析并校验
8
  ↓
9
返回 Pydantic 对象或 dict

1
langchain_core/output_parsers/
2
langchain_core/language_models/chat_models.py
3
langchain_core/utils/function_calling.py

1
1. with_structured_output 底层如何绑定 schema？
2
2. Pydantic schema 如何转换为 JSON schema？
3
3. 结构化输出和 tool calling 有什么关系？
4
4. 解析失败时异常在哪里抛出？
5
5. parser 应该放在 chain 里，还是依赖模型原生 structured output？

1
IntentResult
2
TravelPlan
3
SearchQuery
4
ReflectionResult
5
FinalResponse

1
字段
2
含义
3
是否必填
4
失败样例
5
校验规则

1
03_langchain_core_structured_output_source_deep_dive.md

1
@tool
2
StructuredTool
3
BaseTool
4
args_schema
5
tool name
6
tool description
7
tool call
8
ToolMessage
9
tool error handling

1
Tool-use Agent
2
→ LangChain Tool / StructuredTool

1
from langchain.tools import tool
2

3
@tool
4
def search_attractions(city: str, preference: str) -> str:
5
    """Search attractions in a city based on user preference."""
6
    return f"{city} attractions for {preference}"
7

8
print(search_attractions.name)
9
print(search_attractions.description)
10
print(search_attractions.args)

1
Python function
2
  ↓
3
@tool 包装
4
  ↓
5
BaseTool / StructuredTool
6
  ↓
7
生成 name / description / args_schema
8
  ↓
9
绑定到模型
10
  ↓
11
模型输出 tool_calls
12
  ↓
13
工具执行层调用真实函数
14
  ↓
15
结果包装为 ToolMessage
16
  ↓
17
重新放回 messages

1
langchain_core/tools/
2
langchain/tools/
3
langchain_core/messages/tool.py

1
BaseTool
2
StructuredTool
3
ToolException
4
ToolMessage

1
1. @tool 装饰器如何读取函数名、参数和 docstring？
2
2. args_schema 如何从函数签名生成？
3
3. Tool.invoke 和普通函数调用有什么区别？
4
4. ToolMessage 如何保存 tool_call_id？
5
5. 工具异常如何被捕获并返回给模型？
6
6. 工具描述如何影响模型选择？

1
04_langchain_core_tool_calling_source_deep_dive.md

1
create_agent
2
model node
3
tools node
4
middleware
5
tool loop
6
stop condition
7
iteration limit
8
structured response

1
ReAct
2
→ model call
3
→ tool call
4
→ ToolMessage
5
→ model call
6
→ final answer

1
AIMessage
2
tool_calls
3
ToolMessage
4
final AIMessage

1
from langchain.agents import create_agent
2
from langchain.tools import tool
3

4
@tool
5
def search_city_info(query: str) -> str:
6
    """Search city information for travel planning."""
7
    return f"Search result for {query}"
8

9
agent = create_agent(
10
    model="openai:gpt-4.1-mini",
11
    tools=[search_city_info],
12
    system_prompt="你是一个旅行规划助手。"
13
)
14

15
result = agent.invoke({
16
    "messages": [
17
        {"role": "user", "content": "帮我规划东京 5 天行程，需要查一下热门景点"}
18
    ]
19
})
20

21
print(result)

1
用户 messages
2
  ↓
3
model node
4
  ↓
5
AIMessage 是否包含 tool_calls？
6
  ↓
7
如果有：tools node 执行工具
8
  ↓
9
ToolMessage 加入 messages
10
  ↓
11
回到 model node
12
  ↓
13
直到模型输出 final answer 或达到停止条件

1
langchain/agents/
2
langchain/agents/factory.py
3
langgraph/prebuilt/
4
langchain_core/messages/
5
langchain_core/tools/

1
1. create_agent 返回的是什么对象？
2
2. 它内部是否构建了 LangGraph graph？
3
3. model node 和 tools node 如何连接？
4
4. tool_calls 如何被识别？
5
5. 工具执行结果如何重新进入 messages？
6
6. middleware 在模型调用前后能做什么？
7
7. stop condition 在哪里判断？

1
05_langchain_create_agent_react_source_deep_dive.md

1
StateGraph
2
State schema
3
add_node
4
add_edge
5
START
6
END
7
compile
8
CompiledStateGraph
9
partial state update

1
Workflow Agent
2
→ StateGraph + Node + Edge + State

1
from typing_extensions import TypedDict
2
from langgraph.graph import StateGraph, START, END
3

4
class TravelState(TypedDict):
5
    user_request: str
6
    destination: str | None
7
    days: int | None
8
    plan: str | None
9

10
def parse_request(state: TravelState):
11
    return {
12
        "destination": "东京",
13
        "days": 5
14
    }
15

16
def generate_plan(state: TravelState):
17
    return {
18
        "plan": f"为 {state['destination']} 生成 {state['days']} 天旅行计划"
19
    }
20

21
builder = StateGraph(TravelState)
22

23
builder.add_node("parse_request", parse_request)
24
builder.add_node("generate_plan", generate_plan)
25

26
builder.add_edge(START, "parse_request")
27
builder.add_edge("parse_request", "generate_plan")
28
builder.add_edge("generate_plan", END)
29

30
graph = builder.compile()
31

32
result = graph.invoke({
33
    "user_request": "我想去东京玩 5 天",
34
    "destination": None,
35
    "days": None,
36
    "plan": None
37
})
38

39
print(result)

1
StateGraph 构建阶段
2
  ↓
3
注册节点
4
  ↓
5
注册边
6
  ↓
7
compile 生成可执行图
8
  ↓
9
invoke 初始化 state
10
  ↓
11
按边执行节点
12
  ↓
13
节点返回 partial update
14
  ↓
15
合并 state
16
  ↓
17
到达 END

1
langgraph/graph/state.py
2
langgraph/graph/graph.py
3
langgraph/pregel/

1
StateGraph
2
CompiledStateGraph
3
Pregel

1
1. StateGraph 为什么是 builder，而不是执行器？
2
2. add_node 如何保存节点函数？
3
3. add_edge 如何保存图结构？
4
4. compile 做了哪些校验？
5
5. CompiledStateGraph 如何实现 Runnable？
6
6. node 返回 partial update 后如何合并到 state？

1
START
2
→ parse_user_request
3
→ route_by_intent
4
→ generate_plan
5
→ search_tools
6
→ optimize_plan
7
→ final_response
8
→ END

1
输入 state
2
输出 partial update
3
是否调用 LLM
4
是否调用 tool
5
是否是路由节点
6
是否可能失败

1
06_langgraph_stategraph_source_deep_dive.md

1
Router Agent / Workflow Branch
2
→ add_conditional_edges

1
from typing_extensions import TypedDict, Literal
2
from langgraph.graph import StateGraph, START, END
3

4
class TravelState(TypedDict):
5
    user_request: str
6
    intent: str | None
7
    result: str | None
8

9
def classify_intent(state: TravelState):
10
    text = state["user_request"]
11
    if "预算" in text:
12
        intent = "budget_planning"
13
    elif "景点" in text:
14
        intent = "attraction_planning"
15
    else:
16
        intent = "general_planning"
17
    return {"intent": intent}
18

19
def route_by_intent(state: TravelState) -> Literal[
20
    "budget_node",
21
    "attraction_node",
22
    "general_node"
23
]:
24
    if state["intent"] == "budget_planning":
25
        return "budget_node"
26
    if state["intent"] == "attraction_planning":
27
        return "attraction_node"
28
    return "general_node"

1
classify_intent 节点执行
2
  ↓
3
state.intent 被更新
4
  ↓
5
route_by_intent(state)
6
  ↓
7
返回下一个节点名
8
  ↓
9
LangGraph 进入对应节点

1
langgraph/graph/state.py
2
langgraph/graph/branch.py

1
1. add_conditional_edges 如何注册 path function？
2
2. path function 的返回值可以是什么？
3
3. 返回 END 时如何终止？
4
4. 返回多个节点时如何并行？
5
5. mapping 参数有什么作用？
6
6. Literal 类型提示对可视化和校验有什么帮助？

1
07_langgraph_conditional_edge_router_source_deep_dive_v2.md

1
并行检索 / 多 Agent 协作 / 多工具结果聚合
2
→ Reducer

1
from typing import Annotated
2
from operator import add
3
from typing_extensions import TypedDict
4
from langgraph.graph import StateGraph, START, END
5

6
class TravelState(TypedDict):
7
    user_request: str
8
    research_notes: Annotated[list[str], add]
9

10
def search_food(state: TravelState):
11
    return {"research_notes": ["推荐美食：寿司、拉面、居酒屋"]}
12

13
def search_attractions(state: TravelState):
14
    return {"research_notes": ["推荐景点：浅草寺、涩谷、上野公园"]}

1
START 同时触发 search_food 和 search_attractions
2
  ↓
3
两个节点都写 research_notes
4
  ↓
5
LangGraph 根据 Annotated[list, add] 合并
6
  ↓
7
merge_plan 读取合并后的 research_notes

1
langgraph/graph/state.py
2
langgraph/channels/
3
langgraph/pregel/

1
1. State schema 如何解析 Annotated？
2
2. Reducer 是在什么时候被调用的？
3
3. 多节点写同一个 key 时如何判断是否冲突？
4
4. 覆盖式更新和追加式更新有什么区别？
5
5. messages 为什么通常需要特殊 reducer？

1
08_langgraph_reducer_parallel_state_source_deep_dive.md

1
Plan-and-Execute
2
→ planner node
3
→ plan validator
4
→ executor node / subgraph
5
→ step verifier
6
→ replan

1
START
2
→ parse_request
3
→ planner_node
4
→ validate_plan
5
→ execute_step
6
→ check_step_result
7
→ route_next_step
8
  ├── continue_execute
9
  ├── replan
10
  └── final_response

1
from typing_extensions import TypedDict
2
from langgraph.graph import StateGraph, START, END
3

4
class PlanStep(TypedDict):
5
    step_id: int
6
    task: str
7
    status: str
8

9
class TravelState(TypedDict):
10
    user_request: str
11
    plan_steps: list[PlanStep]
12
    current_step_index: int
13
    final_answer: str | None
14

15
def planner_node(state: TravelState):
16
    return {
17
        "plan_steps": [
18
            {"step_id": 1, "task": "确认目的地和天数", "status": "pending"},
19
            {"step_id": 2, "task": "查询景点和交通信息", "status": "pending"},
20
            {"step_id": 3, "task": "生成每日行程", "status": "pending"},
21
        ],
22
        "current_step_index": 0
23
    }
24

25
def route_next_step(state: TravelState):
26
    if state["current_step_index"] >= len(state["plan_steps"]):
27
        return "final_response"
28
    return "execute_step"

1
langgraph/graph/state.py
2
langgraph/graph/branch.py
3
langgraph/pregel/

1
1. 计划列表应该存在 state 的哪个字段？
2
2. current_step_index 如何控制执行进度？
3
3. 每一步执行结果如何写回 state？
4
4. replan 是重新覆盖计划，还是追加修复步骤？
5
5. executor 应该直接调用工具，还是进入子图？

1
Planner 输出了什么？
2
Executor 每一步怎么选？
3
工具结果如何影响下一步？
4
什么时候需要 replan？
5
什么时候直接 final？

1
09_langgraph_plan_execute_subgraph_source_deep_dive.md

1
Reflection
2
→ generator node
3
→ evaluator node
4
→ revise node
5
→ conditional loop

1
class TravelState(TypedDict):
2
    draft_plan: str | None
3
    critique: str | None
4
    quality_score: float | None
5
    revision_count: int
6
    final_plan: str | None

1
generate_draft
2
→ evaluate_draft
3
→ route_by_score
4
  ├── revise_draft
5
  └── final_response
6
→ revise_draft
7
→ evaluate_draft

1
MAX_REVISION = 2
2

3
def route_by_score(state):
4
    if state["quality_score"] >= 0.8:
5
        return "final_response"
6
    if state["revision_count"] >= MAX_REVISION:
7
        return "final_response"
8
    return "revise_draft"

1
langgraph/graph/branch.py
2
langgraph/pregel/

1
1. 反思循环如何避免无限执行？
2
2. revision_count 应该如何更新？
3
3. evaluator 输出应该结构化吗？
4
4. evaluator 和 generator 是否应该用不同模型？
5
5. 高风险任务中 Reflection 能不能替代 Policy / Verifier？

1
Reflection 适合质量优化，不适合替代规则校验；
2
必须设置 max_revision_count；
3
必须有明确评价指标；
4
如果新版本更差，需要保留旧版本或回滚；
5
高风险动作不能由 Reflection 自动放行。

1
10_langgraph_reflection_loop_source_deep_dive.md

1
Memory / Long-running Agent / Durable Workflow / HITL
2
→ checkpoint + thread_id

1
from typing_extensions import TypedDict
2
from langgraph.graph import StateGraph, START, END
3
from langgraph.checkpoint.memory import InMemorySaver
4

5
class TravelState(TypedDict):
6
    user_request: str
7
    plan: str | None
8

9
def plan_node(state: TravelState):
10
    return {"plan": "生成旅行计划草稿"}
11

12
builder = StateGraph(TravelState)
13
builder.add_node("plan_node", plan_node)
14
builder.add_edge(START, "plan_node")
15
builder.add_edge("plan_node", END)
16

17
checkpointer = InMemorySaver()
18
graph = builder.compile(checkpointer=checkpointer)
19

20
config = {"configurable": {"thread_id": "travel-thread-001"}}
21

22
graph.invoke({"user_request": "东京 5 天旅行", "plan": None}, config=config)
23

24
latest_state = graph.get_state(config)
25
history = list(graph.get_state_history(config))

1
graph.invoke(input, config)
2
  ↓
3
根据 thread_id 找到执行线程
4
  ↓
5
每个 super-step 后保存 checkpoint
6
  ↓
7
get_state 读取最新状态
8
  ↓
9
get_state_history 读取历史状态

1
langgraph/checkpoint/
2
langgraph/pregel/
3
langgraph/types.py

1
thread_id
2
checkpoint
3
StateSnapshot
4
super-step
5
writes
6
next
7
metadata

1
1. checkpoint 保存了哪些信息？
2
2. thread_id 如何区分不同会话？
3
3. get_state 和 get_state_history 从哪里读数据？
4
4. super-step 的边界在哪里？
5
5. checkpoint 如何支持 time travel？
6
6. checkpoint 如何支持故障恢复？

1
11_langgraph_checkpoint_thread_durable_execution_source_deep_dive.md

1
Human-in-the-loop
2
→ interrupt()
3
→ checkpoint 保存
4
→ Command(resume=...)

1
from typing_extensions import TypedDict
2
from langgraph.graph import StateGraph, START, END
3
from langgraph.types import interrupt, Command
4
from langgraph.checkpoint.memory import InMemorySaver
5

6
class TravelState(TypedDict):
7
    plan: str
8
    approved: bool | None
9

10
def approval_node(state: TravelState):
11
    approved = interrupt({
12
        "question": "是否确认使用该旅行计划？",
13
        "plan": state["plan"]
14
    })
15
    return {"approved": approved}
16

17
def final_node(state: TravelState):
18
    if state["approved"]:
19
        return {"plan": state["plan"] + "\n用户已确认。"}
20
    return {"plan": "用户未确认，需要重新规划。"}
21

22
builder = StateGraph(TravelState)
23
builder.add_node("approval_node", approval_node)
24
builder.add_node("final_node", final_node)
25
builder.add_edge(START, "approval_node")
26
builder.add_edge("approval_node", "final_node")
27
builder.add_edge("final_node", END)
28

29
graph = builder.compile(checkpointer=InMemorySaver())
30

31
config = {"configurable": {"thread_id": "travel-approval-001"}}
32

33
graph.invoke({"plan": "东京 5 天行程草案", "approved": None}, config=config)
34

35
graph.invoke(Command(resume=True), config=config)

1
执行到 approval_node
2
  ↓
3
调用 interrupt(payload)
4
  ↓
5
图暂停
6
  ↓
7
checkpoint 保存当前 state
8
  ↓
9
外部系统展示 payload
10
  ↓
11
用户确认
12
  ↓
13
graph.invoke(Command(resume=True), config)
14
  ↓
15
interrupt 返回 True
16
  ↓
17
图继续执行 final_node

1
langgraph/types.py
2
langgraph/pregel/
3
langgraph/checkpoint/

1
1. interrupt 如何让图暂停？
2
2. 暂停时 state 保存在哪里？
3
3. Command(resume=...) 如何把值传回 interrupt？
4
4. 为什么 interrupt 必须依赖 checkpointer？
5
5. 多个 interrupt 同时存在时如何恢复？

1
12_langgraph_interrupt_hitl_source_deep_dive.md

1
stream
2
astream
3
stream_mode="values"
4
stream_mode="updates"
5
stream_mode="messages"
6
event stream
7
metadata
8
callbacks

1
Agent Trace / Debugging / Observability
2
→ stream / events / tracing

1
for chunk in graph.stream(
2
    {"user_request": "东京 5 天旅行"},
3
    stream_mode="updates"
4
):
5
    print(chunk)

1
langgraph/pregel/
2
langchain_core/runnables/base.py
3
langchain_core/tracers/

1
1. stream 与 invoke 的执行路径有什么不同？
2
2. stream_mode 如何影响输出？
3
3. token streaming 和 node update streaming 有什么区别？
4
4. callbacks 和 events 如何记录执行过程？
5
5. LangSmith trace 如何接入？

1
13_langgraph_streaming_events_observability_source_deep_dive.md

1
thread_id
2
checkpoint
3
StateSnapshot
4
store
5
namespace
6
message trimming
7
summarization
8
memory write policy

1
Memory / Context Engineering
2
→ checkpointer + store + messages state + retrieval policy

1
graph = builder.compile(checkpointer=checkpointer, store=store)
2

3
config = {
4
    "configurable": {
5
        "thread_id": "trip-thread-001",
6
        "user_id": "jupiter",
7
    }
8
}
9

10
graph.invoke({"messages": [("user", "记住我偏好城市漫步")]}, config=config)

1
langgraph/checkpoint/
2
langgraph/store/
3
langgraph/pregel/
4
langgraph/graph/message.py
5
langchain_core/messages/

1
1. checkpoint 保存的是执行状态，还是用户长期记忆？
2
2. store 的 namespace 和 key 如何设计？
3
3. message trimming 与 summarization 应该发生在模型调用前还是图运行后？
4
4. 节点如何读取和写入 store？
5
5. 长期记忆写入失败是否应该中断主图？

1
14_langgraph_memory_store_context_engineering_source_deep_dive.md

1
middleware
2
before_model
3
after_model
4
wrap_model_call
5
wrap_tool_call
6
tool retry
7
PII / safety guardrails
8
dynamic model selection
9
context editing

1
Agent Runtime Control / Guardrails
2
→ create_agent(..., middleware=[...])

1
agent = create_agent(
2
    model=model,
3
    tools=tools,
4
    middleware=[
5
        dynamic_model_middleware,
6
        pii_guardrail_middleware,
7
        tool_retry_middleware,
8
    ],
9
)

1
langchain/agents/factory.py
2
langchain/agents/middleware/
3
langchain/agents/structured_output.py
4
langgraph/prebuilt/tool_node.py

1
1. middleware 是改图结构，还是包装节点执行？
2
2. before_model / after_model 与 wrap_model_call 有什么边界？
3
3. tool retry 应该由 ToolNode、middleware 还是业务工具处理？
4
4. guardrail 拒绝后返回什么状态？
5
5. middleware 顺序如何影响最终行为？

1
15_langchain_agent_middleware_guardrails_source_deep_dive.md

1
Document
2
Retriever
3
VectorStore
4
retrieval tool
5
query rewrite
6
rerank
7
context assembly
8
grounding

1
Agentic RAG
2
→ retriever node / retrieval tool / query rewrite node / grounding evaluator