使用内存和压缩构建可靠的智能体

来源: https://developers.openai.com/cookbook/examples/agents_sdk/building_reliable_agents_memory_compaction 抓取时间: 2026-07-21 16:19:32

本指南展示如何使用 OpenAI Agents SDK 为综合合规调查构建证据审查智能体。 您将从一个简单的沙箱智能体开始，然后添加两个可靠性原语：

• 压缩 通过保留后续轮次所需的状态同时减小上下文大小，即使在有限上下文窗口的情况下也能支持长时间运行的对话。
• 内存 让未来的沙箱智能体运行可以重用之前运行的工作流经验，而无需重播之前的每一轮。

这种可靠性模式非常简单：压缩帮助当前运行继续进行，内存帮助后续运行以有用的工作流指导开始，生成的备忘录仍然是调查的人工审查事实来源。

参考：

• OpenAI Agents SDK
• Agents SDK sessions
• Agents SDK sandboxing

用例：合规调查的证据审查智能体

合规团队正在调查供应商例外是否遵循了内部政策。证据以一小部分文件的形式到达：政策语言、例外说明、审计观察、批准记录和补救计划。

智能体的工作不是成为调查记录。它的工作是帮助审查者浏览证据，跟踪发生的变化，并编写一份简洁的备忘录，将支持的发现与未解决的问题分开。

这使得该示例对于内存和压缩非常有用，因为调查具有三个在实际工作中出现的特征：

• 记录随时间变化。 后续文档可能会缩小或取代先前的假设。
• 对话可能会变得长时间运行。 审查者可能会提出后续问题，请求修订，并稍后返回相同的工作。
• 最终工件需要来源。 备忘录应引用证据并保留不确定性，而不是将审查简化为自信但无支持的结论。

此模式适用场景

虽然本笔记本使用合规审查，但相同的模式适用于知识工作者审查不断演变的上下文并生成人类可审计工件的任何地方。

良好的适用场景包括：

• 客户支持团队将新政策更新应用于未解决的升级。
• 安全团队审查事件证据并编写事件摘要。
• 财务团队核对政策、批准和审计备注中的例外情况。
• 产品团队在新产品发布或模型发布后更新竞争定位。
• 法律或采购团队审查合同、电子邮件和批准历史。
• 并购团队吸收新的业务规则、操作程序和尽职调查笔记。

在每种情况下，压缩随着上下文的增长保持主动审查可行，内存携带可重用的工作流经验，最终工件仍然是审查的输出。

您将构建什么

用例是合规证据审查。团队随着时间的推移收到政策文档、例外说明、审计发现、批准和补救计划。智能体帮助审查证据，保留不确定性，并生成带有引用的简洁备忘录。

您将构建：

1. 一个带有文件夹结构、清单和输出目录的综合证据工作区。
2. 一个可以检查文件和编写备忘录的简单 SandboxAgent。
3. 一个用于长时间运行工作的压缩检查点。
4. 用于可重用工作流经验的 SDK 内存生成。
5. 一个结合使用沙箱工具、压缩、内存和生成工件的组合运行。

您可以在不进行模型调用的情况下检查笔记本，因为 RUN_AGENT 默认为 False。仅当您想要执行实时沙箱工作流时才设置 RUN_AGENT = True。

目录

• 用例
• 此模式适用场景
• 前提条件
• 设置
• 在此笔记本中使用 Agents SDK
• 内存与压缩
• 文件夹结构和清单
• 准备一个小型证据工作区
• 步骤1：从简单的智能体配置开始
• 步骤2：添加压缩
• 步骤3：附加内存
• 步骤4：同时使用压缩和内存运行
• 检查生成的工件

前提条件

要运行实时智能体工作流，您需要：

• Python 3.10 或更高版本。
• openai-agents 包。
• 可用作 OPENAI_API_KEY 的 OpenAI API 密钥。
• UnixLocalSandboxClient 的本地类 Unix 环境。笔记本使用从沙箱 Manifest 创建的综合文件，因此不需要外部数据集。

笔记本可以安全地在没有凭据的情况下检查，因为 RUN_AGENT 默认为 False。仅当您想要执行模型支持的沙箱运行时才设置 RUN_AGENT = True。

# Install or upgrade the OpenAI Agents SDK.
%pip install --upgrade openai-agents

设置

