本页目录

工程篇Note 13

第 13 章：源码阅读练习

围绕 reducer、工具协议、配置和异常路径的练习。

这一章不新增框架概念，而是把前 12 章变成可以检查的阅读能力。每题都指向当前项目真实代码；做完后再运行相应的真实 Agent 示例验证你的理解。

练习 1：解释一条消息怎么走到最终报告

从 deep_researcher.py 的 clarify_with_user 开始，按顺序写出：

输入 messages
-> ClarifyWithUser
-> ResearchQuestion
-> supervisor_messages
-> ConductResearch
-> researcher_messages
-> compressed_research
-> notes
-> final_report

要求：每一步写出对应的 state 字段、写它的节点、消费它的节点，以及 reducer 是追加还是覆盖。

验证：运行第 8 章的真实 Agent 示例，给每个节点返回的 state update 打印一个键名列表。不要打印 API Key 或整段长工具输出。

练习 2：为什么项目有三套消息历史

比较 state.py 中：

字段	归属	用途
`messages`	`AgentState`	用户输入、澄清和最终报告
`supervisor_messages`	`AgentState` / `SupervisorState`	研究计划与 researcher 委派结果
`researcher_messages`	`ResearcherState`	单个子题的 ReAct 工具循环

问题：如果用同一个 messages 承担三种职责，最终报告 prompt 会多出什么噪声？哪个子图会误读父图上下文？

验证：运行第 6 章。观察两个子图的结果是否分别回到各自的收集位置。

练习 3：手算 reducer 的结果

从 override_reducer 开始，不运行模型，写出下面更新后 raw_notes 的值：

current = ["old"]
append = ["new"]
replace = {"type": "override", "value": ["only-this"]}

答案应区分：

override_reducer(current, append)  -> ["old", "new"]
override_reducer(current, replace) -> ["only-this"]

再解释为什么 supervisor_messages 的初始 system prompt 必须用 override。

验证：运行第 2 章的真实 Command 示例；模型调用验证路由，手算验证 reducer 的确定性语义。

练习 4：工具协议的最小不变量

阅读 researcher_tools，列出一个 ToolMessage 必须保留的字段：

ToolMessage(
    content=observation,
    name=tool_call["name"],
    tool_call_id=tool_call["id"],
)

问题：

为什么同一条 AIMessage 的多个 tool call 能并发执行，却仍要一一对应 tool_call_id？
execute_tool_safely 为什么返回错误文本而不是直接抛异常？
原生 web search 为什么不在 tools_by_name 中再执行一遍？

验证：运行第 4 章，确认模型返回工具请求、代码执行工具、模型再根据 ToolMessage 完成回答这一完整闭环。

练习 5：定位配置覆盖问题

假设调用方传入：

context = Configuration(search_api="none")

但程序仍在使用 Tavily。检查顺序应是：

context.search_api 是否确实为 SearchAPI.NONE。
Configuration.from_env() 创建 context 时是否读到了 SEARCH_API 环境变量。
get_all_tools(context, store) 如何使用 search_api。
get_search_tool 返回了哪些工具。

验证：先执行第 11 章的最小配置检查，再运行第 5 章。后者只装配并调用 think_tool，不会触发真实搜索。

练习 6：画出异常传播边界

给下列函数标注“异常继续抛出”“转换为文本工具结果”“降级为原文”“结束图”：

函数	预期答案
`execute_tool_safely`	转换为文本工具结果
`summarize_webpage`	超时/异常降级为原文
`tavily_search_async`	继续向上抛出
`supervisor_tools`	当前代码中任意 researcher 异常会结束图
`final_report_generation`	token 超限时缩短 findings；其他异常返回错误报告

验证：不调用外部搜索。直接阅读第 12 章的对应路径，并用第 7 章的事件流理解“错误在哪个节点发生”。

练习 7：为一个修复选择正确的测试

场景 A：override_reducer 覆盖时没有清掉旧值。
场景 B：researcher 超过并发上限时，overflow 的 tool call 没得到消息。
场景 C：模型提供商更新异常类型，token 超限不再被识别。

选择测试：

场景	最小合适的验证
A	不调用模型的 reducer 单元测试
B	构造 `AIMessage.tool_calls` 的 async 子图测试，替换 researcher 子图为固定输出
C	用代表性异常对象测试 `is_token_limit_exceeded`，再做一次受控真实调用

