从请求到 agent:入口、StreamProcessor、BaseAgent 与四种 agent
30 秒导读: 用户在 DocsGPT 里问一句话,后端要把这句话变成一个正在流式吐字的「agent」。 这一章追一条主线:HTTP 请求 →
StreamProcessor组装配置与依赖 → 工厂挑一个 agent 类 →BaseAgent拼好消息、发给 LLM。终点落在两个最简 agent 的对比上——ClassicAgent(先检索 再问模型)和AgenticAgent(让模型自己决定要不要检索),它们代表了 RAG 的两种基本范式。
本章只讲「一次对话请求如何变成一个 agent 并开始生成」。工具循环怎么多轮转、怎么暂停续跑,是 第 2 章;检索内部怎么改写查询、怎么路由到多个检索器,是 第 4 章;这里都只点到边界,不展开。
1. 这是什么(零基础也能懂)
一句话定义: DocsGPT 后端有一个 /stream HTTP 接口,你 POST 一个问题进去,它用
Server-Sent Events(SSE,服务器持续往下推的事件流)一段段把答案吐回来。这条接口背后真正干活
的东西,叫 agent。
那 agent 是什么?把它想成「一个配好了大模型、工具、检索器和历史记忆的可执行对象」。
你的问题送进来时它还不存在——系统要先读一堆配置(用哪个模型?挂了哪些知识源?这是新会话还是
接着上一轮工具调用往下走?),据此现场组装出一个 agent,然后调用它的 gen() 开始生成。
为什么要分这么多类 agent? 因为「怎么用知识库」有不同玩法:
- 有的场景应该先把相关文档捞出来塞进 prompt,再让模型基于这些文档回答(稳、快、可控)。
- 有的场景应该给模型一把「搜索」工具,让它自己判断该不该搜、搜什么、搜几次(灵活、能多跳)。
这两种就是本章要讲清楚的 ClassicAgent(预取 RAG)与 AgenticAgent(agentic RAG)。
用起来什么样(一个最小请求):
# 示意,非源码:向 /stream 发一个问题,拿回一串 SSE 事件
curl -N -X POST https://your-docsgpt/api/answer/stream \
-H "Content-Type: application/json" \
-d '{"question": "How do I deploy?", "conversation_id": null}'
# 回来的是一段段 data: {...} 事件:先是文本增量,最后是 sources / tool_calls
一句话直觉: StreamProcessor 像餐厅后厨的「配菜台」——把订单(请求)拆解成模型、工具、
文档、历史这些原料备齐;AgentCreator 是「按菜单选锅」;BaseAgent 是「统一的灶台流程」;
而 ClassicAgent / AgenticAgent 是两道用同一灶台、但备料方式不同的菜。
2. 顶层全景(它大概怎么转)
怎么读这张图: 从上到下是一次请求的生命周期。左边是「入口分叉」,中间是「组装」,右边是 「生成」。注意最上面有两条入口——正常提问和工具续跑,它们汇入不同的组装方法。
POST /api/answer/stream
│
StreamResource.post() stream.py:87
│
┌───────────────────┴────────────────────┐
有 tool_actions? 没有(新提问)
「续跑模式」 「正常模式」
│ │
processor.resume_from_tool_actions() processor.build_agent(question)
stream.py:104 stream.py:139
│ │
│ ┌─────────── StreamProcessor ────────────┐
│ │ initialize(): 配 agent/模型/源/检索/历史/附件 │
│ │ (续跑) 从已保存的 pending state 重建 │
│ │ (正常) 预取文档 + 预取工具,渲染 prompt │
│ └──────────────────┬──────────────────────┘
│ │
└──────────────► AgentCreator.create_agent(type, **kw)
agent_creator.py:20
│
┌─────────────────────┼─────────────────────┐
classic/react agentic / research workflow
ClassicAgent AgenticAgent/… WorkflowAgent
│
BaseAgent 子类
__init__ 装配 llm / llm_handler / tool_executor
│
续跑 → agent.gen_continuation(...) 正常 → agent.gen(query)
│
_gen_inner():_build_messages → _llm_gen → _handle_response
│
SSE 事件流:answer / sources / tool_calls
部件一句话职责:
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
StreamResource.post | HTTP 入口,校验请求,分「正常/续跑」两路 | application/api/answer/routes/stream.py:87 |
StreamProcessor | 把请求数据组装成一个 ready-to-run 的 agent | application/api/answer/services/stream_processor.py:107 |
AgentCreator | 工厂:按 agent_type 字符串挑 agent 类 | application/agents/agent_creator.py:11 |
BaseAgent | 所有 agent 的抽象基类:装配依赖、拼消息、发 LLM | application/agents/base.py:25 |
ClassicAgent | 预取 RAG:先检索塞进 prompt,可选挂 internal_search | application/agents/classic_agent.py:15 |
AgenticAgent | agentic RAG:不预取,把 internal_search 交给 LLM 自主调 | application/agents/agentic_agent.py:15 |
主线走一遍(高层): 请求进 post() → 造一个 StreamProcessor → 分叉:正常提问走
build_agent(),工具续跑走 resume_from_tool_actions() → 两者都终结于
AgentCreator.create_agent() 造出某个 BaseAgent 子类 → route 层把 agent 的生成器包成 SSE
Response 返回。
3. 入口:StreamResource.post 的两条路
这一节讲清楚:同一个接口,为什么有两种进入方式。
post() 拿到 JSON、做完基础校验后,第一件事就是看有没有 tool_actions 字段:
# 真实源码节选,application/api/answer/routes/stream.py:96
if data.get("tool_actions"):
# ---- Continuation mode ----
(agent, messages, tools_dict, pending_tool_calls,
tool_actions, reasoning_content) = processor.resume_from_tool_actions(
data["tool_actions"], data["conversation_id"]
)
...
# ---- Normal mode ----
agent = processor.build_agent(data["question"]) # stream.py:139
两条路是干什么用的:
| 模式 | 触发条件 | 处理器方法 | 语义 |
|---|---|---|---|
| 正常模式 | 请求里没有 tool_actions | build_agent(question) | 一次全新提问:组装 agent,从头生成 |
| 续跑模式 | 请求里带 tool_actions | resume_from_tool_actions(...) | 上一轮 agent 请求执行某个工具、暂停等用户批准;现在带着「批准/拒绝/客户端结果」回来,让 agent 接着跑 |
为什么需要续跑模式? 有些工具动作(比如写文件、发消息)需要用户点头。agent 跑到这里会暂停,
把「当前消息、待执行的工具调用、agent 配置」这些续跑状态存起来,先把「请批准」这个事件流给前端。
用户批准后,前端把决定作为 tool_actions 再打回 /stream——这就是续跑模式的入口。暂停/续跑的
内部机制是第 2 章的主题,这里只需知道入口在此分叉。
两条路造好 agent 后,都会走同一个鉴权与配额检查(decoded_token 为空则 401,check_usage
拦配额),再把 complete_stream(...)(定义在 routes/base.py:177)包成
mimetype="text/event-stream" 的 Response 返回。complete_stream 内部对正常模式调 agent.gen()、
对续跑模式调 agent.gen_continuation()——两者的契约见 §5。
4. StreamProcessor:把请求「配菜」成 agent
这一节讲清楚:从裸 JSON 到一个能跑的 agent,中间到底做了哪些准备。 只讲职责与顺序,不逐行抠。
4.1 组装的入口:build_agent
正常模式的一站式方法是 build_agent,它把四步串成一条:
# 真实源码节选,application/api/answer/services/stream_processor.py:158
def build_agent(self, question: str):
self.initialize() # ① 备齐所有配置
agent_type = self.agent_config.get("agent_type", "classic")
... # ② 按类型决定要不要预取
initialize()(stream_processor.py:149)按固定顺序把「原料」备齐,每一步落到一个实例属性上:
| 步骤 | 方法 | 产出 |
|---|---|---|
| 配 agent | _configure_agent | agent_config(含 agent_type、prompt_id、API key、结构化输出 schema) |
| 定模型 | _validate_and_set_model | model_id + model_user_id(BYOM 归属域) |
| 配数据源 | _configure_source | source / all_sources(挂了哪些知识库) |
| 配检索器 | _configure_retriever | retriever_config(retriever 名、chunks、文档 token 预算) |
| 载历史 | _load_conversation_history | history(从 DB 或请求体,必要时压缩) |
| 处理附件 | _process_attachments | attachments |
备齐后,build_agent 按 agent_type 决定要不要在造 agent 之前先把文档捞出来:
agent_type ∈ {agentic, research}?
有 agentic_tool 源 → 预取 prefetch 子集 + 预取工具 → create_agent(…, agentic_sources)
否则 → 只预取工具(不取文档) → create_agent(tools_data)
否则(classic/react):
有 agentic_tool 源 → 预取 prefetch 子集,agentic_tool 子集交给搜索工具
否则 → 无脚本地全量预取所有 active_docs(今天的经典行为)
→ create_agent(docs_together, docs, tools_data)
这里的关键是 pre_fetch_docs(question)(stream_processor.py:1038):在造 agent 之前就跑一次
检索,把命中的文档格式化进 docs_together,后面渲染 prompt 时塞进去。这正是「预取 RAG」的物理
体现——文档在模型看到问题之前就已经在 prompt 里了。检索本身怎么做,见第 4 章。
4.2 收尾:create_agent 拼依赖、造实例
initialize + 预取跑完,create_agent(stream_processor.py:1510)做最后的组装:
- 渲染 prompt:用
prompt_renderer把预取的docs_together、工具数据、附件填进 prompt 模板 (agentic/research 会换用agentic_*预设,里面是「搜索工具指引」而非文档块——见_get_prompt_content,stream_processor.py:1266)。 - 造 LLM 与 handler:据
model_id解析出 provider,LLMCreator.create_llm(...)造 llm,LLMHandlerCreator.create_handler(...)造对应的流式处理器。 - 造 tool_executor:注入用户身份、agent_id、客户端工具。
- 按类型补 kwargs:agentic/research(或带
agentic_tool源的 classic)会带上一个retriever_config,里面装着 internal_search 工具要用的源列表;workflow 则带workflow_id/图。 - 最后
AgentCreator.create_agent(agent_type, **agent_kwargs)造出实例。
4.3 续跑的入口:resume_from_tool_actions
续跑模式不走 initialize,而是 resume_from_tool_actions(stream_processor.py:1340):从
ContinuationService 领回上次暂停时存下的状态(messages、pending_tool_calls、
tools_dict、agent_config),据此重建 llm / handler / tool_executor,用类名映射回工厂键
(ClassicAgent → classic 等,stream_processor.py:1445),造出 agent,返回一个可直接调
gen_continuation() 的元组。它的职责是「精确复原上一轮的现场」,而不是从头组装。
还有个第三入口
build_continuation_from_messages(stream_processor.py:218):OpenAI 兼容 客户端会把整段对话重发、但不带conversation_id,于是没有服务端状态可领——这时直接从重发的 messages 里定位最后一条带tool_calls的 assistant 消息来重建。属于同一续跑家族的无状态变体。
5. AgentCreator 与 BaseAgent:统一的灶台
5.1 工厂:一张映射表
AgentCreator 极简——一个字典 + 一个查表:
# 真实源码,application/agents/agent_creator.py:12
agents = {
"classic": ClassicAgent,
"react": ClassicAgent, # 向后兼容:react 退化为 classic
"agentic": AgenticAgent,
"research": ResearchAgent,
"workflow": WorkflowAgent,
}
五个键映射到四个类(react 复用 ClassicAgent)。create_agent(type, **kwargs)
(agent_creator.py:20)按小写 type 取类、实例化;取不到就抛 ValueError。这就是「四种
agent」的全部路由逻辑——选择哪一种由上游 agent_config["agent_type"] 决定。
本章聚焦 classic / agentic 两个最简 agent;research 与 workflow 是更重的高级 agent,留给
第 6 章。
5.2 BaseAgent.init:装配三件套
BaseAgent(application/agents/base.py:25)是所有 agent 的抽象基类。它的 __init__
(base.py:26)核心是依赖注入——三个关键组件优先用外部传入的,没传才自己造:
| 组件 | 作用 | 装配处 |
|---|---|---|
self.llm | 与某个模型 provider 对 话的客户端 | base.py:71(传入优先,否则 LLMCreator.create_llm) |
self.llm_handler | 解析流式响应、驱动工具循环的处理器 | base.py:94 |
self.tool_executor | 发现、执行工具,累积 tool_calls | base.py:102 |
StreamProcessor.create_agent 总是把三者造好再注入,所以生产路径走的是「传入优先」分支;
__init__ 里的自造分支是给 worker / 测试等轻量调用兜底。此外还缓存了 upstream_model_id
(base.py:88)——BYOM 场景下注册表 UUID 与上游真实模型名不同,这里存的是发给上游用的那个名。
5.3 gen / _gen_inner 契约
对外只有一个入口 gen,它是模板方法:
# 真实源码,application/agents/base.py:140
@log_activity()
def gen(self, query, log_context=None):
yield from self._gen_inner(query, log_context) # 子类实现真正的生成
yield from self._emit_responses_metadata() # 收尾:吐 Responses 续跑元数据/用量
_gen_inner(base.py:228)是 @abstractmethod——基类不实现,每个子类必须给出自己的生成
流程。这是整个 agent 体系的核心契约:基类提供拼消息、发 LLM、处理响应等公共零件,子类用
_gen_inner 把它们编排成自己的执行流。续跑走的是另一个入口 gen_continuation(base.py:234),
它先把用户对工具的决定拼进 messages,再接回 LLM 循环——属于第 2 章。
5.4 _build_messages:按 token 预算裁历史、拼多模态
_build_messages(base.py:496)把「system prompt + 历史 + 工具往返 + 当前 user 消息」拼成
provider 要的 messages 数组。它不是简单拼接,而是带 token 预算的:
context_limit(该模型上下文上限)
− system_tokens(系统 prompt 占用)
− safety_buffer(≈10% 余量)
= available_after_system
├─ query 最多用 80%,超了就 _truncate_text_middle 掐中间
└─ 剩下的给历史:_truncate_history_to_fit 从最近往前塞,塞不下就停
要点:
- 预算自适应模型:
get_token_limit(self.model_id, ...)拿的是当前模型的上限,BYOM 的小 上下文(8k/32k)不会被按默认 128k 撑爆(base.py:515)。 - 历史从新到旧贪心保留:
_truncate_history_to_fit(base.py:645)reversed(history)逐条 累加 token,超预算即break——优先保住最近的对话。 - 历史里的工具往返也会被回放:一轮里的
tool_calls会被重建成 assistant(带 tool_calls)- 若干 tool 结果消息(
base.py:545-619),并对复用的 call_id 合成稳定的 replay id 防冲突。
- 若干 tool 结果消息(
- 多模态走特殊出口:若本轮是多模态请求,最后的 user 消息用完整的
multimodal_content(文本 + image_url 分块),而纯文本query只用于 token 预算与检索(base.py:637-642)。
5.5 _llm_gen:组装 gen_kwargs
_llm_gen(base.py:691)把 messages 变成一次真正的流式 LLM 调用。它的活是按能力条件地
往 gen_kwargs 里塞参数:
| 参数 | 何时加入 | 源码 |
|---|---|---|
model / messages | 总是;model 用 upstream_model_id | base.py:707 |
tools | llm _supports_tools 且 self.tools 非空 | base.py:711-716 |
response_format / response_schema | 有 json_schema 且模型支持结构化输出(OpenAI 用 response_format,Google 用 response_schema) | base.py:717-729 |
response_format={"type":"json_object"} | 请求要 json_object 模式(保证合法 JSON、不强 schema) | base.py:730-737 |
previous_response_id | 开了 Responses store 且模型走 Responses API,且上一轮留下了同链 id | base.py:738-746 |
| 采样参数(temperature/max_tokens…) | 请求带了 llm_params | base.py:748-749 |
previous_response_id 值得单说:OpenAI 的 Responses API 支持服务端接续对话——不必每次重传
整段历史,只要给上一轮的 response id。_previous_response_id(base.py:174)会校验「链 key」
匹配才复用,保证不会把张三的会话接到李四头上。装配完 gen_kwargs,self.llm.gen_stream(**gen_kwargs)
返回一个响应,交给 _handle_response / llm_handler 去驱动工具循环(第 2 章)。
6. 两种范式:ClassicAgent vs AgenticAgent
这是本章的落点。两个类代码几乎一模一样,差别就一行——而这一行,就是「预取 RAG」和 「agentic RAG」的分水岭。
6.1 ClassicAgent:预取 RAG
ClassicAgent._gen_inner(classic_agent.py:36):
# 真实源码节选,application/agents/classic_agent.py:41
tools_dict = self.tool_executor.get_tools()
if self.retriever_config: # 仅当有 agentic_tool 源才挂
add_internal_search_tool(tools_dict, self.retriever_config)
...
messages = self._build_messages(self.prompt, query) # prompt 里已含预取文档
llm_response = self._llm_gen(messages, log_context)
...
if self.retriever_config:
self._collect_internal_sources() # 合并搜索工具捞到的源
关键点:文档在 StreamProcessor.pre_fetch_docs 阶段就检索好、渲染进了 self.prompt,所以
模型第一次看到问题时,相关文档已经在上下文里了。internal_search 工具默认不挂——只有当
某些源被标成 agentic_tool 曝光度、create_agent 才会传 retriever_config,此时 classic 变成
「预取一部分 + 让模型按需搜另一部分」的混合体。默认单源/无配置时 retriever_config 为空,行为与
最朴素的预取 RAG 逐字节一致(见类 docstring,classic_agent.py:15)。
6.2 AgenticAgent:agentic RAG
AgenticAgent._gen_inner(agentic_agent.py:34):
# 真实源码节选,application/agents/agentic_agent.py:37
tools_dict = self.tool_executor.get_tools()
add_internal_search_tool(tools_dict, self.retriever_config) # 无条件挂搜索工具
...
# prompt 里【没有】预取文档
messages = self._build_messages(self.prompt, query)
llm_response = self._llm_gen(messages, log_context) # handler 管工具循环
...
self._collect_internal_sources()
关键点:没有 if——internal_search 工具无条件加入,交给 LLM。prompt 里不含预取文档
(StreamProcessor.build_agent 对 agentic/research 不做全量预取)。于是模型自己决定:要不要搜、
搜什么关键词、搜几轮。检索的控制权从代码手里,交到了模型手里。
6.3 一图看清差异
Classic(预取 RAG) │ Agentic(agentic RAG)
│
问题 ──► 先检索 ──► 文档进 │ 问题 ──► 直接进 prompt
prompt ──► 问模型 │ (无文档)──► 问模型
│ │
一次检索,发生在提问前 │ 模型: "我需要搜吗?"
模型只管基于给定文档答 │ ├─ 不需要 ──► 直接答
│ └─ 需要 ──► 调 internal_search
稳、快、可控 │ ──► 拿到文档 ──► 可再搜 ──► 答
│ 多跳、自适应、能处理复杂问题
| 维度 | ClassicAgent | AgenticAgent |
|---|---|---|
| 检索时机 | 提问前,由 StreamProcessor 预取 | 生成中,由 LLM 调工具 |
| 检索次数 | 一次(固定) | 0 到多次(模型自定) |
internal_search 工具 | 默认不挂,仅混合曝光时挂 | 无条件挂 |
| 谁控制检索 | 代码/流水线 | 大模型 |
| 适合 | 单跳问答、可控可预测 | 多跳、需要判断「要不要查」的复杂问题 |
| 代价 | 可能取回无关文档;不会追问 | 更多 LLM 往返、更慢、更贵 |
6.4 共同的收尾:_collect_internal_sources 去重合并
两个类都有一个字节相同的 _collect_internal_sources(classic_agent.py:65 /
agentic_agent.py:63)。它解决一个具体问题:引用来源要既包含预取的文档、也包含工具搜到的
文档,还不能重复。
# 真实源码节选,application/agents/classic_agent.py:74
def _key(d): # 用 (source,title,text) 做指纹
if isinstance(d, dict):
return (d.get("source"), d.get("title"), d.get("text"))
return id(d)
merged = list(self.retrieved_docs or []) # 先保留已有(预取)文档
seen = {_key(d) for d in merged}
for doc in tool.retrieved_docs: # 再并入工具捞到的、跳过重复
if _key(doc) not in seen:
seen.add(_key(doc)); merged.append(doc)
self.retrieved_docs = merged
它从 tool_executor._loaded_tools 里按缓存键取回那个 InternalSearchTool 实例
(classic_agent.py:69),把它的 retrieved_docs 去重并入 agent 的 retrieved_docs——最终作为
{"sources": ...} 事件吐给前端。先放已有、再并新的,保证混合曝光的 agent 两边来源都能被引用。
7. 边界与本章不讲的
- 工具循环内部:LLM 吐出 tool_call 后怎么执行、怎么多轮、怎么暂停等审批再续跑——
llm_handler的process_message_flow(application/llm/handlers/base.py:125)与gen_continuation,是 第 2 章。 - 检索内部:
pre_fetch_docs/internal_search背后的查询改写、按源路由的 Dispatcher、多 检索器与向量库,是第 4 章。 - LLM 抽象:
LLMCreator、provider 无关、BYOM、跨模型回退、结构化输出的底层,是 第 5 章。 - research / workflow:两个高级 agent 的展开,是第 6 章。
- 本章看到的
ClassicAgent「无retriever_config即退化为纯预取」是刻意的向后兼容设计;混合 曝光(prefetch + agentic_tool)是较新的能力,默认单源请求不会触发。
8. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| HTTP 入口、正常/续跑分叉 | application/api/answer/routes/stream.py | StreamResource.post |
| SSE 响应包装、正常 vs 续跑调用 | application/api/answer/routes/base.py | BaseAnswerResource.complete_stream |
| 请求→agent 组装(正常) | application/api/answer/services/stream_processor.py | StreamProcessor.build_agent / initialize |
| 造 agent、拼 llm/tool_executor、渲染 prompt | application/api/answer/services/stream_processor.py | StreamProcessor.create_agent |
| 续跑:从保存状态重建 agent | application/api/answer/services/stream_processor.py | StreamProcessor.resume_from_tool_actions |
| 无状态续跑(OpenAI 兼容重发) | application/api/answer/services/stream_processor.py | StreamProcessor.build_continuation_from_messages |
| 预取文档 | application/api/answer/services/stream_processor.py | StreamProcessor.pre_fetch_docs |
| agent 工厂映射表 | application/agents/agent_creator.py | AgentCreator.agents / create_agent |
| 抽象基类、依赖装配 | application/agents/base.py | BaseAgent.__init__ |
| 对外生成契约 | application/agents/base.py | BaseAgent.gen / _gen_inner(abstract) |
| 续跑生成 | application/agents/base.py | BaseAgent.gen_continuation |
| 按 token 预算拼消息、裁历史、多模态 | application/agents/base.py | BaseAgent._build_messages / _truncate_history_to_fit |
| 组装 gen_kwargs(tools/结构化输出/Responses id) | application/agents/base.py | BaseAgent._llm_gen |
| 预取 RAG agent | application/agents/classic_agent.py | ClassicAgent._gen_inner / _collect_internal_sources |
| agentic RAG agent | application/agents/agentic_agent.py | AgenticAgent._gen_inner |