笔记本将所有综合文件写入 examples/agents_sdk/.tmp/evidence_review_memory_compaction/ 下。它不需要外部数据。

默认情况下，跟踪被禁用，因为一些组织使用零数据保留（ZDR），其中跟踪摄入可能会被阻止。对于综合数据或非 ZDR 环境，您可以设置 ENABLE_TRACING = True 以在开发时检查跟踪。

from __future__ import annotations

import json
import os
import textwrap
from pathlib import Path

RUN_AGENT = False
MODEL = "gpt-5.5"
COMPACTION_MODEL = "gpt-5.4-mini"
WORKFLOW_NAME = "evidence-review-memory-compaction"
FORCE_COMPACTION_CHECKPOINT = True
DISABLE_TRACING = True

if DISABLE_TRACING:
    os.environ["OPENAI_AGENTS_DISABLE_TRACING"] = "true"
else:
    os.environ.pop("OPENAI_AGENTS_DISABLE_TRACING", None)

if RUN_AGENT and not os.environ.get("OPENAI_API_KEY"):
    raise RuntimeError("Set OPENAI_API_KEY before running the live sandbox workflow.")

print({
    "run_agent": RUN_AGENT,
    "model": MODEL,
    "compaction_model": COMPACTION_MODEL,
    "force_compaction_checkpoint": FORCE_COMPACTION_CHECKPOINT,
    "tracing_disabled": DISABLE_TRACING,
})

在此笔记本中使用 Agents SDK

沙箱智能体 是一个在受控工作区运行的 Agents SDK 智能体。在本笔记本中，该工作区包含综合证据文件、manifest.csv、输出文件夹和 SDK 生成的内存文件。

沙箱为智能体提供了一个有界的地方来检查文件和编写工件。应用程序不是将每个文档粘贴到提示中，而是创建一个工作区，并让智能体使用以下功能：

• Filesystem() 用于读取和写入工作区文件。
• Shell() 用于列出文件、检查文档和跨批次搜索。
• Compaction() 用于在活动上下文增长时支持长时间运行的审查。
• Memory() 用于为未来的沙箱智能体运行存储可重用的工作流经验。

备忘录仍然是人工审查的工件。工具帮助智能体工作，压缩帮助它继续，内存帮助未来的运行改进，生成的工件保存可审查的输出。

内存与压缩

一个有用的方法来区分这些概念是问每个概念被允许保留什么。

对于本笔记本，合规备忘录是调查输出的事实来源。内存有意限定为审查者偏好和工作流习惯，例如首先使用清单、保留不确定性以及保持被取代的假设可见。

文件夹结构和清单

智能体在一个小型文件工作区中工作。文件夹结构有意简单：证据文件按批次分组，生成的输出进入 outputs/ 下，清单为智能体提供可用文档的紧凑映射。

清单功能

在 Agents SDK 中，Manifest 是沙箱智能体的新会话工作区契约。它描述了新沙箱会话启动时应存在的文件、目录、挂载、环境、用户、组和相关工作区配置。

本地 SDK 实现定义了这些核心字段：

清单条目路径应相对于工作区。避免绝对路径和 .. 转义，以便同一智能体可以在 Unix 本地、Docker 和托管沙箱提供者之间移动。

文件夹和清单最佳实践

• 将源文档、清单、帮助文件和输出目录放在 Manifest 中，而不是将大内容粘贴到提示中。
• 将较长的任务说明放在工作区文件中，如 README.md、task.md 或 AGENTS.md；保持智能体说明专注于行为和边界。
• 使用稳定的文档 ID 和机器可读的清单文件，以便生成的备忘录可以引用来源，审查者可以检查返回证据的路径。
• 让 Memory() 管理其自己的内存工件。默认情况下，沙箱内存使用工作区下的 memories/ 和 sessions/。
• 将生成的工件保存在 outputs/ 下，以便应用程序可以在运行后检查、复制、验证或存档它们。
• 保持挂载范围狭窄。如果挂载数据室，仅挂载智能体应该读取或写入的内容。
• 将密钥视为由应用程序或沙箱提供者注入的运行时配置，而不是作为提示文本或提交的清单内容。
• 对于教程，首选小型的综合 File(...) 或 Dir(...) 条目，然后切换到 LocalDir、GitRepo 或存储挂载以用于生产规模的数据集。

