引擎与配置:RAG 等用例封装 + Defaults 全局装配
30 秒导读: 前面几章讲的都是「零件」——结构与任务图、PromptTask 循环、Driver、工具、记忆与制品。这一章讲顶层的两块拼装板:Engine 把 Driver 包成「一个用例」(如 RAG 问答、信息抽取、摘要、评估),Config 用一个全局单例
Defaults把「该用哪些 driver」一次性设好,让所有零件不用手填也能跑。
1. 这是什么(零基础也能懂)
一句话定义: 本章讲两层「装配」——
- Engine(引擎) = 把底层 Driver 打包成一个完整用例的胶水层。你不用自己拼「查向量库 → 塞进 prompt → 调模型」,
RagEngine一行process_query("问题")就把整套 RAG(Retrieval-Augmented Generation,检索增强 生成:先检索资料、再让模型带着资料回答)跑完。 - Config(配置) = 一个叫
Defaults的全局单例,规定「没指定时默认用哪个 prompt/embedding/vector store driver」。这样PromptTask、RagEngine里的组件都能留空默认值,跑起来自动取全局默认。
解决什么问题: 假设你要做一个「文档问答」功能。手写的话要自己管:把问题向量化、去向量库检索、按 token 预算裁剪命中片段、拼系统提示、调 LLM、解析输出。RagEngine 把这一整条流水线封装成用例,每一步都可换零件。而 Defaults 解决另一件烦事:几十个组件都要一个 prompt_driver,总不能每个都手传——设一次全局默认,全都自动继承。
Engine 和 Task 什么关系(关键边界): Engine 不自己调度 Task。它就是个「功能函数包」。真正把 Engine 接入任务图调度的是 RagTask,它内部持有一个 RagEngine 并在 try_run 里调用它(griptape/tasks/rag_task.py:11-18,RagTask.rag_engine / try_run):
# griptape/tasks/rag_task.py:14-18,节选真实源码
def try_run(self) -> ListArtifact | ErrorArtifact:
outputs = self.rag_engine.process_query(self.input.to_text()).outputs
if len(outputs) > 0:
return ListArtifact(outputs)
return ErrorArtifact("empty output")
一句话记住这层关系:
Structure(Agent/Pipeline/Workflow) ← 调度
└── Task(RagTask 等) ← 被调度的一步
└── Engine(RagEngine) ← 用例功能 包(本章)
└── Driver(prompt/vector/embedding…) ← provider 中立零件(第3章)
用起来什么样: 一个最小的可运行 RAG(改自 tests/unit/engines/rag/test_rag_engine.py:14-25 的 fixture):
# 示意,非源码(结构与真实 fixture 一致)
engine = RagEngine(
retrieval_stage=RetrievalRagStage( # 第二段:检索
retrieval_modules=[VectorStoreRetrievalRagModule()], # 从向量库捞片段
),
response_stage=ResponseRagStage( # 第三段:回答
response_modules=[PromptResponseRagModule()], # 带着片段问 LLM
),
)
result = engine.process_query("griptape 是什么?") # 一行跑完整条流水线
print(result.outputs[0].value) # 模型的回答
一句话直觉: 把 Engine 想成「预制菜」——Driver 是生食材,Engine 把它们炒成一道能直接上桌的菜;Defaults 则是厨房的「默认调料架」,不特别指定就用架上那套。
2. 顶层全景(Engine 层与 Config 层怎么转)
本章两条主线,先看它们各自的部件与职责。
2.1 Engine 家族:四种用例封装
Griptape 提供四类 Engine,都遵循同一个定位——包住 Driver,暴露一个用例方法:
| Engine | 用例(做什么) | 入口方法 | 定义位置 |
|---|---|---|---|
RagEngine | 检索增强问答(本章重点范例) | process_query / process | griptape/engines/rag/rag_engine.py:14 |
BaseExtractionEngine(CSV/JSON 子类) | 从文本抽结构化数据 | extract_artifacts | griptape/engines/extraction/base_extraction_engine.py:18 |
BaseSummaryEngine(PromptSummaryEngine) | 长文本摘要 | summarize_artifacts | griptape/engines/summary/base_summary_engine.py:15 |
EvalEngine | 用 LLM 给输出打分评估 | (评估方法) | griptape/engines/eval/eval_engine.py:34 |
后三者本章只做一句话定位:它们都比 RAG 简单——通常就是「 拿 prompt_driver + 一套 Jinja 模板,把文本喂给 LLM 拿结果」。例如 PromptSummaryEngine 的核心就是 prompt_driver 加 engines/summary/system.j2 / user.j2 两个模板(griptape/engines/summary/prompt_summary_engine.py:20-27)。它们没有 RAG 那种多阶段流水线,所以下文只深入 RAG。
2.2 Config 层:全局默认装配
| 部件 | 职责 | 定义位置 |
|---|---|---|
Defaults | 全局单例,持有 drivers_config 和 logging_config | griptape/configs/defaults_config.py:33(Defaults = _DefaultsConfig()) |
BaseDriversConfig | 抽象基类,声明 8 类默认 driver 的槽位 + with 上下文切换 | griptape/configs/drivers/base_drivers_config.py:25 |
DriversConfig | 基础实现,默认全是 Dummy* 占位 driver | griptape/configs/drivers/drivers_config.py:30 |
OpenAiDriversConfig 等 | 各 provider 的一站式默认(OpenAI/Anthropic/Bedrock/Google/Cohere/Azure) | griptape/configs/drivers/openai_drivers_config.py:14 |
两层怎么接上: Engine 里的组件字段默认值就是去 Defaults 取 driver。例如向量检索模块的 vector_store_driver 默认 Defaults.drivers_config.vector_store_driver(griptape/engines/rag/modules/retrieval/vector_store_retrieval_rag_module.py:21-23)。这就是「留空默认、自动继承全局配置」的实现方式。
3. 核心原理之一:RAG Engine 的三阶段流水线
这是本章的主菜。先讲思路,再拆每一阶段。
3.1 它要解决的小问题
「检索增强问答」天然是三步走:把用户问题整理好 → 拿问题去捞资料 → 带着资料生成回答。Griptape 把这三步固化成三个阶段(Stage),每个阶段里塞若干可插拔模块(Module)。你换模块 = 换某一步的具体做法,流水线骨架不动。
3.2 整体流水线(一张图)
怎么读这张图:从上到下是执行顺序;
RagContext(贯穿全程的状态袋)被每一阶段依次改写、传给下一段。方框里第二行是负责的类。
用户问题 query
│
▼
┌─────────────────────────────────────┐
│ ① Query 阶段(整理问题) │ 串行跑各 query 模块
│ QueryRagStage │ 例:翻译成目标语言
│ → 改写 context.query │
└─────────────────────────────────────┘
│ RagContext 往下传
▼
┌─────────────────────────────────────┐
│ ② Retrieval 阶段(捞资料) │ 多个检索模块【并行】跑
│ RetrievalRagStage │ → 合并 → 去重 → 可选重排 → 截断
│ → 写入 context.text_chunks │
└─────────────────────────────────────┘
│ RagContext 往下传
▼
┌─────────────────────────────────────┐
│ ③ Response 阶段(生成回答) │ 各 response 模块并行跑
│ ResponseRagStage │ 带着 text_chunks 问 LLM
│ → 写入 context.outputs │
└─────────────────────────────────────┘
│
▼
context.outputs(答案 Artifact 列表)
RagEngine.process 就是按这三段顺序跑,哪段为 None 就跳过(griptape/engines/rag/rag_engine.py:39-47,process):
# griptape/engines/rag/rag_engine.py:39-47,节选真实源码
def process(self, context: RagContext) -> RagContext:
if self.query_stage:
context = self.query_stage.run(context)
if self.retrieval_stage:
context = self.retrieval_stage.run(context)
if self.response_stage:
context = self.response_stage.run(context)
return context
三段都可选、无一必填——你可以只要检索+回答(最常见),也可以三段全上。
3.3 贯穿全程的状态袋:RagContext
三个阶段之间不靠返回值层层传参,而是共享一个可变对象 RagContext。每个模块在自己的 run 里就地改写它的字段(griptape/engines/rag/rag_context.py:16-33):
| 字段 | 谁写 | 装什么 |
|---|---|---|
query | Query 阶段 | 用户问题(可被改写,如翻译后) |
text_chunks | Retrieval 阶段 | 检索到的文本片段(TextArtifact 列表) |
outputs | Response 阶段 | 最终答案(BaseArtifact 列表) |
module_configs | 任意模块 | 按模块名隔离的运行时参数字典 |
before_query / after_query | 调用方 | 想插在 query 前后的附加串 |
module_configs 是个巧思:每个模块用 get_context_param / set_context_param 只读写以自己 name 为键的那一格,互不干扰(griptape/engines/rag/modules/base_rag_module.py:32-38)。
3.4 三个阶段各自怎么跑(串行 vs 并行)
三段的执行策略不同,这是理解 RAG 性能的关键:
| 阶段 | 类 | 执行方式 | 依据 |
|---|---|---|---|
| ① Query | QueryRagStage | 串行(一个改完交下一个) | query_rag_stage.py:25-29,[qm.run(context) for qm in ...] |
| ② Retrieval | RetrievalRagStage | 并行(多路检索同时发) | retrieval_rag_stage.py:36-42,futures_executor.submit(...) |
| ③ Response | ResponseRagStage | 并行 | response_rag_stage.py:35-40 |
为什么 Query 串行?因为多个 query 模块是接力改写同一个 query(前一个的输出是后一个的输入),并 行没意义。而检索是多路独立捞(可能同时查多个向量库/loader),并行能省时间。
所有阶段共享抽象基类 BaseRagStage,它要求实现 run 和 modules 两个方法,并混入 FuturesExecutorMixin 提供并行能力(griptape/engines/rag/stages/base_rag_stage.py:12-20)。
3.5 Retrieval 阶段的精华:合并 → 去重 → 重排 → 截断
检索阶段做的不止「捞」,run 里有一条完整的后处理链(griptape/engines/rag/stages/retrieval_rag_stage.py:36-68,run):
多路检索结果(list of lists)
│ itertools.chain.from_iterable —— 拍平成一个列表
▼
去重:{str(c.value): c}.values() —— 按片段文本内容去重(:49)
▼
可选 rerank_module —— 用重排 driver 按相关度重新排序(:61-64)
▼
可选 max_chunks —— 只保留前 N 条(:66-67)
▼
写回 context.text_chunks
去重那行是个惯用法:用「片段文本 → 片段对象」建字典,同文本后者覆盖前者,.values() 即得去重结果(retrieval_rag_stage.py:49,results = list({str(c.value): c for c in results}.values()))。
3.6 可插拔模块:换一步做法只需换一个 module
每个阶段的具体行为由模块决定。模块都继承 BaseRagModule(griptape/engines/rag/modules/base_rag_module.py:17),自带唯一 name(默认 类名-uuid,:18-20)——RagEngine 构造时会校验所有模块名唯一,重名直接报错(rag_engine.py:19-34,__attrs_post_init__)。
现成模块一览(griptape/engines/rag/modules/__init__.py:1-14):
| 阶段 | 模块 | 干什么 |
|---|---|---|
| Query | TranslateQueryRagModule | 把问题翻译成指定语言 |
| Retrieval | VectorStoreRetrievalRagModule | 从向量库查相似片段 |
| Retrieval | TextLoaderRetrievalRagModule | 现加载文本再检索 |
| Retrieval(重排) | TextChunksRerankRagModule | 用 rerank driver 重排片段 |
| Response | PromptResponseRagModule | 带片段问 LLM 生成答案 |
| Response | FootnotePromptResponseRagModule | 生成带脚注引用的答案 |
| Response | TextChunksResponseRagModule | 直接返回片段(不调 LLM) |
范例 1 —— 最简单的检索模块。 VectorStoreRetrievalRagModule.run 就一句:把 context.query 丢给向量库 driver,结果转成 Artifact(vector_store_retrieval_rag_module.py:29-31):
# griptape/engines/rag/modules/retrieval/vector_store_retrieval_rag_module.py:29-31,节选真实源码
def run(self, context: RagContext) -> Sequence[TextArtifact]:
query_params = utils.dict_merge(self.query_params, self.get_context_param(context, "query_params"))
return self.process_query_output(self.vector_store_driver.query(context.query, **query_params))
注意它的 vector_store_driver 默认取自 Defaults(:21-23)——这正是 §4 的配置层在起作用。
范例 2 —— 会改写状态的 Query 模块。 TranslateQueryRagModule.run 演示了「就地改写 context.query」(translate_query_rag_module.py:25-31):调 LLM 翻译,然后 context.query = output.to_text(),交给下游用翻译后的问题去检索。
3.7 回答阶段的精华:按 token 预算塞片段
PromptResponseRagModule 是 RAG 的「最后一公里」,它有个不显然的设计——边塞片段边算 token,塞不下就停(griptape/engines/rag/modules/response/prompt_response_rag_module.py:30-55,run):
# 示意,非源码(提炼自 prompt_response_rag_module.py:36-49 的核心逻辑)
included = []
for chunk in context.text_chunks: # 逐个尝试加入命中片段
included.append(chunk)
system_prompt = self.generate_system_template(context, included)
tokens = count_tokens(system_prompt + query)
if tokens + self.answer_token_offset >= max_input_tokens: # 留出答案的余量
included.pop() # 这片放不下,撤回
break # 之后的也别试了
# 用装得下的片段拼最终 prompt 调模型
answer_token_offset(默认 400,:24)是给模型回答预留的 token 余量——不能把上下文塞满到没地方生成答案。片段和规则最终由 Jinja 模板 engines/rag/modules/response/prompt/system.j2 渲染成系统提示(:57-66,default_generate_system_template)。它的 prompt_driver 同样默认取自 Defaults(:23)。
4. 核心原理之二:Config 层——Defaults 单例与一站式装配
上 面反复出现「默认取自 Defaults」,这一节讲清它是什么、怎么装配。
4.1 它要解决的小问题
Griptape 里几十个组件都需要一个 prompt_driver(PromptTask、每个 Engine、每个 Response 模块……)。若每处都手传,既啰嗦又容易漏。解法:一个全局单例存「默认该用谁」,所有组件留空时自动去取。
4.2 Defaults 是个单例
Defaults 是模块加载时就实例化好的唯一实例(griptape/configs/defaults_config.py:33):
# griptape/configs/defaults_config.py:17-33,节选真实源码
@define(kw_only=True)
class _DefaultsConfig(BaseConfig, SingletonMixin):
_logging_config: LoggingConfig | None = field(default=None)
_drivers_config: BaseDriversConfig | None = field(default=None)
@lazy_property()
def drivers_config(self) -> BaseDriversConfig:
from griptape.configs.drivers.openai_drivers_config import OpenAiDriversConfig
return OpenAiDriversConfig() # 全局默认:OpenAI 那套
Defaults = _DefaultsConfig() # 模块级唯一实例
两个要点:
drivers_config是懒加载(lazy_property):第一次访问才建,默认给OpenAiDriversConfig(:26-30)。所以「开箱即用」的默认 provider 是 OpenAI。- 它继承
SingletonMixin,全进程共享一份,改一处处处生效。