原则：能由确定性输入验证的问题，不要消耗真实模型预算；涉及模型协议、工具回填或提供商响应时，再用章节中的真实 Agent 调用补验证。

完成标准

你可以不看提示，解释：

为什么 AgentState、SupervisorState、ResearcherState 不合并。
Command 如何同时改变状态与控制流。
模型调用、搜索调用、MCP 调用分别由谁付费、谁执行、谁负责权限。
一个 token-limit 异常最终在哪些节点可能被截断、降级或终止。

能回答这四题后，当前项目涉及的核心 LangGraph/LangChain 知识已经完整覆盖；下一阶段再向人机中断、长期记忆、人工审批、部署和生产评估扩展。

相关资源

查看示例代码：docs/langgraph-learning/examples/08_integrated_mini_researcher.py

"""Chapter 8: a compact mini deep researcher using the prior chapters."""

import asyncio
import operator
from typing import Annotated, Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, MessageLikeRepresentation, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.types import Command
from pydantic import BaseModel, Field
from typing_extensions import TypedDict

from open_deep_research.configuration import Configuration


load_dotenv()


class MiniPlan(BaseModel):
    """A small structured research plan."""

    research_brief: str = Field(description="One focused Chinese research brief.")
    topics: list[str] = Field(
        description="Exactly two focused Chinese subtopics.",
        min_length=2,
        max_length=2,
    )


class MiniState(MessagesState):
    research_brief: str
    topics: list[str]
    summaries: list[str]
    final_report: str
    route_log: Annotated[list[str], operator.add]


class ResearcherState(TypedDict):
    topic: str
    researcher_messages: Annotated[list[MessageLikeRepresentation], operator.add]
    summary: str


class ResearcherOutput(TypedDict):
    summary: str


configurable_model = init_chat_model(
    configurable_fields=("model", "max_tokens", "api_key"),
)


def configured_model(settings: Configuration, max_tokens: int = 180):
    return configurable_model.with_config(
        {
            "configurable": {
                "model": settings.research_model,
                "max_tokens": max_tokens,
            },
            "tags": ["langsmith:nostream"],
        }
    )


def text_content(content) -> str:
    if isinstance(content, list):
        return "".join(
            part.get("text", str(part)) if isinstance(part, dict) else str(part)
            for part in content
        )
    return str(content)


@tool
def record_learning_boundary(boundary: str) -> str:
    """Record one learning boundary before summarizing a research topic."""
    return f"已记录边界: {boundary}"


async def researcher_agent(
    state: ResearcherState,
    runtime: Runtime[Configuration],
) -> Command[Literal["researcher_tools", "__end__"]]:
    model = configured_model(runtime.context)
    has_tool_result = any(
        getattr(message, "type", None) == "tool"
        for message in state.get("researcher_messages", [])
    )
    if not has_tool_result:
        response = await model.bind_tools(
            [record_learning_boundary],
            tool_choice="record_learning_boundary",
        ).ainvoke(
            [
                HumanMessage(
                    content=(
                        f"研究主题: {state['topic']}。先调用工具记录一个学习边界。"
                    )
                )
            ]
        )
        return Command(
            update={"researcher_messages": [response]},
            goto="researcher_tools",
        )

    response = await model.ainvoke(
        state["researcher_messages"]
        + [HumanMessage(content=f"基于工具记录，用一句中文总结: {state['topic']}")]
    )
    return Command(
        update={"researcher_messages": [response], "summary": text_content(response.content)},
        goto=END,
    )


async def researcher_tools(
    state: ResearcherState,
) -> Command[Literal["researcher_agent"]]:
    last_message = state["researcher_messages"][-1]
    outputs = []
    for tool_call in last_message.tool_calls:
        outputs.append(
            ToolMessage(
                content=record_learning_boundary.invoke(tool_call["args"]),
                name=tool_call["name"],
                tool_call_id=tool_call["id"],
            )
        )
    return Command(update={"researcher_messages": outputs}, goto="researcher_agent")


researcher_graph = (
    StateGraph(
        ResearcherState,
        context_schema=Configuration,
        input_schema=ResearcherState,
        output_schema=ResearcherOutput,
    )
    .add_node("researcher_agent", researcher_agent)
    .add_node("researcher_tools", researcher_tools)
    .add_edge(START, "researcher_agent")
    .compile()
)