# This is a visual preview of the sandbox workspace structure.
# The next cell builds the actual Manifest entries manually.
WORKSPACE_TREE = """
/workspace/
  README.md
  manifest.csv
  docs/
    batch_1/
    batch_2/
    batch_3/
  outputs/
  memories/    # Generated by Memory()
  sessions/    # Generated by Memory()
""".strip()

print(WORKSPACE_TREE)

准备一个小型证据工作区

Manifest 描述新沙箱工作区中的起始文件。对于本教程，工作区包括：

• 一个 manifest.csv，按批次和文档 ID 列出文档，
• 三个小文档批次，
• 审查备忘录的输出目录。

我们稍后附加的唯一内存原语是 SDK 的 Memory() 功能。调查结果保留在生成的审查者备忘录中，在那里可以引用和检查它们。

from agents.sandbox import Manifest
from agents.sandbox.entries import Dir, File


def workspace_file(text: str) -> File:
    return File(content=textwrap.dedent(text).strip().encode("utf-8") + b"\n")


def build_evidence_manifest() -> Manifest:
    return Manifest(
        entries={
            "README.md": workspace_file(
                """
                # Evidence Review Workspace

                Review the documents in batch order. Cite document IDs from
                `manifest.csv` when making findings. Write the final memo to
                `outputs/compliance_review_memo.md`.
                """
            ),
            "manifest.csv": workspace_file(
                """
                doc_id,batch,path,description
                ACME-B1-001,1,docs/batch_1/payment_policy.txt,Baseline payment policy
                ACME-B1-002,1,docs/batch_1/vendor_exception.txt,Vendor exception note
                ACME-B2-001,2,docs/batch_2/audit_followup.txt,Audit follow-up request
                ACME-B2-002,2,docs/batch_2/approval_thread.txt,Approval clarification
                ACME-B3-001,3,docs/batch_3/remediation_plan.txt,Remediation plan
                """
            ),
            "docs/batch_1/payment_policy.txt": workspace_file(
                """
                doc_id: ACME-B1-001
                ACME requires two approvals for payments over $50,000. Exceptions must
                be logged with Finance Ops and reviewed within five business days.
                """
            ),
            "docs/batch_1/vendor_exception.txt": workspace_file(
                """
                doc_id: ACME-B1-002
                A vendor onboarding exception was approved verbally for Northwind
                Logistics because the renewal was time-sensitive. The note does not show
                a Finance Ops log entry.
                """
            ),
            "docs/batch_2/audit_followup.txt": workspace_file(
                """
                doc_id: ACME-B2-001
                Internal Audit asked Finance Ops to confirm whether Northwind Logistics
                received post-approval review. The request says missing exception logs
                should be treated as a control gap until resolved.
                """
            ),
            "docs/batch_2/approval_thread.txt": workspace_file(
                """
                doc_id: ACME-B2-002
                The approval thread says Legal approved the vendor exception, but Finance
                Ops approval was still pending when the payment was released.
                """
            ),
            "docs/batch_3/remediation_plan.txt": workspace_file(
                """
                doc_id: ACME-B3-001
                The remediation plan requires Finance Ops to reconcile all verbal vendor
                exceptions from Q4 and add retrospective control attestations.
                """
            ),
            "outputs": Dir(),
        }
    )


manifest = build_evidence_manifest()
print(f"Workspace entries: {len(manifest.entries)}")

步骤1：从简单的智能体配置开始

首先，构建没有内存或压缩的智能体。目标是在添加原语之前使基线行为清晰。

一个微妙的要点：SandboxAgent 默认值可以包括内置功能。为了使这个基线明确，传递您想要的确切功能列表。在这里，我们只包括智能体检查文件和写入工件所需的工作区工具：Filesystem() 和 Shell()。我们有意不附加 Compaction() 或 Memory()。

• Filesystem() 为沙箱智能体提供面向文件的工作区访问，以便它可以读取分阶段的证据并编写备忘录工件。在 沙箱智能体指南 中，功能被描述为将沙箱原生行为和工具附加到 SandboxAgent 的方式。
• Shell() 允许智能体使用终端命令检查工作区，例如列出文件、打开证据文档和跨批次搜索术语。沙箱智能体指南指出，Shell() 是默认功能之一，Shell 工具指南 解释说，shell 为模型提供了用于托管或本地执行的终端环境。
• 对于此基线，这两个功能就足够了：Filesystem() 处理工作区读取和写入，而 Shell() 处理确定性检查和搜索。内存和压缩仅在基线工具清晰后添加。

