跳到主要内容

引擎与配置: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」。这样 PromptTaskRagEngine 里的组件都能留空默认值,跑起来自动取全局默认。

解决什么问题: 假设你要做一个「文档问答」功能。手写的话要自己管:把问题向量化、去向量库检索、按 token 预算裁剪命中片段、拼系统提示、调 LLM、解析输出。RagEngine 把这一整条流水线封装成用例,每一步都可换零件。而 Defaults 解决另一件烦事:几十个组件都要一个 prompt_driver,总不能每个都手传——设一次全局默认,全都自动继承

Engine 和 Task 什么关系(关键边界): Engine 不自己调度 Task。它就是个「功能函数包」。真正把 Engine 接入任务图调度的是 RagTask,它内部持有一个 RagEngine 并在 try_run 里调用它(griptape/tasks/rag_task.py:11-18RagTask.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 / processgriptape/engines/rag/rag_engine.py:14
BaseExtractionEngine(CSV/JSON 子类)从文本抽结构化数据extract_artifactsgriptape/engines/extraction/base_extraction_engine.py:18
BaseSummaryEnginePromptSummaryEngine长文本摘要summarize_artifactsgriptape/engines/summary/base_summary_engine.py:15
EvalEngine用 LLM 给输出打分评估(评估方法)griptape/engines/eval/eval_engine.py:34

后三者本章只做一句话定位:它们都比 RAG 简单——通常就是「拿 prompt_driver + 一套 Jinja 模板,把文本喂给 LLM 拿结果」。例如 PromptSummaryEngine 的核心就是 prompt_driverengines/summary/system.j2 / user.j2 两个模板(griptape/engines/summary/prompt_summary_engine.py:20-27)。它们没有 RAG 那种多阶段流水线,所以下文只深入 RAG。

2.2 Config 层:全局默认装配

部件职责定义位置
Defaults全局单例,持有 drivers_configlogging_configgriptape/configs/defaults_config.py:33Defaults = _DefaultsConfig()
BaseDriversConfig抽象基类,声明 8 类默认 driver 的槽位 + with 上下文切换griptape/configs/drivers/base_drivers_config.py:25
DriversConfig基础实现,默认全是 Dummy* 占位 drivergriptape/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_drivergriptape/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-47process):

# 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):

字段谁写装什么
queryQuery 阶段用户问题(可被改写,如翻译后)
text_chunksRetrieval 阶段检索到的文本片段(TextArtifact 列表)
outputsResponse 阶段最终答案(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 性能的关键:

阶段执行方式依据
① QueryQueryRagStage串行(一个改完交下一个)query_rag_stage.py:25-29[qm.run(context) for qm in ...]
② RetrievalRetrievalRagStage并行(多路检索同时发)retrieval_rag_stage.py:36-42futures_executor.submit(...)
③ ResponseResponseRagStage并行response_rag_stage.py:35-40

为什么 Query 串行?因为多个 query 模块是接力改写同一个 query(前一个的输出是后一个的输入),并行没意义。而检索是多路独立捞(可能同时查多个向量库/loader),并行能省时间。

所有阶段共享抽象基类 BaseRagStage,它要求实现 runmodules 两个方法,并混入 FuturesExecutorMixin 提供并行能力(griptape/engines/rag/stages/base_rag_stage.py:12-20)。

3.5 Retrieval 阶段的精华:合并 → 去重 → 重排 → 截断

检索阶段做的不止「捞」,run 里有一条完整的后处理链(griptape/engines/rag/stages/retrieval_rag_stage.py:36-68run):

多路检索结果(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:49results = list({str(c.value): c for c in results}.values()))。

3.6 可插拔模块:换一步做法只需换一个 module

每个阶段的具体行为由模块决定。模块都继承 BaseRagModulegriptape/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):

阶段模块干什么
QueryTranslateQueryRagModule把问题翻译成指定语言
RetrievalVectorStoreRetrievalRagModule从向量库查相似片段
RetrievalTextLoaderRetrievalRagModule现加载文本再检索
Retrieval(重排)TextChunksRerankRagModule用 rerank driver 重排片段
ResponsePromptResponseRagModule带片段问 LLM 生成答案
ResponseFootnotePromptResponseRagModule生成带脚注引用的答案
ResponseTextChunksResponseRagModule直接返回片段(不调 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-55run):

# 示意,非源码(提炼自 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-66default_generate_system_template)。它的 prompt_driver 同样默认取自 Defaults:23)。


4. 核心原理之二:Config 层——Defaults 单例与一站式装配

