跳到主要内容

记忆与制品:对话记忆、off-prompt 任务记忆、Artifact 类型系统

30 秒导读: Griptape 有两块数据基座。一块是记忆——把"该记住的东西"存在框架里,分三种:对话历史(Conversation Memory)、大块工具输出(Task Memory)、跨任务共享的元数据(Meta Memory)。另一块是 Artifact 类型系统——所有部件之间传的每一份数据都装在一个带类型的"信封"里(文本、JSON、二进制、图片、动作、错误……)。两者在一个关键机制上交汇:off-prompt——工具吐出的大/敏感结果不直接塞进提示词,而是留在 Task Memory 里,只把一句"它存哪了"的引用喂回模型。

本章讲这两块数据侧的东西。工具为什么/如何触发 off-prompt(off_prompt 开关怎么接线)在工具与 activity 机制讲;Task Memory 底下的向量检索(RAG)在引擎与配置讲。这里只讲:记忆怎么存、Artifact 有哪些类型、off-prompt 的数据怎么流。

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

先建立两个心智模型。

记忆 = 智能体的"记事本"。 一个 LLM 本身是无状态的:每次调用都是一张白纸,不记得上一轮说过什么。要让智能体"记得住",就得有人在框架侧把该记的东西存下来、下一轮再拼回提示词。Griptape 把"该记的东西"分成三类,对应三种记忆:

记忆种类记什么一句话类比
Conversation Memory(对话记忆)每一轮的"你问 / 它答"聊天记录本
Task Memory(任务记忆)工具吐出的大块/敏感结果后台文件柜(只给别人递取件号)
Meta Memory(元记忆)跨任务共享的零碎元数据便利贴

Artifact = 数据的"信封"。 Griptape 里,部件之间(任务→任务、工具→记忆、驱动→任务)传的从来不是裸的 str 或 bytes,而是一个 Artifact 对象——它裹着 value(真数据)、name(名字)、meta(附加信息),并保证有一个 to_text() 能变成给 LLM 看的文本。不同种类的数据用不同的 Artifact 子类:文本用 TextArtifact、二进制用 BlobArtifact、出错用 ErrorArtifact……

为什么要有信封? 因为整条流水线要能统一处理任意数据:记忆要能判断"这是文本还是二进制,该存哪";任务要能把上游产物 to_text() 拼进提示词;工具要能把结果打包回传。有了统一的 BaseArtifact 契约,这些部件就不用关心里面到底是什么——它们只跟"信封"打交道。

2. 顶层全景(两块基座怎么转)​

                     ┌─────────────────────────────────────────┐
                     │            Artifact 类型系统              │
                     │   所有部件之间流动的数据都装在信封里       │
                     │   Text / Json / List / Blob / Image /    │
                     │   Action / Error / Info / Boolean ...    │
                     └─────────────────────────────────────────┘
                          ▲            ▲              ▲
              装进信封     │            │ 装进信封      │ 装进信封
                          │            │              │
   ┌──────────────────────┴──┐  ┌──────┴───────┐  ┌───┴────────────────┐
   │  Conversation Memory     │  │ Task Memory  │  │   Meta Memory       │
   │  跨轮:每轮 input/output   │  │ off-prompt:  │  │  跨任务:元数据条目   │
   │  → 拼回提示词的对话历史    │  │ 大块工具输出  │  │  (thought/action…)  │
   └──────────────────────────┘  └──────────────┘  └─────────────────────┘
              memory/structure       memory/task        memory/meta

三种记忆各自用 Artifact 当存取单位,但职责完全不同:

部件干什么在哪个文件
ConversationMemory累积每轮 Run,to_prompt_stack 把历史还原成 user/assistant 消息memory/structure/conversation_memory.py
SummaryConversationMemory同上,但超过 offset 的老 Run 自动用 LLM 压成一段摘要memory/structure/summary_conversation_memory.py
TaskMemory按 Artifact 类型把工具输出分派到不同 storage,返回"取件号"memory/task/task_memory.py
TextArtifactStorage / BlobArtifactStorage具体落地:文本进向量库、二进制进内存字典memory/task/storage/
MetaMemory存 BaseMetaEntry 列表,供下游任务读memory/meta/meta_memory.py
BaseArtifact 及其子类数据信封,定义 value/name/meta/to_text 契约artifacts/*.py

3. Conversation Memory:跨轮的对话历史​

3.1 一个 Run 就是一轮问答​

对话记忆的最小单位是 Run——一轮交互的"输入 + 输出",两边都是 Artifact(run.py:14-19):

@define(kw_only=True)
class Run(SerializableMixin):
    id: str = field(...)                 # 随机 uuid
    meta: dict | None = field(...)
    input: BaseArtifact = field(...)     # 这轮喂进去的
    output: BaseArtifact = field(...)    # 这轮吐出来的

ConversationMemory 就是一串 Run 的容器。它的两个核心动作极其朴素(conversation_memory.py:11-20):

def try_add_run(self, run: Run) -> None:
    self.runs.append(run)                       # 追加一轮

def to_prompt_stack(self, last_n=None) -> PromptStack:
    for run in runs:
        prompt_stack.add_user_message(run.input)      # 还原成 user 消息
        prompt_stack.add_assistant_message(run.output)  # 还原成 assistant 消息

也就是说:存的时候按轮追加,取的时候把每轮拆成一条 user + 一条 assistant,拼成一段对话史。

3.2 落记忆的时机:per_structure vs per_task​

记忆什么时候被写进去?Griptape 给了两个策略,由 conversation_memory_strategy 控制。

per_structure(默认):整个结构跑完才落一条 Run。 在 Structure.after_run 里,拿首个任务的输入和末个任务的输出打包成一个 Run(structures/structure.py:176-185):

if (self.conversation_memory_strategy == "per_structure"
        and self.conversation_memory is not None ...):
    run = Run(input=self.input_task.input, output=self.output_task.output)
    self.conversation_memory.add_run(run)

per_task:每个任务各落一条 Run。 在 PromptTask.after_run 里,只要不是 per_structure 就每个任务自己落(tasks/prompt_task.py:198-206):

if ((self.structure is None or self.structure.conversation_memory_strategy == "per_task")
        and conversation_memory is not None and self.output is not None):
    run = Run(input=self.input, output=self.output)
    conversation_memory.add_run(run)

一句话记住这个区别:

策略落记忆的粒度适合
per_structure一次 run() 存一条(首输入→末输出)把多任务流水线当成"一轮"对外
per_task每个任务各存一条想让每个任务步骤都进历史

3.3 历史注入到提示词的哪儿:system 之后​

历史存下来后,由 PromptTask 在组装提示词时插回去。关键是插的位置——紧跟在 system 提示词后面(tasks/prompt_task.py:142-144):

if memory is not None:
    # inserting at index 1 to place memory right after system prompt
    memory.add_to_prompt_stack(self.prompt_driver, stack, 1 if system_template else 0)

提示词怎么组装的全貌在 PromptTask 智能体循环 讲;本节只讲存储侧——记忆怎么把自己变成一段可插入的消息。

3.4 存取的旋钮:autoload / max_runs / autoprune​

BaseConversationMemory 上挂了几个控制存取行为的字段(base_conversation_memory.py:21-28):

字段默认作用
autoloadTrue初始化时自动从 driver 把历史 Run 读回来(:30-32, :58-63)
max_runsNone上限;add_run 后超出就从头 pop(:44-48)
autopruneTrue注入提示词时,按 token 上限能塞几条塞几条

autoprune 是最值得看的一处巧思(base_conversation_memory.py:65-114)。它不是简单截断,而是试探性地二分:从"想塞全部 Run"开始,反复用 tokenizer 数一遍拼进去后还剩多少 token,剩负数就砍掉一条再试,直到刚好塞得下:

while should_prune and num_runs_to_fit_in_prompt > 0:
    memory_inputs = self.to_prompt_stack(num_runs_to_fit_in_prompt).messages
    tokens_left = prompt_driver.tokenizer.count_input_tokens_left(...)
    if tokens_left > 0:
        should_prune = False        # 塞得下,停
    else:
        num_runs_to_fit_in_prompt -= 1   # 塞不下,少一条再试

每次 add_run 之后还会调 conversation_memory_driver.store(...) 落盘(:48)——具体存到哪(内存 / 本地文件 / Redis 等)由 driver 决定,属于驱动与 provider 中立的范畴。

3.5 变体:SummaryConversationMemory 自动摘要​

对话一长,历史 token 会爆。SummaryConversationMemory 的办法是:只保留最近 offset 条原文,更老的用 LLM 压成一段摘要(summary_conversation_memory.py)。

每次加新 Run 时判断:未摘要的 Run 超过 offset 了吗?超了就把多出来的老 Run 交给 LLM 压进摘要,并把 summary_index 往前推(:87-94):

def try_add_run(self, run: Run) -> None:
    self.runs.append(run)
    unsummarized_runs = self.unsummarized_runs()
    runs_to_summarize = unsummarized_runs[: max(0, len(unsummarized_runs) - self.offset)]
    if len(runs_to_summarize) > 0:
        self.summary = self.summarize_runs(self.summary, runs_to_summarize)
        self.summary_index = 1 + self.runs.index(runs_to_summarize[-1])

注入提示词时,就只发摘要 + 最近 offset 条原文(:67-74)。注意一个细节:self.runs 里始终保留全部原始 Run(供检视),摘要只影响"发给 LLM 的那份"。

4. Task Memory:off-prompt 机制​

4.1 要解决的问题:别让大象钻进提示词​

设想一个工具去读了一个 50MB 的网页、或查出一张含敏感字段的表。如果把这坨原文直接塞回提示词,后果是:

  • token 直接爆掉,或烧掉大量成本;
  • 敏感内容白白进了发往模型厂商的请求;
  • 模型注意力被无关噪声淹没。

off-prompt 的思路:把大象留在框架里,只递给模型一张取件号。 工具的大块输出存进 Task Memory,提示词里只出现一句"它存到 memory_name=…、artifact_namespace=… 了";后续别的工具想用,报上名字去把它取回来。模型全程没见过原文。

4.2 按 Artifact 类型分派 storage​

TaskMemory 的核心是一张"Artifact 类型 → storage"的分派表(memory/task/task_memory.py:20-40):

@define
class TaskMemory(ActivityMixin, SerializableMixin):
    artifact_storages: dict[type, BaseArtifactStorage] = field(
        default=Factory(lambda: {
            TextArtifact: TextArtifactStorage(),   # 文本 → 向量库
            BlobArtifact: BlobArtifactStorage(),   # 二进制 → 内存字典
        }))
    namespace_storage: dict[str, BaseArtifactStorage] = ...   # 名字 → 用了哪个 storage
    namespace_metadata: dict[str, Any] = ...                  # 名字 → 产生它的 action JSON

选 storage 的逻辑在 get_storage_for(:51-59):按 isinstance 找匹配的类型;若是 ListArtifact,就看列表里第一个元素的类型来决定整份存哪:

def get_storage_for(self, artifact):
    if isinstance(artifact, ListArtifact):
        if artifact.has_items():
            return find_storage(artifact.value[0])   # 按首元素类型分派
        return None
    return find_storage(artifact)

两个内置 storage 的落地方式完全不同:

storage存到哪关键代码
TextArtifactStoragevector_store_driver.upsert(...),进向量库(为后续语义检索铺路)storage/text_artifact_storage.py:24-31
BlobArtifactStorage就是个 dict[namespace → list[BlobArtifact]],进程内存storage/blob_artifact_storage.py:16-26

文本走向量库这条线为什么、怎么检索,属于 RAG,归 引擎与配置。本章只需知道:分派表把不同类型送去不同 storage。

4.3 process_output:存工件、换引用​

工具跑完,BaseTool.after_run 会把结果交给 TaskMemory.process_output(触发点在 tools/base_tool.py:189,是否触发取决于工具的 off_prompt,见 04)。process_output 干三件事(memory/task/task_memory.py:61-97):

def process_output(self, tool_activity, subtask, output_artifact):
    namespace = output_artifact.name              # 取件号 = artifact 的 name
    if output_artifact:
        result = self.store_artifact(namespace, output_artifact)   # ① 存
        if result:
            return result                          # 存不下就把原物/错误直接返回
        self.namespace_metadata[namespace] = subtask.actions_to_json()  # ② 记来源
        output = J2("memory/tool.j2").render(...)  # ③ 渲染"存哪了"这句话
        ...
        return InfoArtifact(output, name=namespace)   # 只把引用喂回提示词
    return InfoArtifact("tool output is empty")

渲染出来的那句引用长这样(templates/memory/tool.j2):

Output of "{{tool_name}}.{{activity_name}}" was stored in memory with memory_name "{{memory_name}}" and artifact_namespace "{{artifact_namespace}}".

模型看到的只有这句话,原文留在 storage 里。

store_artifact 里有个值得注意的守卫(:99-122):同一个 namespace 不允许被两种不同 storage 混用,否则返回 ErrorArtifact("error storing tool output in memory");ListArtifact 会被拆开、逐个元素存进同一 namespace。

4.4 取回:报上名字 load​

后续工具要用这份数据时,TaskMemory.load_artifacts(namespace) 按名字回到对应 storage 把它取回来(:124-129):

def load_artifacts(self, namespace: str) -> ListArtifact:
    storage = self.namespace_storage.get(namespace)   # 当初存哪了
    if storage:
        return storage.load_artifacts(namespace)
    return ListArtifact()

而 find_input_memory 让工具能按名字认出"我要读的是不是这块记忆"(:131-134)。这样,一份工具输出就能在多个工具之间以取件号流转,自始至终不进提示词。

4.5 一张图看懂 off-prompt 数据流​

怎么读:左边是工具吐结果,顺着实线走到 Task Memory 存下;只有虚线那一小股(引用文本)回到提示词;后续工具靠取件号(namespace)横向取回原物。

  工具 activity 跑完
        │  value: BaseArtifact(大块/敏感)
        ▼
  BaseTool.after_run ──► TaskMemory.process_output(namespace = artifact.name)
        │                        │
        │                        ├─① store_artifact ─► get_storage_for(按类型)
        │                        │        ├─ TextArtifact ─► TextArtifactStorage ─► 向量库
        │                        │        └─ BlobArtifact ─► BlobArtifactStorage ─► 内存 dict{ns:[blob]}
        │                        │
        │                        ├─② namespace_metadata[ns] = 产生它的 action JSON
        │                        │
        │                        └─③ 渲染 memory/tool.j2 → InfoArtifact("存到 ns 了")
        │                                 ╎ (只有这一句引用)
        ▼                                 ╎
  ┌───────────────────────────────────────▼───────────┐
  │ 提示词 (PromptStack)   ← 模型只看到"取件号",没见原文 │
  └────────────────────────────────────────────────────┘
                              ▲
   后续工具 ── load_artifacts(namespace) ──┘   (报上名字,原物取回,仍不进提示词)

5. Meta Memory:跨任务的便利贴​

MetaMemory 是最简单的一块:一个 BaseMetaEntry 列表,加一个 add_entry(memory/meta/meta_memory.py:12-22):

@define
class MetaMemory:
    entries: list[BaseMetaEntry] = field(factory=list, kw_only=True)
    def add_entry(self, entry: BaseMetaEntry) -> None:
        self.entries.append(entry)

它存的是"任务/子任务之间想共享的额外元数据"。目前主要一种条目是 ActionSubtaskMetaEntry——把一次 ReAct 子任务的 thought(思考)、actions(动作 JSON)、answer(经记忆处理后的回答)记下来(memory/meta/action_subtask_meta_entry.py:9-21)。

这几块其实是串起来的:回看 §4.3,process_output 存完工件后,就会把一条 ActionSubtaskMetaEntry 写进 subtask.structure.meta_memory(task_memory.py:87-94)。这样下游子任务不仅能凭取件号取回原物,还能读到"当初是哪一步、基于什么思考产生的它"。

条目基类 BaseMetaEntry 本身是空的抽象壳(base_meta_entry.py:11),留作扩展点。

6. Artifact 类型系统​

6.1 BaseArtifact:统一契约​

所有 Artifact 都继承 BaseArtifact,它规定了每个"信封"必须具备的字段与行为(artifacts/base_artifact.py:16-57):

成员作用
value真数据(子类各自指定类型)
name名字,默认等于 id——在 Task Memory 里它就是取件号(namespace)
meta附加元数据字典
reference可选来源引用
to_text()抽象方法,子类必须实现:把自己变成给 LLM 看的文本
to_bytes() / __str__ / __bool__ / __len__默认实现,基于 value/to_text

一句话:只要是 Artifact,就一定能 to_text()——这正是记忆能把任意产物拼进提示词、工具能统一回传结果的前提。

6.2 各类型的职责​

Artifactvalue 类型职责 / 在部件间的角色文件:符号
TextArtifactstr最常见的文本载体;可 generate_embedding / token_count,能被 + 拼接;是 Task Memory 走向量库那条线的输入text_artifact.py:14
JsonArtifactJSON 值存结构化数据,构造时自动把值规整成 JSON 兼容形式,to_text 出 json.dumpsjson_artifact.py:13
ListArtifactSequence[Artifact]一组 Artifact 的容器;child_type/is_type/has_items 让别人问"里面装的是啥类型"——Task Memory 就靠它决定分派list_artifact.py:15
BlobArtifactbytes任意二进制;有 base64/mime_type;是 BlobArtifactStorage 的存储单位blob_artifact.py:10
ImageArtifactbytes(继承 Blob)图片;加 format/width/height,mime_type 出 image/<format>,to_text 出一句描述而非乱码image_artifact.py:12
ActionArtifactToolAction表示"LLM 决定调某个工具"这个动作本身;是 ReAct 循环里流动的数据action_artifact.py:13
ErrorArtifactstr表示要传达给 LLM 的错误;可带原始 exceptionerror_artifact.py:8
InfoArtifactstr传达提示性信息(如"无结果""已存入记忆")——§4.3 的取件号就是它info_artifact.py:8
BooleanArtifactbool布尔;parse_bool 能从 "true"/"false" 字符串解析boolean_artifact.py:8
GenericArtifact任意 T逃生舱:装不进上面任何类型时兜底generic_artifact.py:12

仓库里还有 AudioArtifact / ModelArtifact / 各种 *UrlArtifact(artifacts/ 目录),职责同理,不逐一展开。

6.3 一个能看出设计意图的细节​

TextArtifact.__bool__ 特意用了 bool(self.value.strip())(text_artifact.py:22-23):纯空白的文本被判为"假"。这不是随手写的——回看 §4.3,process_output 里 if output_artifact: 靠的就是这个真值判断:一份只有空白的工具输出会被当成"空",直接走 InfoArtifact("tool output is empty") 而不去占用记忆。信封的 __bool__ 语义,直接影响了记忆的存储决策。

7. 巧妙之处(可借鉴)​

  • 取件号复用 name,零额外机制。 Task Memory 的 namespace 直接就是 artifact.name(task_memory.py:71),而 name 默认等于 id(base_artifact.py:34-38)。不用另造一套 ID 体系,信封天生自带钥匙。

  • 分派表用类型当 key,扩展只需加一行。 artifact_storages: dict[type, BaseArtifactStorage](task_memory.py:27-35)。想支持新类型的 off-prompt 存储,注册一条 NewArtifact: NewStorage() 即可,get_storage_for 的 isinstance 逻辑不用改。

  • autoprune 是"试着塞、塞不下退一条",而非拍脑袋截断。 它用真实 tokenizer 反复量,保证既不溢出又尽量多带历史(base_conversation_memory.py:87-105)。

  • 摘要记忆"发给模型的"和"留着看的"分离。 SummaryConversationMemory 里 self.runs 永远是全量原文,摘要只作用于 to_prompt_stack(summary_conversation_memory.py:20-45 的文档串已点明)。检视与省 token 两不误。

8. 边界与局限​

  • BlobArtifactStorage 是纯进程内存字典(blob_artifact_storage.py:11),进程一停就没了;要持久化二进制得自己换 storage 实现。

  • 同一 namespace 不能跨 storage 混用。 一旦某个名字先存了文本、又来个二进制想用同名,store_artifact 直接返回 ErrorArtifact(task_memory.py:105-106)。

  • ListArtifact 的分派只看第一个元素(task_memory.py:55-58)。混合类型的列表,后面元素的类型不参与决策——异构列表要当心。

  • 摘要靠 LLM,可能失败。 summarize_runs 把异常吞掉、退回旧摘要并记 log(summary_conversation_memory.py:96-107);摘要失败时是静默降级,不会中断主流程,但也意味着那批老 Run 没被压缩。

  • 对话记忆的注入位置写死在 index 1(system 之后)(prompt_task.py:142-144),不是可配置项。

9. 横向对比​

同为 agent 框架,"把工具大输出留在框架内"这件事各家做法不同:Griptape 的 off-prompt 是显式类型分派 + 取件号的一等公民机制;不少框架则靠开发者自己在工具里做截断/摘要,或把中间产物塞进外部 KV。Griptape 的取舍是:用统一 Artifact 类型系统把"数据是什么类型"变成框架能感知的信息,从而让"该不该进提示词、该存哪"成为可自动决策的事,而不是每个工具各写一遍。

对话记忆侧,ConversationMemory(全量) vs SummaryConversationMemory(摘要)的二选一,对应了长对话里"保真"与"省 token"的经典权衡,与其它框架的滑窗/摘要策略同源。

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

主题文件路径符号名
对话记忆基类(autoload/max_runs/autoprune、注入)griptape/memory/structure/base_conversation_memory.pyBaseConversationMemory, add_run, after_add_run, add_to_prompt_stack
全量对话记忆griptape/memory/structure/conversation_memory.pyConversationMemory, try_add_run, to_prompt_stack
摘要对话记忆griptape/memory/structure/summary_conversation_memory.pySummaryConversationMemory, try_add_run, summarize_runs, unsummarized_runs
一轮问答griptape/memory/structure/run.pyRun
per_structure 落记忆griptape/structures/structure.pyafter_run(176-185)
per_task 落记忆 / 历史注入位置griptape/tasks/prompt_task.pyafter_run(198-206), prompt_stack(142-144)
任务记忆(off-prompt 核心)griptape/memory/task/task_memory.pyTaskMemory, get_storage_for, process_output, store_artifact, load_artifacts, find_input_memory
存储基类griptape/memory/task/storage/base_artifact_storage.pyBaseArtifactStorage, store_artifact, load_artifacts, can_store
文本存储(→向量库,见 06)griptape/memory/task/storage/text_artifact_storage.pyTextArtifactStorage
二进制存储(内存字典)griptape/memory/task/storage/blob_artifact_storage.pyBlobArtifactStorage
off-prompt 触发点(见 04)griptape/tools/base_tool.pyafter_run(189)
元记忆griptape/memory/meta/meta_memory.pyMetaMemory, add_entry
元记忆条目griptape/memory/meta/action_subtask_meta_entry.pyActionSubtaskMetaEntry
Artifact 基类契约griptape/artifacts/base_artifact.pyBaseArtifact, to_text, name, value
各 Artifact 类型griptape/artifacts/*.pyTextArtifact, JsonArtifact, ListArtifact, BlobArtifact, ImageArtifact, ActionArtifact, ErrorArtifact, InfoArtifact, BooleanArtifact, GenericArtifact