async def make_plan(
    state: MiniState,
    runtime: Runtime[Configuration],
) -> Command[Literal["run_researchers"]]:
    model = configured_model(runtime.context).with_structured_output(MiniPlan)
    plan = await model.ainvoke(
        state["messages"]
        + [
            HumanMessage(
                content=(
                    "把用户学习目标整理成一个 research_brief 和两个子主题，"
                    "必须聚焦当前 open_deep_research 项目。"
                )
            )
        ]
    )
    return Command(
        update={
            "research_brief": plan.research_brief,
            "topics": plan.topics,
            "route_log": ["make_plan -> run_researchers"],
        },
        goto="run_researchers",
    )


async def run_researchers(
    state: MiniState,
    runtime: Runtime[Configuration],
) -> Command[Literal["write_final"]]:
    results = await asyncio.gather(
        *(
            researcher_graph.ainvoke(
                {
                    "topic": topic,
                    "researcher_messages": [],
                    "summary": "",
                },
                context=runtime.context,
            )
            for topic in state["topics"]
        )
    )
    return Command(
        update={
            "summaries": [result["summary"] for result in results],
            "route_log": ["run_researchers -> write_final"],
        },
        goto="write_final",
    )


async def write_final(state: MiniState, runtime: Runtime[Configuration]):
    response = await configured_model(runtime.context, max_tokens=240).ainvoke(
        [
            HumanMessage(
                content=(
                    "请用三句中文写一个迷你研究报告。\n"
                    f"研究 brief: {state['research_brief']}\n"
                    f"研究摘要: {state['summaries']}"
                )
            )
        ]
    )
    return {
        "final_report": text_content(response.content),
        "messages": [response],
        "route_log": ["write_final -> END"],
    }


memory = InMemorySaver()
mini_researcher = (
    StateGraph(MiniState, context_schema=Configuration)
    .add_node("make_plan", make_plan)
    .add_node("run_researchers", run_researchers)
    .add_node("write_final", write_final)
    .add_edge(START, "make_plan")
    .add_edge("write_final", END)
    .compile(checkpointer=memory)
)


async def main():
    config = {"configurable": {"thread_id": "mini-researcher-learning"}}
    context = Configuration.from_env()
    input_value = {
        "messages": [
            HumanMessage(
                content=(
                    "我想通过 open_deep_research 学会 LangGraph 的工具循环和子图并发。"
                )
            )
        ],
        "route_log": [],
        "summaries": [],
    }

    event_count = 0
    stream = await mini_researcher.astream_events(
        input_value,
        config=config,
        context=context,
        version="v3",
    )
    async for _event in stream:
        event_count += 1

    snapshot = mini_researcher.get_state(config)
    values = snapshot.values
    print(f"事件数: {event_count}")
    print("主题: " + " | ".join(values["topics"]))
    print("路由: " + " | ".join(values["route_log"]))
    print("摘要数: " + str(len(values["summaries"])))
    print("最终报告: " + values["final_report"])
    print("持久化消息数: " + str(len(values["messages"])))


if __name__ == "__main__":
    asyncio.run(main())

查看示例代码：docs/langgraph-learning/examples/06_subgraphs_and_concurrency.py

"""Chapter 6: invoke two compiled subgraphs concurrently with real models."""

import asyncio
from typing import TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime

from open_deep_research.configuration import Configuration


load_dotenv()


class ResearcherState(TypedDict):
    topic: str
    summary: str


class ResearcherOutput(TypedDict):
    summary: str


class ParentState(TypedDict):
    topics: list[str]
    summaries: list[str]


configurable_model = init_chat_model(
    configurable_fields=("model", "max_tokens", "api_key"),
)


async def summarize_topic(
    state: ResearcherState,
    runtime: Runtime[Configuration],
):
    settings = runtime.context
    model = configurable_model.with_config(
        {
            "configurable": {
                "model": settings.research_model,
                "max_tokens": 120,
            },
            "tags": ["langsmith:nostream"],
        }
    )
    response = await model.ainvoke(
        [
            HumanMessage(
                content=f"只用一句中文概括这个 LangGraph 学习主题: {state['topic']}"
            )
        ]
    )
    return {"summary": str(response.content)}


researcher_graph = (
    StateGraph(
        ResearcherState,
        context_schema=Configuration,
        input_schema=ResearcherState,
        output_schema=ResearcherOutput,
    )
    .add_node("summarize_topic", summarize_topic)
    .add_edge(START, "summarize_topic")
    .add_edge("summarize_topic", END)
    .compile()
)


