Griptape — 架构与原理

30 秒导读: Griptape 是一个 Python 的 genAI 应用框架。它不替你写 prompt,而是给你一套可替换的抽象积木:你把工作拆成一串 Task(任务),用 Agent / Pipeline / Workflow 把这些任务编排成一张有向图;每个 Task 通过一个 provider 无关的 PromptStack(提示词栈)去驱动一个可热插拔的 Driver(驱动)调具体 LLM;带工具的 PromptTask 内置一个"想 → 调工具 → 看结果 → 再想"的循环,ReAct 文本协议和模型原生 function-calling 两套都支持;工具的大块/敏感输出可以走 off-prompt 的 Task Memory,不塞进提示词只留一个引用。核心卖点是:换模型、换向量库、换记忆后端,业务代码基本不动。

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

一句话定义: Griptape 是一个把"调 LLM 做事"这件事拆成标准积木的 Python 框架——你搭积木,它管编排、提示词组装、工具调用和记忆。

解决什么问题 / 给谁用。 假设你要写一个 AI:先上网搜资料、再总结、再按格式输出,中途还得让模型自己决定"要不要调计算器 / 要不要再搜一次"。裸调 OpenAI SDK 你得手写:怎么把历史对话拼进 prompt、怎么解析模型说的"我要调工具 X"、怎么把工具结果喂回去、怎么防止 20 KB 的网页正文撑爆上下文。Griptape 把这些全都做成可复用的部件,给的是:

写 LLM 应用的后端工程师;
想要"业务逻辑和具体 provider 解耦"的团队(今天用 OpenAI,明天换 Anthropic,不想重写)。

它能做什么(功能):

单步智能体(Agent)、串行流水线(Pipeline)、并行工作流(Workflow);
工具调用(内置计算器、网页搜索、文件、SQL 等,也可自定义);
对话记忆、任务记忆(off-prompt)、元数据记忆三种记忆;
20+ 类 Driver:prompt / embedding / 向量库 / rerank / 图像 / 语音…… 可随意换 provider;
RAG、抽取、总结等用例被封成 Engine。

用起来什么样。 最小的一个真实例子(来自 README,gpt-4.1):

from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.rules import Rule
from griptape.tasks import PromptTask

task = PromptTask(
    prompt_driver=OpenAiChatPromptDriver(model="gpt-4.1"),
    rules=[Rule("Keep your answer to a few sentences.")],
)

result = task.run("How do I do a kickflip?")
print(result.value)

给它几个 tools=[...],task.run() 内部就会自动跑起"调工具—看结果—再想"的循环,直到模型给出 Answer。

一句话直觉/类比。 把 Griptape 想成乐高底板 + 一堆标准接口的积木:Task 是积木块,Structure 是把积木拼成图的底板,Driver 是"任何一块积木背面都长一样的插槽"——插 OpenAI 还是插 Anthropic,底板不用改。

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

2.1 六层抽象栈

Griptape 的价值全在"分层且每层可替换"。怎么读下面这张图:从上到下是调用/依赖方向,上层用下层,越往下越贴近 LLM provider。

      你的代码
         │  给 input、给 tools、.run()
         ▼
┌─────────────────────────────────────────────┐
│ Structure   Agent / Pipeline / Workflow      │  编排:把 Task 排成一张 DAG,决定谁先谁后、能否并行
└───────────────┬─────────────────────────────┘
                │ 调度 Task.run()
                ▼
┌─────────────────────────────────────────────┐
│ Task        PromptTask / ToolTask / RagTask… │  一个工作单元;PromptTask 内含"工具循环"
└───────────────┬─────────────────────────────┘
                │ 组装 prompt_stack、跑 subtask
                ▼
┌─────────────────────────────────────────────┐
│ PromptStack + Message   (provider 无关的对话) │  统一的"对话数据模型",谁都看得懂
└───────────────┬─────────────────────────────┘
                │ driver.run(prompt_stack)
                ▼
┌─────────────────────────────────────────────┐
│ Driver      OpenAiChat / Anthropic / …       │  把统一模型翻译成某家 API 的请求/响应
└───────────────┬─────────────────────────────┘
                │ 真正的 HTTP 调用
                ▼
             LLM Provider

旁挂两条支线,横切所有层:

   Tool (@activity) ──► 被 PromptTask 转成 LLM 可调的 JSON schema
   Memory:  Conversation(对话历史) / Task(off-prompt 大输出) / Meta(元数据)