上面反复出现「默认取自 Defaults」,这一节讲清它是什么、怎么装配。

4.1 它要解决的小问题

Griptape 里几十个组件都需要一个 prompt_driverPromptTask、每个 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,全进程共享一份,改一处处处生效。

4.3 组件怎么取默认:以 PromptTask 为例

「留空自动继承」的标准写法——字段默认值是一个 lambda,去 Defaults 取(griptape/tasks/prompt_task.py:52-54prompt_driver):

# griptape/tasks/prompt_task.py:52-54,节选真实源码
prompt_driver: BasePromptDriver = field(
default=Factory(lambda: Defaults.drivers_config.prompt_driver), kw_only=True, metadata={"serializable": True}
)

同一套写法遍布全库:PromptResponseRagModule.prompt_driverprompt_response_rag_module.py:23)、VectorStoreRetrievalRagModule.vector_store_drivervector_store_retrieval_rag_module.py:21-23)、BaseExtractionEngine.prompt_driverbase_extraction_engine.py:21-23)、EvalEngine.prompt_drivereval_engine.py:42)……全都是「不传就取 Defaults.drivers_config.<driver>」。

4.4 DriversConfig:一站式设定 8 类默认 driver

BaseDriversConfig 声明了 8 类 driver 的槽位griptape/configs/drivers/base_drivers_config.py:26-49):prompt、image_generation、embedding、vector_store、conversation_memory、text_to_speech、audio_transcription、ruleset。每类都有一个抽象 lazy_property 让子类填(:70-108)。

各 provider 的子类填满这 8 格就成了「换一个类 = 全库默认切到该 provider」。看 OpenAI 那套(griptape/configs/drivers/openai_drivers_config.py:14-37):

# griptape/configs/drivers/openai_drivers_config.py:14-29,节选真实源码
@define
class OpenAiDriversConfig(DriversConfig):
@lazy_property()
def prompt_driver(self) -> OpenAiChatPromptDriver:
return OpenAiChatPromptDriver(model="gpt-4.1")
@lazy_property()
def embedding_driver(self) -> OpenAiEmbeddingDriver:
return OpenAiEmbeddingDriver(model="text-embedding-3-small")
@lazy_property()
def vector_store_driver(self) -> LocalVectorStoreDriver:
return LocalVectorStoreDriver(embedding_driver=OpenAiEmbeddingDriver(model="text-embedding-3-small"))
# …其余 5 类同理

现成的 provider 配置(griptape/configs/drivers/ 目录):

配置类provider默认 prompt 模型
OpenAiDriversConfigOpenAI(也是全局兜底默认)gpt-4.1
AnthropicDriversConfigAnthropic见对应文件
AmazonBedrockDriversConfigAWS Bedrock见对应文件
GoogleDriversConfigGoogle见对应文件
CohereDriversConfigCohere见对应文件
AzureOpenAiDriversConfigAzure OpenAI见对应文件
DriversConfig(基础)无——全是 Dummy* 占位DummyPromptDriverdrivers_config.py:31-33

DriversConfig 是「基础实现」,八类全填 Dummy* 占位 driver(griptape/configs/drivers/drivers_config.py:30-45)——一旦真被调用就抛异常,提醒你「这里必须配一个真 driver」。所有 provider 配置都继承它,只覆盖需要的格子。

4.5 用 with 临时切换默认

BaseDriversConfig 实现了上下文管理器:with SomeDriversConfig(): 块内把 Defaults.drivers_config 换成它,出块自动还原(griptape/configs/drivers/base_drivers_config.py:53-68__enter__ / __exit__):

# 示意,非源码(用法演示)
with AnthropicDriversConfig(): # 这块里默认 provider = Anthropic
agent = Agent() # agent 的 prompt_driver 自动取 Anthropic
# 出块后 Defaults 还原成原来的(默认 OpenAI)

原理:__enter__ 把当前 Defaults.drivers_config 存进 _last_drivers_config 再替换,__exit__ 还原(:53-68)。这让「局部切 provider」不污染全局。