from agents.sandbox import SandboxAgent
from agents.sandbox.capabilities import Filesystem, Shell

BASELINE_INSTRUCTIONS = """
You are an evidence review agent for a compliance investigation.

Review documents in batch order. Keep these boundaries clear:
- Cite document IDs from `manifest.csv` for each finding.
- If evidence is incomplete, record an open question instead of guessing.
- Write a concise reviewer memo to `outputs/compliance_review_memo.md`.
- Use the generated memo as the reviewer-facing investigation artifact.
""".strip()


def build_baseline_agent() -> SandboxAgent:
    return SandboxAgent(
        name="Evidence Review Agent",
        model=MODEL,
        instructions=BASELINE_INSTRUCTIONS,
        default_manifest=build_evidence_manifest(),
        capabilities=[
            Filesystem(),
            Shell(),
        ],
    )


baseline_agent = build_baseline_agent()
print([type(capability).__name__ for capability in baseline_agent.capabilities])

from agents import Runner
from agents.run import RunConfig
from agents.sandbox import SandboxRunConfig
from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient

BASELINE_TASK = """
Review Batch 1 only, then draft `outputs/compliance_review_memo.md` for a
compliance reviewer. Include cited findings and open questions.
""".strip()


async def _read_text_file(session, path: str) -> str | None:
    try:
        handle = await session.read(Path(path))
    except Exception as exc:
        if "NotFound" not in type(exc).__name__:
            raise
        return None
    try:
        return handle.read().decode("utf-8", errors="replace")
    finally:
        handle.close()


async def _list_workspace_files(session) -> str:
    result = await session.exec(
        "find outputs memories sessions -maxdepth 4 -type f 2>/dev/null | sort || true",
        timeout=30,
    )
    return result.stdout.decode("utf-8", errors="replace").strip()


async def _read_memory_artifacts(session) -> dict[str, str]:
    memory_artifacts = {}
    for path in ["memories/MEMORY.md", "memories/memory_summary.md"]:
        text = await _read_text_file(session, path)
        if text:
            memory_artifacts[path] = text
    return memory_artifacts


async def run_in_unix_sandbox(agent: SandboxAgent, task: str, *, sdk_session=None) -> dict[str, object]:
    client = UnixLocalSandboxClient()
    session = await client.create(manifest=agent.default_manifest)
    try:
        await session.start()
        result = await Runner.run(
            agent,
            task,
            max_turns=12,
            run_config=RunConfig(
                sandbox=SandboxRunConfig(session=session),
                workflow_name=WORKFLOW_NAME,
                tracing_disabled=DISABLE_TRACING,
            ),
            session=sdk_session,
        )

        compaction_checkpoint = None
        if sdk_session is not None and FORCE_COMPACTION_CHECKPOINT:
            compaction_checkpoint = await force_compaction_checkpoint(sdk_session)

        # Memory generation runs as a sandbox pre-stop hook. Flush it before reading artifacts
        # so `memories/MEMORY.md` and `memories/memory_summary.md` are available here.
        await session.run_pre_stop_hooks()

        memo = await _read_text_file(session, "outputs/compliance_review_memo.md")
        files = await _list_workspace_files(session)
        memory_artifacts = await _read_memory_artifacts(session)

        return {
            "result": result,
            "final_output": str(result.final_output),
            "memo": memo,
            "workspace_files": files,
            "memory_artifacts": memory_artifacts,
            "compaction_checkpoint": compaction_checkpoint,
        }
    finally:
        await session.aclose()


if RUN_AGENT:
    baseline_run = await run_in_unix_sandbox(baseline_agent, BASELINE_TASK)
    print(baseline_run["final_output"])
    print("\n----- END AGENT OUTPUT -----")
else:
    print("RUN_AGENT is False. Baseline agent is configured but not executed.")

步骤2：添加压缩

压缩用于长时间运行的工作。随着对话的增长，压缩会减小上下文大小，同时保留后续轮次所需的状态。有三种有用的思考方式：