async def run_researchers(
    state: ParentState,
    runtime: Runtime[Configuration],
):
    results = await asyncio.gather(
        *(
            researcher_graph.ainvoke(
                {"topic": topic},
                context=runtime.context,
            )
            for topic in state["topics"]
        )
    )
    return {"summaries": [result["summary"] for result in results]}


async def main():
    parent_graph = (
        StateGraph(ParentState, context_schema=Configuration)
        .add_node("run_researchers", run_researchers)
        .add_edge(START, "run_researchers")
        .add_edge("run_researchers", END)
        .compile()
    )
    result = await parent_graph.ainvoke(
        {
            "topics": ["子图隔离状态", "asyncio.gather 并发调用"],
            "summaries": [],
        },
        context=Configuration.from_env(),
    )
    print(f"子图数: {len(result['summaries'])}")
    for summary in result["summaries"]:
        print("- " + summary)


if __name__ == "__main__":
    asyncio.run(main())

查看示例代码：docs/langgraph-learning/examples/02_state_and_command.py

"""Chapter 2: state reducers and Command routing with one real model call."""

import asyncio
import operator
from typing import Annotated, Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.types import Command

from open_deep_research.configuration import Configuration


load_dotenv()


class LearningState(MessagesState):
    """Conversation state plus an append-only record of graph routing."""

    route_log: Annotated[list[str], operator.add]


configurable_model = init_chat_model(
    configurable_fields=("model", "max_tokens", "api_key"),
)


async def answer(
    state: LearningState,
    runtime: Runtime[Configuration],
) -> Command[Literal["finish"]]:
    """Call the real model, update state, then choose the next node."""
    settings = runtime.context
    model = configurable_model.with_config(
        {
            "configurable": {
                "model": settings.research_model,
                "max_tokens": 80,
            },
            "tags": ["langsmith:nostream"],
        }
    )
    response = await model.ainvoke(state["messages"])
    return Command(
        update={
            "messages": [response],
            "route_log": ["answer -> finish"],
        },
        goto="finish",
    )


def finish(_: LearningState):
    """Append one local state update, then use its static edge to end."""
    return {"route_log": ["finish -> END"]}


async def main():
    graph = (
        StateGraph(LearningState, context_schema=Configuration)
        .add_node("answer", answer)
        .add_node("finish", finish)
        .add_edge(START, "answer")
        .add_edge("finish", END)
        .compile()
    )
    result = await graph.ainvoke(
        {
            "messages": [
                HumanMessage(content="只用一句中文解释状态机为什么适合 Agent。"),
            ],
            "route_log": [],
        },
        context=Configuration.from_env(),
    )
    print(f"消息数: {len(result['messages'])}")
    print("路由: " + " | ".join(result["route_log"]))
    print("答复: " + str(result["messages"][-1].content))


if __name__ == "__main__":
    asyncio.run(main())

查看示例代码：docs/langgraph-learning/examples/04_tool_loop.py

"""Chapter 4: a tiny real ReAct loop with one bound tool."""

import asyncio

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.types import Command

from open_deep_research.configuration import Configuration


load_dotenv()


@tool
def multiply_by_two(value: int) -> str:
    """Multiply the input integer by two."""
    return str(value * 2)


configurable_model = init_chat_model(
    configurable_fields=("model", "max_tokens", "api_key"),
)


async def agent(state: MessagesState, runtime: Runtime[Configuration]):
    settings = runtime.context
    model = configurable_model.with_config(
        {
            "configurable": {
                "model": settings.research_model,
                "max_tokens": 120,
            },
            "tags": ["langsmith:nostream"],
        }
    ).bind_tools([multiply_by_two])
    response = await model.ainvoke(state["messages"])
    if response.tool_calls:
        return Command(update={"messages": [response]}, goto="run_tools")
    return {"messages": [response]}


async def run_tools(state: MessagesState):
    last_message = state["messages"][-1]
    outputs = []
    for tool_call in last_message.tool_calls:
        result = multiply_by_two.invoke(tool_call["args"])
        outputs.append(
            ToolMessage(
                content=result,
                name=tool_call["name"],
                tool_call_id=tool_call["id"],
            )
        )
    return Command(update={"messages": outputs}, goto="agent")