2.2 部件一句话职责

部件	干什么	在哪个文件
`Structure`	抽象基类:装一组 Task、维护父子关系、跑 `before/after_run`、写对话记忆	`griptape/structures/structure.py`
`Agent`	只有 1 个 Task 的 Structure;构造时自动建一个 `PromptTask`	`griptape/structures/agent.py`
`Pipeline`	把 Task 串成链,前一个输出流进后一个;递归执行	`griptape/structures/pipeline.py`
`Workflow`	Task 组成 DAG,用拓扑排序 + 线程池并行跑就绪任务	`griptape/structures/workflow.py`
`BaseTask`	任务基类:状态机(PENDING→RUNNING→FINISHED/SKIPPED)、`can_run` 门控	`griptape/tasks/base_task.py`
`PromptTask`	核心任务:组装 `PromptStack`、跑工具循环、校验结构化输出	`griptape/tasks/prompt_task.py`
`ActionsSubtask`	一轮工具调用:解析模型意图(ReAct 或原生)、并行执行工具	`griptape/tasks/actions_subtask.py`
`PromptStack` / `Message`	provider 无关的对话数据模型	`griptape/common/prompt_stack/`
`BasePromptDriver`	驱动基类:`run()` 统一入口,子类实现 `try_run` / `try_stream`	`griptape/drivers/prompt/base_prompt_driver.py`
`BaseTool` + `@activity`	工具:每个被 `@activity` 标注的方法变成 LLM 可调的一个动作	`griptape/tools/base_tool.py`、`griptape/utils/decorators.py`
`TaskMemory`	off-prompt 记忆:把大/敏感工具输出存起来,提示词里只留引用	`griptape/memory/task/task_memory.py`
`Defaults`	单例全局配置:默认用哪套 Driver、日志配置	`griptape/configs/defaults_config.py`

2.3 主线走一遍(高层,不进代码)

以"带工具的 Agent 回答一个问题"为例,一次 agent.run("...") 大致是:

输入
  │
  ▼
Agent.run()  ── before_run:reset 所有 task、发 StartStructureRunEvent、解析父子关系
  │
  ▼
PromptTask.try_run()
  │  1) 组装 prompt_stack:system 模板(含工具 schema/规则)+ 对话记忆 + 用户输入 + 历史子任务
  │  2) prompt_driver.run(prompt_stack) → 得到模型这一轮的 Message
  │  3) 交给 subtask_runners 依次处理:
  │        ├─ 工具循环:模型说"要调工具"→ ActionsSubtask 解析并并行执行 → 结果回填 → 再问模型
  │        │            (循环直到模型给 Answer,或超过 max_subtasks=20)
  │        └─ 结构化输出校验:若设了 output_schema,校验/纠正 JSON
  │
  ▼
Agent.after_run() ── 把 (input, output) 作为一条 Run 写入对话记忆、发 FinishStructureRunEvent
  │
  ▼
输出(一个 Artifact,取 .value 拿到实际值)

依据:structures/structure.py:200(run)、structures/agent.py:90(try_run)、tasks/prompt_task.py:208(try_run)、structures/structure.py:172(after_run 写记忆)。

3. 阅读地图(建议顺序)

这是一份多文件讲解,按"由浅入深"排。建议顺序:

Griptape — 架构与原理(总览与阅读地图) —— 你正在读的这页:全景与导航。
结构与任务图:Agent / Pipeline / Workflow 怎么调度任务 —— 三种 Structure 的区别、Task 状态机、DAG 怎么建与怎么跑(串行递归 vs 拓扑并行)。先看这章建立"骨架"认知。
PromptTask 智能体循环:提示词组装与 ReAct 子任务 —— 框架的心脏:prompt_stack 怎么拼、工具循环怎么转、max_subtasks 与 reflect_on_tool_use 的作用。
驱动与 provider 中立:PromptStack、Message 与统一工具协议 —— 为什么换 provider 业务不动:统一对话模型 + Driver 翻译层 + ReAct/native 双工具协议 + 结构化输出三策略。
工具与 activity 机制:@activity 装饰器如何变成 LLM 可调的 schema —— @activity 怎么把普通方法变成带 JSON schema 的可调动作、参数校验、{"values": ...} 派发约定。
记忆与制品:对话记忆、off-prompt 任务记忆、Artifact 类型系统 —— 三种记忆各自解决什么,off-prompt 如何把大输出挡在提示词外,Artifact 类型体系。
引擎与配置:RAG 等用例封装 + Defaults 全局装配 —— RagEngine 的三阶段管线、Defaults 单例与 DriversConfig 上下文管理器怎么做全局/局部装配。

