驱动与 provider 中立:PromptStack、Message 与统一工具协议

30 秒导读: Griptape 里,上层逻辑(任务、工具、记忆)从不直接碰 OpenAI/Anthropic 的 SDK。它们只跟一套 provider 无关的中间表示打交道——PromptStack(一叠消息)、Message(一条消息)、MessageContent(消息里的一块内容)。真正懂某家 API 的只有一个 BasePromptDriver 子类。于是"换供应商"退化成"换一个 driver 对象",代码里就是配置里的一行。

本章讲清一件事:为什么在 Griptape 里换 LLM 供应商只需要改一行。 我们先讲这套抽象是什么、长什么样,再拆开 driver 基类的模板方法,最后看三个可切换开关和配置层怎么把它们装配起来。

不重复的部分:ReAct 文本协议怎么解析,见 02-prompt-task-agent-loop.md;各家 provider driver 的逐行实现不展开,本章只讲基类契约 + 抽象点。

---

1. 这是什么(零基础也能懂)

1.1 一句话定义

Griptape 的"驱动层"是一层翻译官。 上层代码用一种统一的中间语言描述"我要跟大模型说什么、它回了什么";每家供应商配一个专职翻译官(driver),负责把中间语言翻成该家 API 的方言、再把回复翻回来。

1.2 它解决什么问题

假设你写了一个 agent,用 OpenAI 跑通了。现在老板说:"改用 Anthropic,便宜。" 如果你的代码里到处是 openai.chat.completions.create(...)、到处按 OpenAI 的 JSON 结构拼 messages,那这是一次大手术。

Griptape 的答案:上层根本不认识 OpenAI。它只认识 PromptStack。切供应商时你换掉的是装配好的 driver,不是业务代码。

1.3 用起来什么样

下面这段是真实可跑的用法,注意换供应商只动了一行:

# 示意用法,展示 provider 中立
from griptape.structures import Agent
from griptape.configs import Defaults
from griptape.configs.drivers import AnthropicDriversConfig

# 默认走 OpenAI —— 什么都不配就是它
agent = Agent()
agent.run("讲个冷笑话")

# 想换 Anthropic?改这一行全局默认,业务代码一字不动
Defaults.drivers_config = AnthropicDriversConfig()
agent = Agent()
agent.run("讲个冷笑话")   # 现在走 claude

Agent、run("讲个冷笑话")、工具、记忆——全程没出现任何一家供应商的名字。

1.4 一句话直觉

把 PromptStack 想成通用集装箱:货物(你的提示词、图片、工具调用)按标准尺寸打包。港口(driver)只管把集装箱装上不同船公司(OpenAI/Anthropic/…)的船——货物本身不用重新包装。

---

2. 顶层全景(它大概怎么转)

一次调用的数据流:上层把各种 Artifact 塞进 PromptStack → driver 的 run() 模板跑重试、发事件、调用子类的 try_run → 子类把中间表示翻成 provider 请求、拿回结果再翻回一个 Message。

上层(Task / Agent)
      │  add_user_message(TextArtifact / ImageArtifact / ...)
      ▼
┌─────────────────────────────────────────────┐
│  PromptStack   (provider 无关的中间表示)         │
│  messages: [Message, Message, ...]            │
│  tools:    [BaseTool, ...]                     │
│  output_schema                                 │
└───────────────┬───────────────────────────────┘
                │  driver.run(prompt_stack)
                ▼
┌─────────────────────────────────────────────┐
│  BasePromptDriver  (统一模板 · 抽象基类)         │
│  · 重试 + before/after 发事件                    │
│  · 抽象 try_run / try_stream  ◄── 子类填空       │
└───────────────┬───────────────────────────────┘
                │
     ┌──────────┴───────────┐  各 provider 子类
     ▼          ▼           ▼
 OpenAiChat  Anthropic   Cohere / Google / Bedrock ...
     │   翻成该家 API 请求 → 调用 → 翻回 Message
     ▼