async def main():
    graph = (
        StateGraph(MessagesState, context_schema=Configuration)
        .add_node("agent", agent)
        .add_node("run_tools", run_tools)
        .add_edge(START, "agent")
        .add_edge("run_tools", "agent")
        .compile()
    )
    result = await graph.ainvoke(
        {
            "messages": [
                HumanMessage(
                    content="请使用可用工具计算 21 的两倍，然后只用一句中文给出答案。"
                )
            ]
        },
        context=Configuration.from_env(),
    )
    print(f"消息数: {len(result['messages'])}")
    print("最终答复: " + str(result["messages"][-1].content))


if __name__ == "__main__":
    asyncio.run(main())

查看示例代码：docs/langgraph-learning/examples/05_search_and_mcp.py

"""Chapter 5: assemble project tools, then run one safe real tool loop."""

import asyncio

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage

from open_deep_research.configuration import Configuration
from open_deep_research.utils import get_all_tools, get_api_key_for_model


load_dotenv()


configurable_model = init_chat_model(
    configurable_fields=("model", "max_tokens", "api_key"),
)


def tool_name(tool):
    return tool.name if hasattr(tool, "name") else tool.get("name", "web_search")


async def main():
    settings = Configuration.from_env()
    tool_context = Configuration.model_validate(
        {**settings.model_dump(), "search_api": "none"}
    )
    tools = await get_all_tools(tool_context)
    print("可用工具: " + ", ".join(tool_name(tool) for tool in tools))

    think_tool = next(tool for tool in tools if tool_name(tool) == "think_tool")
    model = configurable_model.with_config(
        {
            "configurable": {
                "model": settings.research_model,
                "max_tokens": 160,
                "api_key": get_api_key_for_model(settings.research_model, settings),
            },
            "tags": ["langsmith:nostream"],
        }
    ).bind_tools([think_tool])

    messages = [
        HumanMessage(
            content=(
                "请先调用 think_tool，反思学习搜索与 MCP 时最该关注的一个边界，"
                "然后用一句中文总结。"
            )
        )
    ]
    first_response = await model.ainvoke(messages)
    messages.append(first_response)
    print(f"工具调用数: {len(first_response.tool_calls)}")

    tool_outputs = []
    for tool_call in first_response.tool_calls:
        result = think_tool.invoke(tool_call["args"])
        tool_outputs.append(
            ToolMessage(
                content=result,
                name=tool_call["name"],
                tool_call_id=tool_call["id"],
            )
        )

    final_response = await model.ainvoke(messages + tool_outputs)
    print("最终答复: " + str(final_response.content))


if __name__ == "__main__":
    asyncio.run(main())

查看示例代码：docs/langgraph-learning/examples/07_persistence_streaming_observability.py

"""Chapter 7: persist one thread, stream events, inspect saved state."""

import asyncio

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime

from open_deep_research.configuration import Configuration


load_dotenv()


configurable_model = init_chat_model(
    configurable_fields=("model", "max_tokens", "api_key"),
)


async def reply(state: MessagesState, runtime: Runtime[Configuration]):
    settings = runtime.context
    model = configurable_model.with_config(
        {
            "configurable": {
                "model": settings.research_model,
                "max_tokens": 120,
            },
            "tags": ["langsmith:nostream"],
        }
    )
    response = await model.ainvoke(state["messages"])
    return {"messages": [response]}


memory = InMemorySaver()
graph = (
    StateGraph(MessagesState, context_schema=Configuration)
    .add_node("reply", reply)
    .add_edge(START, "reply")
    .add_edge("reply", END)
    .compile(checkpointer=memory)
)


async def count_events(input_value, config, context):
    count = 0
    stream = await graph.astream_events(
        input_value,
        config=config,
        context=context,
        version="v3",
    )
    async for _event in stream:
        count += 1
    return count


async def main():
    config = {"configurable": {"thread_id": "learning-thread-1"}}
    context = Configuration.from_env()

    first_input = {
        "messages": [HumanMessage(content="你好，我叫老李。请记住我的名字。")]
    }
    event_count = await count_events(first_input, config, context)

    second_result = await graph.ainvoke(
        {
            "messages": [HumanMessage(content="我刚才说我叫什么？只用一句中文回答。")]
        },
        config=config,
        context=context,
    )
    snapshot = graph.get_state(config)

    print(f"事件数: {event_count}")
    print("第二轮答复: " + str(second_result["messages"][-1].content))
    print(f"持久化消息数: {len(snapshot.values['messages'])}")


if __name__ == "__main__":
    asyncio.run(main())