如果你只想快速判断"该读哪章":想懂编排看 01;想懂智能体怎么自己调工具看 02;想懂为什么能换模型看 03;想写自己的工具看 04;想懂上下文怎么不爆看 05;想用 RAG / 全局换 provider 看 06。

4. 巧妙之处(可以带走的技术)

下面每条都是"不显然但值得学"的设计决策。

4.1 一套抽象吃下"单步 / 串行 / 并行"

Agent、Pipeline、Workflow 共享同一个 Structure 基类和同一个 Task 状态机,区别只在 try_run 的调度策略:

Agent:只允许 1 个 Task,直接 task.run()(structures/agent.py:90,且 add_tasks 超过 1 个就报错,见 agent.py:84);
Pipeline:尾递归沿 task.children 一路跑下去(structures/pipeline.py:71,__run_from_task);
Workflow:用标准库 graphlib.TopologicalSorter 把 Task 图拓扑排序,再用线程池 submit 所有 can_run() 为真的任务并行跑(structures/workflow.py:102,try_run;排序在 workflow.py:152,order_tasks)。

妙在:换编排语义只换调度那一小段,记忆/事件/父子关系全复用。

4.2 provider 中立不是靠"适配器接口",而是靠"统一数据模型"

关键不是 Driver 有共同方法,而是大家都围着 PromptStack + Message 转。PromptStack 是纯粹的、与任何 provider 无关的对话结构(common/prompt_stack/prompt_stack.py:36),Message 里装的是 TextMessageContent / ImageMessageContent / ActionCallMessageContent 等内容块(prompt_stack.py:90,__to_message_content)。Driver 的唯一职责就是把这套统一模型翻译成某家 API 的字段。所以多模态、工具调用、结果回填全都用同一套结构表达,换 provider = 换翻译器。

4.3 ReAct 与原生 function-calling,一个开关切换

同一个 PromptTask,靠 prompt_driver.use_native_tools 一个布尔值决定走哪条路(drivers/prompt/base_prompt_driver.py:64):

	`use_native_tools=False`(ReAct)	`use_native_tools=True`(原生)
模型怎么表达"要调工具"	输出 `Thought:` / `Actions:` 文本,框架用正则解析	走模型 API 的 tool_call 字段,直接是结构化 `ActionArtifact`
解析在哪	`tasks/actions_subtask.py:266`(`__init_from_prompt`,正则 `THOUGHT/ACTIONS/ANSWER`)	`actions_subtask.py:284`(`__init_from_artifact`)
历史怎么回填	渲染成 Jinja 文本贴回 stack	用 `ActionCall` / `ActionResult` 内容块贴回(`actions_subtask.py:200`)

system prompt 模板里也用 {% if not use_native_tools %} 决定要不要教模型 ReAct 格式(templates/tasks/prompt_task/system.j2)。一份任务代码,两套协议无缝切。

4.4 off-prompt:把大输出"寄存"起来,只给模型一张取件单

工具默认 off_prompt=False(tools/base_tool.py:66)。一旦设为 True,工具输出不进提示词,而是被 TaskMemory.process_output 存进 storage,提示词里只留一段"结果已存到 memory X 的 namespace Y,你可以用它"的引用(memory/task/task_memory.py:61)。这样几十 KB 的网页/查询结果不会撑爆上下文,也不会把敏感数据直接喂给模型——后续工具可以按 namespace 把它取回来接着处理。

4.5 工具 = 加了装饰器的普通方法

写工具不需要手写 JSON schema。你给方法加 @activity({"description": ..., "schema": ...}),装饰器就挂上 is_activity / name / config 三个属性(utils/decorators.py:31);ActivityMixin.activities() 用 inspect.getmembers 反射把它们收集起来(mixins/activity_mixin.py:55),再自动生成给 LLM 看的 schema。派发时有个约定:LLM 面向的 schema 不含 values 包裹,框架在 try_run 里补上 {"values": <input>} 再喂给方法(tools/base_tool.py:152)。业务方法保持干净,schema 全自动。

4.6 全局默认用单例 + 上下文管理器做"局部覆盖"

