回调与调试（全课程收官章）

本章配套可运行示例：demo/demo_callback.py。demo 里写了两个自定义回调——控制台计时器 + 文件日志器——一次挂到同一条荐书链上；Ollama 在线用真模型，离线自动换成内置假模型，钩子触发完全相同（回调挂的是「事件」，不是某个具体模型），离线也能完整跑通。运行后还会在 demo/ 目录生成 callback_run.log，亲眼看到文件日志回调写了什么。

本章你将学会

• 回调机制：BaseCallbackHandler 的生命周期钩子——on_chain_start / on_chat_model_start / on_llm_new_token / on_llm_end / on_tool_start 等，链上每个环节的一举一动都能被你听见
• 一个必踩的坑：聊天模型触发的是 on_chat_model_start，不是 on_llm_start——只写后者，挂在 ChatOllama 上一个事件都收不到
• callbacks 参数挂载：调用时传、构造时传、一次挂多个（多回调并存）
• 两个实用回调：控制台计时回调（哪一步慢了一目了然）、文件日志回调（输入输出追加写进日志文件）
• run_id / parent_run_id：从一串 UUID 里读出整棵「执行树」
• 调试三件套：set_debug(True) 全局开关（顺带说清 verbose 和环境变量在 1.x 的现状）、astream_events 事件流、以及一张「什么时候用哪个」的选择地图
• 全课程收官：12 章知识地图、从模板到 Agent 的主线串讲、后续学习方向（LangGraph / LangSmith / 真实项目）

上一章（11 · Agent）收尾时，拾光书屋的 AI 店员已经全须全尾：会荐书（02-05 章）、懂店里的私货（06-08 章）、记得住聊过什么（09 章）、有手脚能查库存算折扣（10 章）、还能自己排工序（11 章）。然后它上线了。

上线第三天，你收到两条反馈：「有时候要干等二十秒」、「有次回答纯属胡编」。你打开代码——prompt | model | parser，或者一整个 Agent——中间发生了什么，一行日志都没有。链是一个黑盒：进去一个问题，出来一个答案，慢在哪一步、错在哪一环，全靠猜。

第 04 章我们给链上过三张保险单：with_fallbacks 换备胎、with_retry 再敲两下门、超时拔电源。但那些都是「出了事之后怎么办」——事发现场没有目击者。本章的回调（callback）补上这块：它是 LangChain 里最底层的观测机制——任何一个 Runnable（模型、链、工具、检索器）跑起来的每一步，都会向外广播事件；你只要写一个「接收器」挂上去，就能在现场听见一切。

比喻：把链想成一条流水线。04 章的 fallback/retry 是流水线尽头的保险丝——炸了知道换备用线；回调是给每个工位装监控探头——哪个工位开工了、干了几秒、交出什么货、中途有没有摔东西，实时传到你的监控室。保险丝管善后，探头管现场，一个都不能少。

顺带一提：本章也是全课程的最后一章。回调这个「旁观一切」的视角，恰好是把前面 11 章串起来复盘的最佳观景台——收官内容放在第 6 节。

技术基线：Python + LangChain 1.x（langchain + langchain-ollama），本地 Ollama 跑 qwen3:4b。没装的话回 01 章看环境准备。本章多数示例不需要模型也能跑（回调机制与模型无关），demo 离线可运行。

1. 回调是什么：给流水线的每个工位装探头

作用：不改动链的任何一行代码，把「执行过程」变成可观测的。你在 invoke / stream 时往 config 里塞一个回调对象，框架在关键节点上自动调用你的钩子方法。

核心类：BaseCallbackHandler（from langchain_core.callbacks import BaseCallbackHandler）。它本身所有钩子都是空方法，你继承它、覆写关心的那几个，其余的保持沉默。

最小示例——一个「什么都记」的回调，挂到 03 章的老朋友荐书链上（本例不需要真模型，用一个内置的假模型代替，钩子触发与真模型完全一致）：

from typing import Any
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate


class EchoHandler(BaseCallbackHandler):
    """把收到的每个事件原样打印出来。"""

    def on_chain_start(
        self, serialized: dict[str, Any], inputs: dict[str, Any],
        *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any,
    ) -> None:
        name = (serialized or {}).get("name") or "Runnable"
        print(f"[链开始] {name}，输入：{inputs}")

    def on_chain_end(
        self, outputs: dict[str, Any],
        *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any,
    ) -> None:
        print(f"[链结束] 输出：{str(outputs)[:50]}")

    def on_chat_model_start(
        self, serialized: dict[str, Any], messages: list[list[Any]],
        *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any,
    ) -> None:
        print(f"[模型开始] 收到 {len(messages[0])} 条消息")

    def on_llm_new_token(
        self, token: str, *, run_id: UUID,
        parent_run_id: UUID | None = None, **kwargs: Any,
    ) -> None:
        print(f"[新token] {token!r}")

    def on_llm_end(
        self, response: Any, *, run_id: UUID,
        parent_run_id: UUID | None = None, **kwargs: Any,
    ) -> None:
        print(f"[模型结束] 生成：{response.generations[0][0].text[:50]}")


prompt = ChatPromptTemplate.from_messages([("human", "用一句话推荐《{book}》")])
model = GenericFakeChatModel(messages=iter([AIMessage(content="《夜航船》：夜航人的掌灯。")]))
chain = prompt | model | StrOutputParser()

for chunk in chain.stream({"book": "夜航船"}, config={"callbacks": [EchoHandler()]}):
    pass

运行后你会看到一长串输出（节选、略去 run_id 细节）：