1. 使用 Compaction() 的自动压缩：附加功能，让 SDK 在上下文压力需要时进行压缩。
2. 使用 StaticCompactionPolicy 的基于阈值的压缩：为您希望具有更可预测的上下文大小行为的环境设置显式阈值。
3. 使用 OpenAIResponsesCompactionSession.run_compaction({"force": True}) 的强制检查点压缩：在应用程序定义的阶段边界进行压缩，例如在主要审查阶段之后和下一个证据批次之前。

本笔记本使用强制检查点，因为综合数据集故意很小。在生产中，自动压缩通常是最简单的起点，而当您想要更严格的操作策略时，基于阈值的压缩很有用。

最佳实践

• 在有意义的工作流边界进行压缩，而不是在每轮之后。
• 保留足够的工作状态，以便下一阶段有意义。
• 将引用的事实保留在生成的工件中，而不仅仅是压缩的对话状态。

from agents.sandbox.capabilities import Compaction, StaticCompactionPolicy


def build_compaction_agent(*, demo_threshold: int | None = None) -> SandboxAgent:
    if demo_threshold is None:
        compaction = Compaction()
    else:
        compaction = Compaction(policy=StaticCompactionPolicy(threshold=demo_threshold))

    return SandboxAgent(
        name="Evidence Review Agent with Compaction",
        model=MODEL,
        instructions=(
            BASELINE_INSTRUCTIONS
            + "\n\nWhen context is compacted, preserve the current batch, cited facts, open "
            "questions, artifact paths, and unresolved reviewer concerns."
        ),
        default_manifest=build_evidence_manifest(),
        capabilities=[
            Filesystem(),
            Shell(),
            compaction,
        ],
    )


compaction_agent = build_compaction_agent()
threshold_compaction_agent = build_compaction_agent(demo_threshold=8_000)
print({
    "automatic": [type(capability).__name__ for capability in compaction_agent.capabilities],
    "threshold_policy": [type(capability).__name__ for capability in threshold_compaction_agent.capabilities],
})

压缩如何被触发

使用 Compaction() 功能，当活动上下文增长到足够大时，服务器端压缩有资格运行。这是当前的默认行为：附加功能，让 SDK 管理上下文压力。

对于小型教程，自动压缩可能很难看到，因为运行可能永远不会接近模型上下文限制。较低的 StaticCompactionPolicy 可以提供帮助，但它仍然取决于渲染的上下文超过阈值。

对于小型证据集，强制检查点是最清晰的操作模式。OpenAIResponsesCompactionSession 包装器存储会话历史记录，并允许应用程序在阶段边界调用 run_compaction({"force": True})。这使得压缩可见，而不会膨胀证据集。

from agents.memory import OpenAIResponsesCompactionSession, SQLiteSession


def build_compaction_session() -> OpenAIResponsesCompactionSession:
    underlying = SQLiteSession("evidence_review_session.sqlite")
    return OpenAIResponsesCompactionSession(
        session_id="evidence-review-demo",
        underlying_session=underlying,
        model=COMPACTION_MODEL,
        compaction_mode="input",
    )


async def force_compaction_checkpoint(session: OpenAIResponsesCompactionSession) -> dict[str, int]:
    items_before = await session.get_items()
    await session.run_compaction({"force": True, "compaction_mode": "input"})
    items_after = await session.get_items()
    return {"items_before": len(items_before), "items_after": len(items_after)}


print("Compaction session helper defined. The final run uses it to show an explicit phase checkpoint.")

步骤3：附加内存

内存用于跨运行重用。在此示例中，内存应捕获工作流经验，而不是调查事实。

良好的内存候选包括：

• 在审查基于文件的证据工作区时，首先使用清单。
• 在备忘录中保留不确定性，而不是猜测。
• 当后续证据缩窄假设时，保持早期假设可见。

不良的内存候选包括：

• “Northwind Logistics 违反了政策。”
• “ACME 的财务运营流程存在缺陷。”
• 属于备忘录的任何特定案例结论。

最佳实践

• 将内存用于稳定的流程经验和用户偏好。
• 将特定案例的事实保留在经过审查的工件中，例如备忘录。
• 在未来的运行中依赖生成的内存之前，请检查它。

from agents.sandbox import MemoryGenerateConfig
from agents.sandbox.capabilities import Memory