Defaults 是个单例(configs/defaults_config.py:17),默认 drivers_config 惰性初始化为 OpenAiDriversConfig。想临时换一整套 provider,DriversConfig 实现了 __enter__/__exit__:进入 with 块时把自己设为全局默认、退出时还原(configs/drivers/base_drivers_config.py:53)。于是"全局配一次 + 局部临时覆盖"两种需求用同一个机制解决。

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

按主题给出文件 + 真实符号名(符号名比行号抗漂移,可直接 grep)。

主题	文件	关键符号
Structure 基类 / 运行骨架	`griptape/structures/structure.py`	`Structure`、`run`、`before_run`、`after_run`、`add_tasks`、`resolve_relationships`
Agent(单任务)	`griptape/structures/agent.py`	`Agent`、`try_run`、`_init_task`、`validate_fail_fast`
Pipeline(串行递归)	`griptape/structures/pipeline.py`	`Pipeline`、`try_run`、`_Pipeline__run_from_task`、`insert_task`
Workflow(拓扑并行)	`griptape/structures/workflow.py`	`Workflow`、`try_run`、`order_tasks`、`to_graph`、`insert_tasks`
Task 状态机 / DAG 门控	`griptape/tasks/base_task.py`	`BaseTask`、`State`、`run`、`can_run`、`reset`
智能体核心任务	`griptape/tasks/prompt_task.py`	`PromptTask`、`prompt_stack`、`try_run`、`default_run_actions_subtasks`、`subtask_runners`、`DEFAULT_MAX_STEPS`
一轮工具调用	`griptape/tasks/actions_subtask.py`	`ActionsSubtask`、`__init_from_prompt`、`__init_from_artifact`、`run_action`、`add_to_prompt_stack`
统一对话模型	`griptape/common/prompt_stack/prompt_stack.py`	`PromptStack`、`add_system_message`、`__to_message_content`、`to_output_json_schema`
消息 / 内容块	`griptape/common/prompt_stack/messages/message.py`	`Message`、`to_artifact`、`to_text`
Driver 基类	`griptape/drivers/prompt/base_prompt_driver.py`	`BasePromptDriver`、`run`、`try_run`、`try_stream`、`use_native_tools`、`_init_structured_output`
工具基类 / 派发	`griptape/tools/base_tool.py`	`BaseTool`、`run`、`try_run`、`after_run`、`off_prompt`、`activity_schemas`
activity 装饰器	`griptape/utils/decorators.py`	`activity`、`lazy_property`
activity 反射收集	`griptape/mixins/activity_mixin.py`	`ActivityMixin`、`activities`、`activity_schema`、`to_activity_json_schema`
工具动作对象	`griptape/common/actions/tool_action.py`	`ToolAction`、`to_native_tool_name`、`from_native_tool_name`
off-prompt 任务记忆	`griptape/memory/task/task_memory.py`	`TaskMemory`、`process_output`、`store_artifact`、`load_artifacts`
对话记忆	`griptape/memory/structure/base_conversation_memory.py`	`BaseConversationMemory`、`add_run`、`add_to_prompt_stack`、`max_runs`
制品类型系统	`griptape/artifacts/base_artifact.py`	`BaseArtifact`、`to_text`、`value`、`meta`
RAG 引擎	`griptape/engines/rag/rag_engine.py`	`RagEngine`、`process_query`、`query_stage`、`retrieval_stage`、`response_stage`
全局配置单例	`griptape/configs/defaults_config.py`	`_DefaultsConfig`、`Defaults`、`drivers_config`
Driver 配置 / 局部覆盖	`griptape/configs/drivers/base_drivers_config.py`	`BaseDriversConfig`、`__enter__`、`__exit__`、`prompt_driver`

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

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

2.1 六层抽象栈​

2.2 部件一句话职责​

2.3 主线走一遍(高层,不进代码)​

3. 阅读地图(建议顺序)​

4. 巧妙之处(可以带走的技术)​

4.1 一套抽象吃下"单步 / 串行 / 并行"​

4.2 provider 中立不是靠"适配器接口",而是靠"统一数据模型"​

4.3 ReAct 与原生 function-calling,一个开关切换​

4.4 off-prompt:把大输出"寄存"起来,只给模型一张取件单​

4.5 工具 = 加了装饰器的普通方法​

4.6 全局默认用单例 + 上下文管理器做"局部覆盖"​

5. 代码地图(导航索引)​