┌─────────────────────────────────────────────┐
│  Message  (provider 无关的回复)                 │
│  content: [TextMessageContent,               │
│            ActionCallMessageContent, ...]     │
└─────────────────────────────────────────────┘

各部件一句话职责:

部件	干什么	在哪个文件
PromptStack	一叠 Message + 可用工具 + 输出 schema,上层唯一要拼的东西	common/prompt_stack/prompt_stack.py
Message	一条消息:一个 role + 一列 MessageContent	common/prompt_stack/messages/message.py
BaseMessageContent	消息里的一"块"内容(文本/图片/音频/工具调用/工具结果)	common/prompt_stack/contents/base_message_content.py
ToolAction	provider 无关地描述"模型要调哪个工具、传什么参"	common/actions/tool_action.py
BasePromptDriver	所有 driver 的模板 + 契约,定义抽象翻译点	drivers/prompt/base_prompt_driver.py
DriversConfig / Defaults	把具体 driver 装配成全局默认,切供应商就切这里	configs/drivers/*、configs/defaults_config.py

---

3. 核心机制一:PromptStack —— provider 无关的中间语

3.1 它要解决的小问题

每家 API 对"一条消息"的定义都不一样:OpenAI 要 {"role": ..., "content": ...},工具结果还要单独一条带 tool_call_id;Anthropic 又是另一套。上层不该关心这些。它需要一个统一的写法:"加一条系统消息""加一条用户消息"。

3.2 三个入口方法

PromptStack 对上只暴露三个便捷方法,分别对应三种 role:

# 真实签名,见 prompt_stack.py:61-68
def add_system_message(self, artifact):     return self.add_message(artifact, Message.SYSTEM_ROLE)
def add_user_message(self, artifact):        return self.add_message(artifact, Message.USER_ROLE)
def add_assistant_message(self, artifact):   return self.add_message(artifact, Message.ASSISTANT_ROLE)

三个 role 常量本身也是 provider 无关的纯字符串,定义在 base_message.py:31-33(SYSTEM_ROLE = "system" 等),配套 is_system()/is_user()/is_assistant() 判定(base_message.py:39-46)。PromptStack 的 system_messages/user_messages/assistant_messages 属性就靠它们过滤(prompt_stack.py:42-52)。

3.3 关键一步:把任意 Artifact 规约成 MessageContent

这是整套中立性的枢纽。上层可以塞进各种 Artifact(文本、图片、音频、工具动作、列表),私有方法 __to_message_content 负责把它们统一映射成 provider 无关的 MessageContent(prompt_stack.py:90-113):

# 真实源码节选,prompt_stack.py:90-113  __to_message_content
def __to_message_content(self, artifact):
    if isinstance(artifact, str):
        return [TextMessageContent(TextArtifact(artifact))]
    if isinstance(artifact, TextArtifact):
        return [TextMessageContent(artifact)]
    if isinstance(artifact, ImageArtifact):
        return [ImageMessageContent(artifact)]
    ...
    if isinstance(artifact, ActionArtifact):
        action = artifact.value
        output = action.output
        if output is None:
            return [ActionCallMessageContent(artifact)]        # 模型"要调工具"
        return [ActionResultMessageContent(output, action=action)]  # 工具"跑完的结果"
    if isinstance(artifact, ListArtifact):
        # 递归展平:一条消息可以同时含文本+图片等多块内容
        processed = [self.__to_message_content(a) for a in artifact.value]
        return [sub for p in processed for sub in p]
    return [TextMessageContent(TextArtifact(artifact.to_text()))]  # 兜底转文本

重点看两处:

• ActionArtifact 分叉——同一个动作,没 output 就是"模型想调工具"(ActionCallMessageContent),有 output 就是"工具已执行的结果"(ActionResultMessageContent)。这让"工具调用"这件事也变成 provider 无关的内容块。
• ListArtifact 递归展平——所以一条 Message 的 content 是一个列表,能同时装文本和图片(多模态),而不是单个字符串。

映射关系一览:

你塞进去的 Artifact	变成的 MessageContent	语义
str / TextArtifact	TextMessageContent	纯文本
ImageArtifact / ImageUrlArtifact	ImageMessageContent	图片
AudioArtifact	AudioMessageContent	音频
GenericArtifact	GenericMessageContent	provider 特有的原样透传块
ActionArtifact(无 output)	ActionCallMessageContent	模型请求调工具
ActionArtifact(有 output)	ActionResultMessageContent	工具执行结果回填
ListArtifact	上述若干块展平	多模态混合
其它	TextMessageContent(to_text() 兜底)	尽力转文本

3.4 MessageContent 的公共契约

所有内容块都继承 BaseMessageContent(base_message_content.py:18-37):它包一个 artifact,提供 to_text(),并要求子类实现一个 from_deltas 类方法——把流式增量拼回完整内容(第 5 节详述)。这个 from_deltas 就是"流式聚合"能 provider 无关的原因:每种内容块自己知道怎么把碎片拼回整体。

3.5 结构化输出的 schema 出口

PromptStack 还带一个 output_schema(可以是 schema.Schema 或 pydantic BaseModel)。to_output_json_schema() 把它统一转成 JSON Schema 字典(prompt_stack.py:70-81),供后面三种结构化输出策略取用。这一步同样与供应商无关——出来的是标准 JSON Schema,谁用谁翻。

---

4. 核心机制二:统一工具协议 —— ToolAction

4.1 它要解决的小问题

OpenAI 的 function calling、Anthropic 的 tool use,在线路上是不同 JSON。但"模型想调某个工具、传某些参数"这个语义是共通的。Griptape 用 ToolAction 把这个语义抽出来。

4.2 ToolAction 的四要素

ToolAction(tool_action.py:16-33)是 provider 无关的"一次工具调用"描述:

字段	含义
tag	这次调用的唯一 id(对回 OpenAI 的 tool_call_id)
name	工具名
path	工具下的具体 activity 名
input	参数字典
output	执行后的结果(回填;None 表示还没执行)

4.3 名字的双向翻译

原生 function calling 只接受一个扁平的函数名,而 Griptape 内部是 name + path 两级。两个方法负责这层双向翻译(tool_action.py:38-55):

# 真实源码,tool_action.py:38-55
def to_native_tool_name(self):                 # ("Calculator","add") -> "Calculator_add"
    parts = [self.name]
    if self.path is not None:
        parts.append(self.path)
    return "_".join(parts)

@classmethod
def from_native_tool_name(cls, native_tool_name):   # "Calculator_add" -> ("Calculator","add")
    parts = native_tool_name.split("_", 1)
    ...

具体 driver 就靠这对方法在"内部两级名"和"provider 扁平名"之间转。以 OpenAI driver 为例:发请求时用 tool.to_native_tool_name(activity) 拼工具名(openai_chat_prompt_driver.py:317),收到工具调用时用 ToolAction.from_native_tool_name(...) 拆回 name/path(openai_chat_prompt_driver.py:403-404)。这就是"工具调用" provider 中立的落点。

---

5. 核心机制三:BasePromptDriver 的模板方法

5.1 思路:把"不变的"写死,把"每家不同的"留白

所有 driver 都要做同样的外围动作:重试、发"开始/结束"事件、流式还是非流式。这些不该每家重写。真正每家不同的只有两点:怎么把请求翻成该家 API、怎么把回复翻回来。

Griptape 用模板方法模式切这一刀——基类写好流程骨架,把两个翻译点留成抽象方法。

5.2 run():统一的骨架

# 真实源码,base_prompt_driver.py:84-100  run()
@observable(tags=["PromptDriver.run()"])
def run(self, prompt_input):
    prompt_stack = (PromptStack.from_artifact(prompt_input)
                    if isinstance(prompt_input, BaseArtifact) else prompt_input)
    for attempt in self.retrying():                 # 指数退避重试
        with attempt:
            self.before_run(prompt_stack)            # 初始化结构化输出 + 发 StartPromptEvent
            result = (self.__process_stream(prompt_stack)   # 流式
                      if self.stream
                      else self.__process_run(prompt_stack)) # 非流式
            self.after_run(result)                   # 发 FinishPromptEvent(带 token 用量)
            return result
    raise Exception("prompt driver failed after all retry attempts")

骨架里三件"不变的事":

1. 重试 —— self.retrying() 来自 ExponentialBackoffMixin,用 tenacity 做指数退避,默认 max_attempts=2、min_retry_delay=2(exponential_backoff_mixin.py:16-26)。
2. 发事件 —— before_run 发 StartPromptEvent,after_run 发 FinishPromptEvent 并带上 input_tokens/output_tokens(base_prompt_driver.py:70-82)。上层的日志、计费、可观测都挂在这套事件上。
3. 流式分叉 —— 由 self.stream 决定走 __process_stream 还是 __process_run。

5.3 两个抽象翻译点

骨架里真正"每家不同"的,就这两个抽象方法(base_prompt_driver.py:128-133):

@abstractmethod
def try_run(self, prompt_stack) -> Message: ...              # 非流式:请求→翻译→回一个 Message
@abstractmethod
def try_stream(self, prompt_stack) -> Iterator[DeltaMessage]: ...  # 流式:吐一串增量

写一个新 provider 支持 = 继承 BasePromptDriver,只实现这两个方法。 重试、事件、流式聚合全都白拿。

5.4 流式聚合:碎片如何拼回完整 Message

流式最麻烦:模型一小段一小段吐(DeltaMessage),但上层想要一个完整 Message。基类的 __process_stream 统一干这件事,子类只管产出增量(base_prompt_driver.py:167-198):

try_stream() 吐出 DeltaMessage 流
        │  每个 delta 带 content.index(标明属于第几块内容)
        ▼
按 index 分桶:delta_contents = { 0: [文本碎片...], 1: [工具调用碎片...] }
        │  同时按类型发 TextChunkEvent / AudioChunkEvent / ActionChunkEvent
        ▼
__build_message():每个桶按类型调 from_deltas 拼回完整块
        │  TextMessageContent.from_deltas / AudioMessageContent.from_deltas
        │  ActionCallMessageContent.from_deltas
        ▼
组装成一个 role=assistant 的完整 Message(带累加的 token usage)

拼接的活分派给每种内容块自己的 from_deltas。比如文本就是把碎片字符串拼起来(text_message_content.py:19-24);工具调用要把 tag/name/path 拼齐、把分片的 partial_input 累积成完整 JSON 再解析(action_call_message_content.py:20-49)——JSON 拼不出来就抛 ValueError。

这个设计的好处:流式聚合逻辑写一次,对所有 provider 生效;新增一种内容类型,只要实现它的 from_deltas,基类不用动。

---

6. 核心机制四:三个可切换开关

基类上有三个字段,是"同一套中间表示、不同投喂方式"的策略开关(base_prompt_driver.py:63-68)。

开关	类型	默认(基类)	作用
use_native_tools	bool	False	用模型原生 function calling,还是退回 ReAct 文本协议
structured_output_strategy	"native"/"tool"/"rule"	"rule"	逼模型产出符合 schema 的结构化输出的三种手法
stream	bool	False	流式还是一次性返回

注意基类默认保守(use_native_tools=False),但具体 driver 可以覆盖——OpenAI driver 就把它默认成 True(openai_chat_prompt_driver.py:90)。

6.1 use_native_tools:原生工具 vs ReAct

• True —— 走该家的原生 function calling。工具会被翻成 provider 的 tools 参数(OpenAI 见 openai_chat_prompt_driver.py:242-243),模型直接返回结构化的工具调用。
• False —— 退回 ReAct:把工具用法写进提示词文本,靠解析模型输出的文字来识别工具调用(解析细节见 02)。

一句话:老模型/不支持 function calling 的供应商用 ReAct 兜底,新模型用原生,上层无感。

6.2 structured_output_strategy:三种逼出结构化输出的策略

当 PromptStack 带了 output_schema,before_run 里的 _init_structured_output 按策略分叉(base_prompt_driver.py:136-162):

output_schema is not None ?
        │
   ┌────┴─────────────────────────────────────────┐
   │ strategy == "tool"                             │
   │   → 塞一个 StructuredOutputTool 进 tools        │
   │     让模型"调这个工具"来交结构化结果             │
   ├────────────────────────────────────────────────┤
   │ strategy == "rule"  (基类默认)                  │
   │   → 把 JSON Schema 当成一条规则文本              │
   │     追加/插入到 system 消息里                    │
   ├────────────────────────────────────────────────┤
   │ strategy == "native"  (在 _base_params 里处理)  │
   │   → 直接用 provider 的原生结构化输出参数          │
   │     (OpenAI: response_format=json_schema)      │
   └────────────────────────────────────────────────┘

前两种("tool"/"rule")在基类的 _init_structured_output 里就地改写 PromptStack——完全 provider 无关。第三种("native")是把 schema 交给 driver 在拼请求时用原生参数,比如 OpenAI 在 _base_params 里翻成 response_format={"type":"json_schema",...}(openai_chat_prompt_driver.py:222-230)。

这就是分层的妙处:能在中间表示层解决的(rule/tool)就 provider 无关地解决,只有真要用原生能力时才下沉到 driver。

---

7. 配置层:换供应商为什么只改一行

前面所有机制,最后靠配置层"装配"起来。

7.1 DriversConfig:一套供应商的打包

一个 DriversConfig 就是"某供应商全家桶"的懒加载容器——prompt driver、embedding、图像、向量库……各配一个。切供应商 = 换一个 config 对象。对比两家配置有多薄:

# anthropic_drivers_config.py:8-12 —— 就覆盖一个 prompt_driver
class AnthropicDriversConfig(DriversConfig):
    @lazy_property()
    def prompt_driver(self):
        return AnthropicPromptDriver(model="claude-3-7-sonnet-latest")

OpenAiDriversConfig 则默认 OpenAiChatPromptDriver(model="gpt-4.1")(openai_drivers_config.py:16-17)。两者暴露给上层的接口完全一样——都是 .prompt_driver 给回一个 BasePromptDriver。上层拿到的是基类类型,不知道也不关心背后是谁。

7.2 Defaults:全局单例

Defaults 是个单例,持有当前生效的 drivers_config,不设时懒加载成 OpenAiDriversConfig(defaults_config.py,drivers_config 的 lazy_property 返回 OpenAiDriversConfig())。这就是"什么都不配默认走 OpenAI"的出处,也是"改一行换供应商"里那一行:

Defaults.drivers_config = AnthropicDriversConfig()   # 全局切到 Anthropic

7.3 局部切换:with 上下文

不想改全局、只想临时用另一家?BaseDriversConfig 实现了上下文管理器:进入 with 时把自己设为 Defaults.drivers_config、退出时还原(base_drivers_config.py:53-68)。

# 示意,非源码 —— 只在这个块里用 Anthropic
with AnthropicDriversConfig():
    Agent().run("只有这句走 claude")
# 出了 with,自动回到原来的默认

---

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

• 中间表示是整套中立性的地基。 只要上层永远只碰 PromptStack/Message/MessageContent,供应商差异就被逼到 driver 一层。妙在 __to_message_content 这一个规约函数(prompt_stack.py:90-113)把"任意输入"收敛成有限几种内容块。
• 模板方法把"不变"和"可变"切得很干净。 重试、事件、流式聚合在基类只写一次,子类只填 try_run/try_stream(base_prompt_driver.py:128-133)。新增一家供应商成本极低。
• 流式聚合的分派下沉给内容块自己。 __build_message 不用 if-else 判类型拼接,而是各内容块实现 from_deltas(base_message_content.py:34-37)。加一种新内容类型不用改基类。
• 同一件事优先在中立层解决。 结构化输出的 rule/tool 两策略在中间表示层就地改写 PromptStack(base_prompt_driver.py:136-162),只有 native 才动 provider 参数。分层克制。
• 工具名双向翻译隔离 provider 命名约束。 to_native_tool_name/from_native_tool_name(tool_action.py:38-55)把"扁平函数名"这种 provider 约束挡在 driver 边界外。

---

9. 边界与局限

• 不是所有内容块都支持流式聚合。 ImageMessageContent.from_deltas 和 ActionResultMessageContent.from_deltas 直接 raise NotImplementedError(image_message_content.py:20-21、action_result_message_content.py:21-22)——图片和工具结果不是流式增量产物,拼接对它们无意义。
• 重试很浅。 默认 max_attempts=2(exponential_backoff_mixin.py:18),且 ignored_exception_types 默认吞掉 ImportError/ValueError(base_prompt_driver.py:60)——这些不重试。想要更强的重试要自己调字段。
• provider 能力不对齐时靠 driver 覆盖开关兜底。 基类默认 use_native_tools=False,但每家实际能力不同,得由具体 driver 覆盖(如 OpenAI 覆盖成 True)。中间表示统一,不代表每家 provider 支持每种模态/能力——不支持的最终还是会在 driver 层暴露。
• prompt_stack_to_string 只是粗糙近似。 基类那版拼字符串仅供 token 估算,注释明说应由子类按各家 token 规则覆盖(base_prompt_driver.py:102-126)。

---

10. 横向对比

同组其它章的衔接:

• 谁在构造 PromptStack、谁在解析 ReAct 文本 → 02-prompt-task-agent-loop.md。
• 工具怎么从 @activity 变成模型可见的 schema(即本章 to_native_tool_name 拼名字之前那半段) → 04-tools-and-activities.md。
• Artifact 类型系统(本章反复出现的 TextArtifact/ImageArtifact/ActionArtifact)→ 05-memory-and-artifacts.md。
• Defaults 全局装配的完整图景 → 06-engines-and-configs.md。
• 总览与阅读地图 → index.md。

---

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

主题	文件路径	符号名
中间表示容器	common/prompt_stack/prompt_stack.py	PromptStack
三个加消息入口	common/prompt_stack/prompt_stack.py	add_system_message / add_user_message / add_assistant_message
Artifact→内容块 规约(枢纽)	common/prompt_stack/prompt_stack.py	PromptStack.__to_message_content
schema 出口	common/prompt_stack/prompt_stack.py	to_output_json_schema
一条消息	common/prompt_stack/messages/message.py	Message
role 常量与判定	common/prompt_stack/messages/base_message.py	SYSTEM_ROLE / is_user
内容块基类 + 流式契约	common/prompt_stack/contents/base_message_content.py	BaseMessageContent / from_deltas
文本块拼接	common/prompt_stack/contents/text_message_content.py	TextMessageContent.from_deltas
工具调用块拼接	common/prompt_stack/contents/action_call_message_content.py	ActionCallMessageContent.from_deltas
统一工具动作	common/actions/tool_action.py	ToolAction
工具名双向翻译	common/actions/tool_action.py	to_native_tool_name / from_native_tool_name
driver 模板骨架	drivers/prompt/base_prompt_driver.py	BasePromptDriver.run
两个抽象翻译点	drivers/prompt/base_prompt_driver.py	try_run / try_stream
结构化输出三策略分叉	drivers/prompt/base_prompt_driver.py	_init_structured_output
流式聚合	drivers/prompt/base_prompt_driver.py	__process_stream / __build_message
三个开关字段	drivers/prompt/base_prompt_driver.py	use_native_tools / structured_output_strategy / stream
重试 mixin	mixins/exponential_backoff_mixin.py	retrying
OpenAI 请求翻译(参考实现)	drivers/prompt/openai_chat_prompt_driver.py	_base_params / __to_openai_messages
供应商全家桶配置	configs/drivers/anthropic_drivers_config.py	AnthropicDriversConfig
全局默认单例	configs/defaults_config.py	Defaults / _DefaultsConfig
局部切换上下文	configs/drivers/base_drivers_config.py	BaseDriversConfig.__enter__