MEMORY_GENERATION_PROMPT = """
Store reusable workflow lessons only.
Do not store ACME-specific compliance findings, document facts, evidence citations,
or memo conclusions. Those belong in outputs/compliance_review_memo.md.
Memory should help future evidence-review workflows behave better; it should not
become a second investigation record.
""".strip()


def workflow_memory() -> Memory:
    return Memory(
        generate=MemoryGenerateConfig(
            extra_prompt=MEMORY_GENERATION_PROMPT,
        )
    )


def build_memory_agent() -> SandboxAgent:
    return SandboxAgent(
        name="Evidence Review Agent with Memory",
        model=MODEL,
        instructions=BASELINE_INSTRUCTIONS,
        default_manifest=build_evidence_manifest(),
        capabilities=[
            Filesystem(),
            Shell(),
            workflow_memory(),
        ],
    )


memory_agent = build_memory_agent()
print([type(capability).__name__ for capability in memory_agent.capabilities])

步骤4：同时使用压缩和内存运行

现在组合这些部分：

• Filesystem() 和 Shell() 让智能体导航证据工作区。
• Compaction() 随着上下文的增长保持主动审查可行。
• Memory() 在运行后捕获可重用的工作流经验。
• 最终备忘录仍然是调查工件。

下面的任务要求智能体审查综合证据，编写备忘录，然后回读备忘录以验证它保留了所需的结构和不确定性。

FINAL_REVIEW_TASK = """
Review all three document batches in order.

For each batch:
1. Read the manifest and relevant documents.
2. Preserve cited findings and uncertainty in your working notes and final memo.
3. Preserve any superseded or narrowed assumption instead of silently deleting it.

After Batch 3, write `outputs/compliance_review_memo.md` with:
- executive summary,
- cited findings table,
- open questions,
- recommended next steps for the reviewer.

Reviewer preference for future runs: keep the memo concise, preserve uncertainty
instead of guessing, and separate reusable workflow lessons from document-specific
compliance findings.
""".strip()


def build_reliable_evidence_agent() -> SandboxAgent:
    return SandboxAgent(
        name="Reliable Evidence Review Agent",
        model=MODEL,
        instructions=(
            BASELINE_INSTRUCTIONS
            + "\n\nUse compaction as working context. Use SDK memory for reusable "
            "workflow lessons across runs. Do not treat memory as the system of record "
            "for ACME-specific findings; those belong in the cited memo artifact."
        ),
        default_manifest=build_evidence_manifest(),
        capabilities=[
            Filesystem(),
            Shell(),
            Compaction(),
            workflow_memory(),
        ],
    )


reliable_agent = build_reliable_evidence_agent()
print([type(capability).__name__ for capability in reliable_agent.capabilities])

EXAMPLE_OUTPUT = """
# Compliance Review Memo

## Executive Summary

The current record supports a control-gap finding for the Northwind Logistics
vendor exception, not a final conclusion that policy was intentionally violated.
The strongest evidence is that ACME required two approvals and Finance Ops
logging for payment exceptions, while the Northwind exception appears to have
been released before Finance Ops approval was complete.

Later evidence narrows the initial concern. The record no longer points only to
an undocumented verbal exception; it now points to a specific process weakness:
Legal approval may have been obtained, but Finance Ops review and exception-log
reconciliation remained incomplete at the time of release.

## Cited Findings

| Finding | Support | Status |
|---|---|---|
| Payments over $50,000 required two approvals, and exceptions had to be logged with Finance Ops. | ACME-B1-001 | Supported |
| The Northwind Logistics exception was approved verbally, but the initial note does not show a Finance Ops log entry. | ACME-B1-002 | Supported |
| Internal Audit treated missing exception logs as a control gap until resolved. | ACME-B2-001 | Supported |
| The approval thread indicates Legal approved the exception, but Finance Ops approval was still pending when payment was released. | ACME-B2-002 | Supported |
| The remediation plan requires Finance Ops to reconcile verbal vendor exceptions from Q4 and add retrospective control attestations. | ACME-B3-001 | Supported |

## Open Questions

- Was Finance Ops approval completed after the payment release, and if so, when?
- How many other Q4 verbal vendor exceptions lack retrospective attestations?
- Did any compensating control apply to the Northwind payment before remediation began?

## Recommended Next Steps

1. Reconcile Northwind Logistics against the Finance Ops exception log and payment-release timestamp.
2. Pull the full Q4 population of verbal vendor exceptions into the same review workflow.
3. Classify the issue as a control gap unless later evidence shows timely Finance Ops approval or an approved compensating control.
""".strip()