[链开始] Runnable，输入：{'input': ''}
[链开始] ChatPromptTemplate，输入：{'book': '夜航船'}
[链结束] 输出：messages=[HumanMessage(content='用一句话推荐《夜航船》'...
[模型开始] 收到 1 条消息
[新token] '《夜航船'
[新token] '》：夜航'
[新token] '人的掌灯'
[新token] '。'
[模型结束] 生成：《夜航船》：夜航人的掌灯。
[链结束] 输出：《夜航船》：夜航人的掌灯。

注意看事件顺序背后的结构：整条链是一个 run，链里的每个组件（提示词、模型、解析器）各自是子 run。每个事件都带着两个身份证号：

• run_id：本次 run（一次组件执行）的唯一编号；
• parent_run_id：父 run 的编号——组件的 parent 是整条链，整条链的 parent 是 None。

把这两个号连起来，就是一棵执行树：

RunnableSequence（整条链）          run_id=…981  parent=None
├── ChatPromptTemplate（提示词）    run_id=…982  parent=…981
├── 模型                            run_id=…983  parent=…981
│     └── （流式吐字：每块一个 on_llm_new_token）
└── StrOutputParser（解析器）       run_id=…984  parent=…981

第 3 节的文件日志回调和第 5 节的 astream_events，读的都是这棵树。排查「二十秒到底慢在哪」时，你只要看每个 run 的开始与结束时间差，一目了然。

常见坑：

• 钩子名全是 on_xxx，不是 JS 版的 handleXxx。从 JS 迁移的肌肉记忆要改：handleLLMStart → on_llm_start，handleChainStart → on_chain_start，首字母小写、驼峰变蛇形，详见 1.1 的对照表。
• serialized 参数别指望太多。1.x 里它不再是完整的序列化对象，只剩 name / id 等少量字段（体积和安全考虑）。想拿组件名，写 (serialized or {}).get("name")；拿更多细节走 kwargs 里的 metadata / tags。
• 覆写钩子时建议保留 **kwargs。框架往钩子里塞的关键字参数（tags、metadata、invocation_params……）版本间会增减，写死签名容易在新版本上直接 TypeError。
• 钩子里抛异常不会炸掉主流程，但会被框架捕获并打一条告警日志——回调代码还是要自己小心，别让探头冒烟。

1.1 JS → Python 钩子名对照表

Python 版比 JS 版多出一类钩子值得单独认识：on_chat_model_start——这是 Python 版最容易踩的坑，1.2 节展开。

1.2 最大的坑：on_llm_start 收不到聊天模型的事件

LangChain 的「模型」分两种血统，开场的钩子不一样：

ChatOllama 是聊天模型——只触发 on_chat_model_start，不触发 on_llm_start。你若只覆写了 on_llm_start，挂在荐书链上会一个事件都收不到，还以为回调坏了。稳妥的写法是两个都覆写，或者在 on_chat_model_start 里干活。

钩子参数的形状也不同：on_llm_start 拿到的是拼好的字符串列表 prompts；on_chat_model_start 拿到的是「双层列表」messages——外层对应一次批量调用里的每个请求（通常长度为 1），内层才是这个请求的消息列表（system / human / ……）。取当前请求的消息写 messages[0]，每条消息的正文在 msg.content、角色在 msg.type。

2. 挂载方式：callbacks 参数的三个姿势

作用：把回调接到链上。有三种接法，覆盖从「临时调试」到「生产常驻」的场景。

2.1 调用时传（最常用）

invoke / stream / batch 的 config 参数里塞 callbacks，只对这一次调用生效：

chain.invoke({"book": "夜航船"}, config={"callbacks": [EchoHandler()]})

注意 callbacks 的值是列表——这直接引出多回调并存：一次挂几个都行，事件会广播给每一个：

result = chain.invoke(
    {"book": "夜航船"},
    config={"callbacks": [EchoHandler(), ConsoleTimingHandler(), FileLogHandler()]},
)

控制台要计时、文件要留档、屏幕要回显——互不干扰，各干各的。第 3 节会把「计时 + 文件日志」这对组合写成完整代码。

2.2 挂在整条链上，还是挂在某个组件上？

挂在最外层（整条链）时，事件会向下继承：链上的每个组件执行时，你的 handler 都能收到对应事件——第 1 节的执行树就是这么看到的（你只挂了一次，却看见了提示词、模型、解析器各自的 start/end）。

想只盯某一个环节，就把回调挂在那个组件上；或者调用时给 run 打上 tags、命名 run_name，在回调里按这些标记挑着看（5.2 节的过滤参数同理）。

2.3 构造时传（常驻）

模型/链构造时就绑上，之后每次调用自动生效，适合生产环境的日志回调：

model = ChatOllama(model="qwen3:4b", callbacks=[FileLogHandler()])

常见坑：

• config 是第几个参数别搞混。chain.invoke(input, config) 里 config 是第二个位置参数；写成关键字 config={...} 最稳。
• 构造时传的回调会常驻。调试用的 EchoHandler 挂在构造参数里，等你上线了它会一直刷屏——临时观测用 2.1 的调用时传法，用完即走。
• callbacks 里放的是实例，不是类。[EchoHandler()] 对，[EchoHandler] 错——后者不会报错但也永远不会触发，静默失败最坑人。

3. 两个实用回调：计时器与文件日志

光会打印不够，本节写两个真的能上生产的回调。场景还是那个：用户说等了二十秒，到底慢在哪？ 控制台计时回调负责现场看，文件日志回调负责留档查。

3.1 控制台计时回调：每个 run 掐一块表

思路：on_chain_start / on_chat_model_start 时按下秒表（记下 run_id → 开始时刻），对应的 end 钩子里查表算差值。run_id 是天然的字典键——每个 run 一块表，互不串线。

import time
from typing import Any
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler


class ConsoleTimingHandler(BaseCallbackHandler):
    """控制台计时回调：每个 run 开始掐表，结束报耗时。"""

    def __init__(self) -> None:
        self._starts: dict[UUID, float] = {}

    def on_chain_start(
        self, serialized: dict[str, Any], inputs: dict[str, Any],
        *, run_id: UUID, **kwargs: Any,
    ) -> None:
        self._starts[run_id] = time.perf_counter()
        name = (serialized or {}).get("name") or "链"
        print(f"[计时] ▶ {name} 开工")

    def on_chain_end(self, outputs: dict[str, Any], *, run_id: UUID, **kwargs: Any) -> None:
        elapsed = time.perf_counter() - self._starts.pop(run_id, time.perf_counter())
        print(f"[计时] ■ 结束，耗时 {elapsed:.2f}s")

    def on_chat_model_start(
        self, serialized: dict[str, Any], messages: list[list[Any]],
        *, run_id: UUID, **kwargs: Any,
    ) -> None:
        self._starts[run_id] = time.perf_counter()
        print(f"[计时] ▶ 模型开工（{len(messages[0])} 条消息进来）")

    def on_llm_end(self, response: Any, *, run_id: UUID, **kwargs: Any) -> None:
        elapsed = time.perf_counter() - self._starts.pop(run_id, time.perf_counter())
        print(f"[计时] ■ 模型收工，耗时 {elapsed:.2f}s")

用 time.perf_counter() 而不是 time.time()——前者是单调高精度计时器，不受系统改时间影响，掐表就该用它。

3.2 文件日志回调：把现场写进档案

思路：追加模式（"a"）打开日志文件，模型开始时记输入、结束时记输出、出错时记异常。工具调用也顺手记上——排查「胡编」类问题，先看它到底调了什么工具、拿到了什么数据。

from pathlib import Path
from typing import Any
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler


class FileLogHandler(BaseCallbackHandler):
    """文件日志回调：模型的输入输出、工具调用、错误，追加写进日志文件。"""

    def __init__(self, path: Path) -> None:
        self._path = path

    def _write(self, text: str) -> None:
        with self._path.open("a", encoding="utf-8") as f:
            f.write(text)

    def on_chat_model_start(
        self, serialized: dict[str, Any], messages: list[list[Any]],
        *, run_id: UUID, **kwargs: Any,
    ) -> None:
        model_name = (serialized or {}).get("name", "未知模型")
        lines = "\n".join(f"  [{m.type}] {m.content}" for m in messages[0])
        self._write(f"[模型开始] {model_name}（run {str(run_id)[:8]}）\n{lines}\n")

    def on_llm_end(self, response: Any, *, run_id: UUID, **kwargs: Any) -> None:
        text = response.generations[0][0].text
        self._write(f"[模型结束] 生成：{text}\n\n")

    def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None:
        self._write(f"[模型出错] {error!r}\n\n")

    def on_tool_start(
        self, serialized: dict[str, Any], input_str: str,
        *, run_id: UUID, **kwargs: Any,
    ) -> None:
        name = (serialized or {}).get("name", "未知工具")
        self._write(f"[工具开始] {name}({input_str})\n")

    def on_tool_end(self, output: Any, *, run_id: UUID, **kwargs: Any) -> None:
        self._write(f"[工具结束] 返回：{output}\n")

两个要点：

• on_llm_end 里取文本：response.generations[0][0].text。generations 是「双层列表」（外层对应批量请求，内层对应每个候选生成），日常取 generations[0][0] 就是那个 ChatGeneration，它的 .text 属性会把 AIMessage.content 拎出来。
• 每条消息的角色：msg.type 是 'system' / 'human' / 'ai' 这样的字符串，日志里标上它，回放时才知道这段话是谁说的。

3.3 多回调并存：一次挂两个

把 3.1 和 3.2 一起挂上，控制台实时看耗时、文件悄悄留档——这就是 demo 的核心段落：

timing = ConsoleTimingHandler()
file_log = FileLogHandler(Path("callback_run.log"))

result = chain.invoke(
    {"reader": "通勤族", "book": "夜航船"},
    config={"callbacks": [timing, file_log]},
)

常见坑：

• 文件回调记得处理编码和并发。open(..., encoding="utf-8") 在中文环境不是可选项；多线程场景（batch 并发）下每条日志独立 open-write-close 是最朴素的线程安全写法，追求性能再考虑加锁或队列。
• 别在回调里干重活。钩子是同步等在现场的：你在 on_llm_new_token 里 sleep 一秒，流式输出就慢一秒。推送监控、写库这类操作，回调里只扔进队列，真正干活放到别处。
• on_llm_new_token 只在流式时逐块触发。invoke 一次性等完整结果，通常不吐 token 事件——想统计 token 得走 stream，或者到 on_llm_end 里读统计字段。

4. 钩子速查表：一网打尽

前文按需零散用了不少钩子，这里给一张全表。四组钩子按「被观测的对象」分类，全部都带 run_id / parent_run_id（下表省略）：

几个补充说明：

• 链钩子的覆盖面最广。模型、工具、检索器在框架里也都是 Runnable，你自定义的 RunnableLambda 更是——它们至少会触发 on_chain_start / on_chain_end。也就是说：一个只覆写了链钩子的 handler，挂在任何链上都有事件可看，这是最小可行的观测方案。
• 错误钩子是「断在哪一环」的答案。链上任何一环抛错，那一环的 on_xxx_error 会先响，然后异常继续向上传播（一路上每层的 error 钩子依次响）。04 章的 with_fallbacks 换备胎时，主链的 error 事件照样会被回调看见——失败和回退的全过程都在监控里（demo 第 3 段演示了这个组合）。
• Agent 场景四组全用得上：on_chat_model_start 看模型每次思考的输入、on_agent_action 看它决定调哪个工具、on_tool_start/end 看工具拿了什么返回什么、on_agent_finish 看最终交付——一章 Agent 的执行轨迹就拼出来了。

5. 调试技巧：三件套

本章的观测工具凑成三件：自定义回调（第 3 节已讲，自己写探头）、set_debug 全局开关、astream_events 事件流。后两件是框架现成的，本节上。定位一句话：set_debug 是全店大喇叭，astream_events 是跟拍镜头，自定义回调是常驻探头。

5.1 set_debug(True)：一行代码打开全量日志

from langchain_core.globals import set_debug

set_debug(True)          # 开
result = chain.invoke({"book": "夜航船"})
set_debug(False)         # 用完关掉，别让它一直刷

打开后每次调用都会打印完整的事件流——每个 run 的进出、输入输出的 JSON、耗时，控制台直接变成监控室：

[chain/start] [chain:RunnableSequence] Entering Chain run with input:
{
  "book": "夜航船"
}
[chain/start] [chain:RunnableSequence > prompt:ChatPromptTemplate] Entering Prompt run with input:
...
[llm/start] [chain:RunnableSequence > llm:ChatOllama] Entering LLM run with input:
{
  "prompts": [
    "Human: 用一句话推荐《夜航船》"
  ]
}
[llm/end] [chain:RunnableSequence > llm:ChatOllama] [1247ms] Exiting LLM run with output:
...

方括号里的 chain:RunnableSequence > prompt:ChatPromptTemplate 就是执行树的「路径表示法」——从根到当前节点；[1247ms] 直接告诉你模型这一环花了多久。

两个历史遗留问题说清楚（老教程和新版本对不上号的根源）：

• verbose 基本退休了。老版本里链和模型构造参数有 verbose=True，会打印「Entering new chain...」。1.x 里 LCEL 链上它几乎不再产生输出（它服务的旧版 Chain 类已整体移除）——别在它身上浪费时间，直接用 set_debug。
• 环境变量 LANGCHAIN_DEBUG / LANGCHAIN_VERBOSE 在 1.x 已被收回。老教程里 export LANGCHAIN_DEBUG=true 的写法现在不生效——框架不再自动读这俩变量，开关收编为代码里的 set_debug() / set_verbose()（从 langchain_core.globals 导入）。从 JS 版迁移要特别注意：JS 版传环境变量，Python 1.x 传函数调用。

5.2 astream_events：事件流方式看链内部

回调要写一个类，astream_events 是零成本的观测方式——不用写任何 handler，直接把链的执行过程当成一个事件流来消费：

import asyncio


async def main() -> None:
    async for ev in chain.astream_events({"book": "夜航船"}, version="v2"):
        print(f"{ev['event']:24s} {ev['name']}")


asyncio.run(main())

每个事件 ev 是一个字典，关键字段：event（事件名）、name（组件名）、run_id、data（输入输出或流式块）、tags、metadata、parent_ids（整条祖先链——执行树直接给你排好了）。运行输出长这样：

on_chain_start          RunnableSequence
on_prompt_start         ChatPromptTemplate
on_prompt_end           ChatPromptTemplate
on_chat_model_start     ChatOllama
on_chat_model_stream    ChatOllama
on_parser_start         StrOutputParser
on_parser_stream        StrOutputParser
on_chain_stream         RunnableSequence
on_chat_model_end       ChatOllama
on_parser_end           StrOutputParser
on_chain_end            RunnableSequence

事件名比回调钩子更细一层：提示词环节有专属的 on_prompt_start / on_prompt_end，解析器有 on_parser_start / on_parser_end，流式有 on_chat_model_stream / on_parser_stream——链上每种组件的进出都有专属事件。

只想盯某个环节？用过滤参数，别的事件直接不产生：

async for ev in chain.astream_events(
    {"book": "夜航船"},
    version="v2",
    include_names=["ChatOllama"],      # 只看名字叫 ChatOllama 的组件
):
    print(ev["event"], ev["data"])

include_types（按类型过滤，如 ["chat_model"]）、include_tags、以及对应的 exclude_* 一家人，用法相同。

常见坑：

• version="v2" 参数别漏。astream_events 有 v1/v2 事件格式之分，v1 是历史包袱，不传 version 会收到弃用警告。
• 它是异步接口。得在 async def 里 async for，脚本要 asyncio.run(main())；同步代码里想看事件流，用 5.1 的 set_debug。
• ev["data"] 的内容随事件类型变：on_chain_start 里是 {"input": ...}，on_chain_end 里是 {"output": ...}，on_chat_model_stream 里是 {"chunk": ...}——取值前先看事件名，别硬取。

5.3 调试工具选择地图

6. 全课程收官：一张地图，一条主线，几条岔路

回调是「旁观一切」的机制，用它收官正好——现在你站在监控室里，回头看这 12 章搭起来的整条流水线。

6.1 12 章知识地图

6.2 主线串讲：从一句话到一个店员

回头看，这 12 章其实是在回答一个不断升级的问题。

第 01 章，问题只是「能不能说上话」——第一次 invoke，等来 qwen3:4b 的一句介绍，你看懂了 AIMessage 里除了正文还有一坨元数据。第 02 章马上发现提示词开始重复：欢迎语、荐书规则、语气要求——于是模板和消息接管了「怎么说话」。

第 03 章是把乐高摊开的章节：提示词、模型、解析器都是 Runnable，| 一接就是一条流水线。第 04 章解决体验——流式让用户一秒见字，并发让书单不用排队，fallback/retry 给链上保险。第 05 章解决出口：模型吐的字符串被拆箱成 str / dict / 对象，业务代码终于拿到能直接用的东西。

第 06-08 章是一整段「给店员喂私货」的旅程：本地文件变成 Document、切成小块、算成向量、存进向量库、按问题检索、塞进提示词——RAG 让店员答得上「我们店的借阅规则」，而不是靠模型的出厂记忆瞎编。

第 09 章给店员装了记性——多轮对话里它记得你上一句说了什么；第 10 章给了它手脚——查库存、算折扣这些事交给真代码去干，模型只负责决定和解释；第 11 章松开了缰绳——Agent 自己决定这一轮先查库存还是先写荐语。

然后就是本章：店员上岗了，你回到监控室。计时回调盯着每一环的耗时，文件日志记着每一次模型输入输出和工具调用，astream_events 随时可以跟拍任何一环。从「说上一句话」到「一个可观测的员工」，这就是全部的主线。

6.3 我们没讲什么，以及接下来去哪

诚实地列一下这门课没覆盖的（也是你接下来可以去的方向）：

1. LangGraph——把「循环决策」升级成「图」。11 章的 Agent 本质是一个「想-做-看-再想」的循环；当流程复杂到需要多角色协作、条件分支、人工审批、断点续跑，就该把循环画成显式的状态图。LangChain 官方当前的发力重心也在这里。
2. LangSmith——把本章的监控室换成指挥中心。回调/set_debug 是往控制台倾倒日志；LangSmith 把同样的 trace 信息（就是本章那棵执行树）送到网页上可视化，外加数据集评测、回归测试、团队共享。开启方式就是一个环境变量：LANGCHAIN_TRACING_V2=true。
3. 真实项目：换掉内存向量库（FAISS / Chroma / 云托管）、把链包进 FastAPI 上线、给 RAG 加重排（rerank）、做输出评测、管成本和 token 预算——工程世界的下半场，从你自己的项目开始。
4. 本课程刻意绕开的：多模态（图片/音频）、语音、微调、各家云端模型的差异细节——都不难，等你需要时它们自然会进入射程。

最后一条建议：别把这 12 章当「学完了」，当成「工具箱满了」。挑一个你身边真实的场景——你的笔记、你的文档、你的业务数据——用 RAG 加几个工具搭一个小 Agent，再用本章的回调观察它。第一次看着自己的 Agent 在监控室里一步步决策、调用、交付时，你会真正感觉到：这些东西是你的了。

7. 本章小结

一张表带走本章：

配套 demo（在本章 demo/ 目录）：

• demo_callback.py：两个自定义回调（控制台计时 + 文件日志）挂到荐书链上流式跑、工具钩子、错误与 with_fallbacks 回退的全程观测、astream_events 事件流、set_debug 截选展示——Ollama 离线也能完整跑（自动换内置假模型，钩子触发一致），运行产物 callback_run.log 当场可查。

全课程完 · 返回 00 · 使用指南

收官寄语——

还记得第 01 章的那个下午吗？五行代码，model.invoke("用一句话向新会员介绍「拾光书屋」。")，然后对着屏幕等 qwen3:4b 慢吞吞吐出第一句话。那时候「AI 店员」四个字听起来像个遥远的工程。

十二个章节走完，你已经亲手做过这些事：用模板和消息规范它的谈吐，用 LCEL 把它拼成流水线，用流式和并发守住体验底线，用解析器把它的话拆成数据，用加载器、切割器、嵌入和向量库把书屋的私货喂给它，用 RAG 让它答得上有据，用记忆让它记得住你，用工具和 Agent 让它自己动起来——而今天，你给这一切装上了探头。

更重要的是，这十二个章节教你的从来不是「背 API」：每次遇到问题——等太久、格式乱、答非所问、记不住、不会动手——你学会的都是先看清机制，再挑对工具。这个习惯比任何一行的代码都值钱。

拾光书屋打烊了，灯还亮着。带着你的工具箱，去做点真东西吧。下一本书，你自己写。