5. 巧妙之处(可借鉴的技术)

  • 状态袋 + 就地改写,替代层层传参。 RagContext 让三阶段解耦——每个模块只认识 context,不需要知道上下游是谁(rag_context.py:16-33)。加一个阶段/模块不改其他人的签名。
  • 模块名隔离运行时参数。 module_configsself.name 分格读写,多模块共享一个 context 却互不串扰(base_rag_module.py:32-38)。
  • 按内容去重的字典惯用法。{str(c.value): c}.values() 一行完成「同文本片段去重」(retrieval_rag_stage.py:49)。
  • token 预算贪心填充。 回答模块边塞片段边算 token、留出答案余量、超了就撤回并停止——朴素但有效地防上下文溢出(prompt_response_rag_module.py:36-49)。
  • 单例 + 懒加载 + 工厂默认值三件套。 Defaults 单例存全局默认,组件字段用 Factory(lambda: Defaults...) 延迟取值——既省去处处手传,又允许任意处覆盖(defaults_config.py:33prompt_task.py:52-54)。
  • 上下文管理器做作用域切换。 with XxxDriversConfig(): 临时换 provider、出块还原,比全局赋值安全(base_drivers_config.py:53-68)。

6. 边界与局限

  • Engine 不做调度。 Engine 只是功能包,不管重试、不管任务依赖、不进任务图。要接入 Structure 得靠 RagTask 这类包装(rag_task.py:11-18)。
  • 三段流水线是 RAG 专属。 抽取/摘要/评估 Engine 没有 Stage/Module 那套可插拔流水线,就是「prompt_driver + Jinja 模板」的直调,别指望在它们身上换阶段。
  • 默认 provider 是 OpenAI,且需要真实凭据。 Defaults.drivers_config 兜底给 OpenAiDriversConfigdefaults_config.py:26-30);不配 OpenAI key 又不换 config,一调就失败。基础 DriversConfig 全是 Dummy*,直接用会抛异常。
  • Defaults 是全局可变单例。 直接赋值 Defaults.drivers_config = ... 会影响全进程;并发或多配置场景下优先用 with 作用域切换,避免互相覆盖。
  • 本章不深入单个 driver 实现。 embedding/vector store 等具体 driver 怎么工作见驱动与 provider 中立

7. 横向对比

同 shelf 的 agent 框架在「用例封装 + 全局默认」上取舍不同(详见各子库 doc):Griptape 的特点是把 RAG 显式拆成三阶段可插拔流水线(多数框架把 RAG 塞进一个黑盒函数或一条 chain),以及用一个单例 Defaults 统一 provider 装配(相比之下 LangChain 等更依赖每处显式传 model)。这种「阶段+模块」结构换来的是可替换性,代价是配置更啰嗦。


8. 代码地图(导航索引)

主题文件路径符号名
RAG 引擎、三段顺序执行griptape/engines/rag/rag_engine.pyRagEngine / process / process_query / __attrs_post_init__
贯穿状态袋griptape/engines/rag/rag_context.pyRagContext / get_references
阶段抽象基类griptape/engines/rag/stages/base_rag_stage.pyBaseRagStage
Query 阶段(串行)griptape/engines/rag/stages/query_rag_stage.pyQueryRagStage
Retrieval 阶段(并行+去重+重排+截断)griptape/engines/rag/stages/retrieval_rag_stage.pyRetrievalRagStage
Response 阶段(并行)griptape/engines/rag/stages/response_rag_stage.pyResponseRagStage
模块基类、名唯一、参数隔离griptape/engines/rag/modules/base_rag_module.pyBaseRagModule / get_context_param / set_context_param
向量检索模块griptape/engines/rag/modules/retrieval/vector_store_retrieval_rag_module.pyVectorStoreRetrievalRagModule
翻译 query 模块griptape/engines/rag/modules/query/translate_query_rag_module.pyTranslateQueryRagModule
重排模块griptape/engines/rag/modules/retrieval/text_chunks_rerank_rag_module.pyTextChunksRerankRagModule
回答模块、token 预算填充griptape/engines/rag/modules/response/prompt_response_rag_module.pyPromptResponseRagModule / default_generate_system_template
Engine 接入任务图griptape/tasks/rag_task.pyRagTask / try_run
抽取/摘要/评估 Enginegriptape/engines/{extraction,summary,eval}/BaseExtractionEngine / PromptSummaryEngine / EvalEngine
全局默认单例griptape/configs/defaults_config.py_DefaultsConfig / Defaults
配置基类griptape/configs/base_config.pyBaseConfig
驱动配置基类、with 切换griptape/configs/drivers/base_drivers_config.pyBaseDriversConfig / __enter__ / __exit__
基础(Dummy)驱动配置griptape/configs/drivers/drivers_config.pyDriversConfig
各 provider 一站式配置griptape/configs/drivers/openai_drivers_config.pyOpenAiDriversConfig / AnthropicDriversConfig / …
组件取默认 driver 的写法griptape/tasks/prompt_task.pyPromptTask.prompt_driver:52-54