if not RUN_AGENT:
    print("RUN_AGENT is False, so this cell shows an example memo shape rather than running the model.\n")
    print(EXAMPLE_OUTPUT)

检查生成的工件

最终的智能体响应很有用，但当您检查沙箱运行生成的文件时，可靠性模式会变得更加清晰。本节使通常隐藏的状态可见：

• outputs/compliance_review_memo.md 中面向审查者的备忘录，
• 生成的 SDK 内存文件，如 memories/MEMORY.md 和 memories/memory_summary.md，
• 运行产生的工作区文件，包括会话日志。

生成的内存工件不是合规备忘录，不应被视为调查事实。它是可重用的工作流内存。Task Group 标题是内存系统自己的分组标签，内存生成器使用 MemoryGenerateConfig.extra_prompt 进行引导，以便它存储工作流经验而不是 ACME 特定的发现。

如果 RUN_AGENT = False，本节显示预期的输出形状，而不是实时沙箱工件。

try:
    from IPython.display import Markdown, display
except ImportError:
    Markdown = None
    display = None

memo_text = final_run.get("memo") if "final_run" in globals() else None
workspace_files = final_run.get("workspace_files") if "final_run" in globals() else None
memory_artifacts = final_run.get("memory_artifacts", {}) if "final_run" in globals() else {}
compaction_checkpoint = final_run.get("compaction_checkpoint") if "final_run" in globals() else None

print("Generated workspace files:")
print(workspace_files or "No generated workspace files were captured.")

print("\nMemory and compaction configuration:")
print({
    "final_agent_capabilities": [type(capability).__name__ for capability in reliable_agent.capabilities],
    "sandbox_compaction": "Compaction() is attached to the final agent for automatic context management",
    "sandbox_memory": "Memory(generate=MemoryGenerateConfig(...)) is attached to the final agent",
    "memory_write_policy": "Store reusable workflow lessons, not ACME-specific compliance findings",
    "forced_checkpoint": "OpenAIResponsesCompactionSession.run_compaction({force: True}) after the final review",
    "compaction_model_for_checkpoint": COMPACTION_MODEL,
    "checkpoint_result": compaction_checkpoint,
})

if memory_artifacts:
    for path, text in memory_artifacts.items():
        heading = f"### Generated SDK memory artifact: `{path}`\n\n"
        explanation = (
            "This block is generated by the SDK `Memory()` primitive. It is reusable "
            "workflow memory, not the compliance memo and not the investigation system "
            "of record.\n\n"
            "```text\n----- BEGIN GENERATED SDK MEMORY ARTIFACT -----\n"
        )
        closing = "\n----- END GENERATED SDK MEMORY ARTIFACT -----\n```"
        if display is not None and Markdown is not None:
            display(Markdown(heading + explanation + text + closing))
        else:
            print(f"\nGenerated SDK memory artifact: {path}\n")
            print("----- BEGIN GENERATED SDK MEMORY ARTIFACT -----")
            print(text)
            print("----- END GENERATED SDK MEMORY ARTIFACT -----")
else:
    print("\nNo generated memory artifacts were captured. Set RUN_AGENT = True and rerun the final workflow.")

if memo_text:
    if display is not None and Markdown is not None:
        display(Markdown("## Generated memo: `outputs/compliance_review_memo.md`\n\n" + memo_text))
    else:
        print("\nGenerated memo:\n")
        print(memo_text)
else:
    print("\nNo memo was captured. Set RUN_AGENT = True and rerun the final workflow.")

常见陷阱

不要将 Memory() 视为未经审查的事实数据库。

内存应该帮助下一次运行记住如何工作。它不应该成为影子合规记录。如果结论很重要，请将其写入带有引用的经过审查的工件中。

结论

您现在拥有构建可靠的长时间运行智能体工作流的构建块：

• 用于受控文件访问的沙箱工作区。
• 帮助智能体跨文档路由的清单。
• 用于有限上下文窗口的压缩。
• 用于可重用工作流经验的内存。
• 作为经过审查的调查工件的生成备忘录。

主要的设计选择是职责分离：上下文帮助智能体工作，内存帮助未来的智能体更好地工作，经过审查的工件保存人们将依赖的事实。