构建具有记忆与压缩功能的可靠AI代理

本文面向有编程基础的开发者,深入讲解如何利用OpenAI Agents SDK构建具备记忆与压缩功能的可靠AI代理。通过合规调查证据审查的实战案例,演示了从简单沙盒代理到集成压缩与记忆的完整演进过程。文章详细解析了压缩与记忆的核心原理、适用场景及最佳实践,并提供了可执行的环境配置、代码实现与故障排查指南,帮助开发者掌握构建长时运行、可扩展AI代理的工程方法。

技术背景

在构建AI代理时,记忆与压缩是确保其可靠性的关键因素。AI代理需要记忆来存储对话历史、用户偏好和任务状态,没有记忆,代理将无法提供连贯的交互体验。然而,随着对话和上下文的增长,存储和检索成本会显著增加,甚至超出模型的上下文窗口限制。压缩策略通过减少冗余信息来优化上下文使用,而记忆机制则允许代理跨会话复用工作流经验,避免重复学习。本教程以合规调查证据审查为例,展示如何结合压缩与记忆构建可靠的长时运行代理,适用于客户支持、安全审查、财务对账等需要处理不断演变上下文并生成可审计产物的场景。

核心原理

压缩与记忆是两种不同的可靠性原语。压缩针对当前运行,当上下文增长时,通过总结或丢弃次要信息来减小上下文大小,保留后续轮次所需的状态,从而支持长时间运行的对话。记忆则针对未来运行,通过存储可复用的工作流经验(如审查偏好、操作流程)来改进后续代理的性能,而无需重放每个先前的轮次。在实现上,压缩可通过Compaction()能力自动触发,或通过StaticCompactionPolicy设置阈值,也可在阶段边界强制检查点。记忆通过Memory()能力生成,并可通过MemoryGenerateConfig引导存储内容,确保不存储具体案例事实,而是流程经验。

架构分析

系统架构围绕沙盒工作区展开,使用Manifest定义工作区初始文件结构,包括证据文档、清单和输出目录。代理通过Filesystem()和Shell()能力访问和操作文件,Compaction()管理上下文,Memory()生成记忆。整个流程分为四个步骤:基线代理(仅文件工具)、添加压缩、附加记忆、组合运行。最终产物是人工审查的备忘录,记忆仅存储工作流经验,不存储调查事实。关键设计决策包括:将任务说明放在工作区文件中而非提示词中,使用稳定的文档ID和清单文件,保持挂载范围狭窄,以及将机密作为运行时配置注入。

完整实战教程

构建具有记忆与压缩功能的可靠AI代理

在构建AI代理时,记忆与压缩是确保其可靠性的关键因素。本文将探讨如何通过有效的记忆管理和压缩策略,提升AI代理的性能和稳定性。

记忆的重要性

AI代理需要记忆来存储对话历史、用户偏好和任务状态。没有记忆,代理将无法提供连贯的交互体验。记忆可以分为短期记忆和长期记忆,短期记忆用于当前会话,长期记忆则用于跨会话的信息保留。

记忆的类型

  1. 短期记忆:存储当前会话中的上下文信息,如用户输入和代理响应。
  2. 长期记忆:存储跨会话的持久信息,如用户偏好和历史数据。

实现记忆的方法

压缩策略

随着记忆的增长,存储和检索成本会增加。压缩策略有助于减少记忆的冗余,提高效率。

最佳实践

结论

通过合理的记忆管理和压缩策略,可以构建出可靠且高效的AI代理。记忆提供了上下文,压缩则确保了系统的可扩展性。结合最佳实践,开发者可以创建出更智能、更可靠的AI代理。

本 Cookbook 展示了如何使用 OpenAI Agents SDK 为合成合规调查构建一个证据审查代理。

您将从简单的沙盒代理开始,然后添加两个可靠性原语:

可靠性模式很简单:压缩有助于当前运行继续,记忆有助于后续运行以有用的工作流指导开始,而生成的备忘录仍然是调查中经过人工审查的事实来源。

参考资料:

用例:合规调查的证据审查代理

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

代理的工作不是成为调查记录本身。它的工作是帮助审查者浏览证据,跟踪变化,并撰写一份简洁的备忘录,将已证实的发现与未解决的问题区分开来。

这使得该示例对记忆和压缩很有用,因为调查具有三个在现实工作中出现的特征:

此模式的适用场景

尽管本笔记本使用了合规审查作为示例,但同样的模式适用于任何知识工作者审查不断演变的上下文并生成可供人工审计的产物的场景。

以下场景尤为适合:

在每种情况下,压缩技术都能在上下文增长时保持活跃审查的可行性,记忆功能将可复用的工作流经验向前传递,而最终产物仍然是经过审查的输出。

你将构建什么

用例是合规证据审查。团队会随时间收到政策文件、异常说明、审计发现、批准和补救计划。该代理帮助审查证据、保留不确定性,并生成带有引用的简明备忘录。

你将构建:

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

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

目录

先决条件

要运行实时代理工作流,你需要:

笔记本可以在没有凭据的情况下安全检查,因为 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 生成的记忆文件。

沙盒为代理提供了一个有界的空间来检查文件和写入工件。应用程序不会将每个文档粘贴到提示中,而是创建一个工作区,让代理使用诸如以下功能:

备忘录仍然是人工审查的工件。工具帮助代理工作,压缩帮助它继续,记忆帮助未来的运行改进,生成的工件保存可审查的输出。

记忆与压缩

区分这两个概念的一个有用方法是问各自允许携带什么。

| 问题 | 压缩 | 记忆 |

|---|---|---|

| 它有助于什么? | 当上下文增长时,继续一个长时间运行的运行。 | 通过可重用的工作流经验改进未来的运行。 |

| 它总结什么? | 当前对话和工作状态。 | 值得重用的模式、偏好和过程经验。 |

| 它应该存储调查结论吗? | 不。它可以保留工作状态,但备忘录是经过审查的产物。 | 不。存储工作流经验,而不是具体案例的事实。 |

| 什么时候有用? | 审查中,尤其是在后续批次或后续轮次之前。 | 在重复审查类似证据工作流时。 |

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

文件夹结构与清单

代理在一个小型文件工作区中工作。文件夹结构刻意保持简单:证据文件按批次分组,生成的输出放在 outputs/ 下,清单为代理提供了可用文档的紧凑地图。

Manifest 特性

在 Agents SDK 中,Manifest 是沙箱代理的新会话工作区契约。它描述了新沙箱会话启动时应存在的文件、目录、挂载点、环境、用户、组以及相关的工作区配置。

本地 SDK 实现定义了以下核心字段:

| Manifest 字段 | 控制内容 | 在本 Cookbook 中的用法 |

|---|---|---|

| root | 工作区根路径。默认为 /workspace。 | 除非沙箱提供方期望不同的根路径,否则保持默认。 |

| entries | 要物化的文件、目录、本地文件、本地目录、仓库或挂载点。 | 将 README.md、manifest.csv、输入文档和 outputs/ 放在此处。 |

| environment | 沙箱启动时可用的环境变量。 | 仅用于非机密运行时配置。避免在提示词和提交的笔记本中保留凭据。 |

| users / groups | 支持这些的提供方的沙箱本地操作系统账户和组。 | 对于 Cookbook 通常不需要,对于生产隔离很有用。 |

| extra_path_grants | 额外的路径授权,特别适用于 Unix 本地工作流。 | 当沙箱需要对主机路径进行范围化读写访问时,谨慎使用。 |

| remote_mount_command_allowlist | 允许对远程挂载点执行的命令。 | 在挂载外部存储或数据室时保持狭窄。 |

Manifest 条目路径应为工作区相对路径。避免绝对路径和 .. 转义,以便同一代理可以在 Unix 本地、Docker 和托管沙箱提供方之间移动。

文件夹与清单最佳实践

# 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 描述了全新沙盒工作区中的起始文件。本教程的工作区包括:

我们稍后附加的唯一记忆原语是 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()。

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 步:附加记忆

记忆用于跨运行复用。在此示例中,记忆应捕获工作流程经验,而非调查事实。

好的记忆候选包括:

不好的记忆候选包括:

最佳实践
- 使用记忆存储稳定的流程经验和用户偏好。
- 将特定案例的事实保留在已审查的工件中,如备忘录。
- 在未来的运行中依赖生成的记忆之前,先检查它。
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:同时使用压缩和记忆运行

现在将各部分组合起来:

下面的任务要求代理审查合成证据,撰写备忘录,然后重新阅读备忘录以验证其保留了所需的结构和不确定性。

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)

检查生成的产物

最终的智能体响应很有用,但当你检查沙箱运行产生的文件时,可靠性模式会更加清晰。本节将通常隐藏的状态可视化:

生成的记忆产物不是合规备忘录,不应视为调查事实。它是可复用的工作流记忆。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() 视为未经审查的事实数据库。

Memory 应帮助下次运行记住如何工作,而不应成为影子合规记录。如果某个结论很重要,请将其写入带有引用的已审查工件中。

结论

现在你已经拥有了构建可靠的长时运行代理工作流的基础组件:

主要的设计选择是职责分离:上下文帮助代理工作,内存帮助未来的代理更好地工作,而审查后的产物则承载人们将依赖的事实。

常见问题

常见错误包括:将Memory()视为事实数据库,存储具体案例结论,导致记忆成为影子记录;未在阶段边界进行压缩,导致上下文溢出;忽略Manifest路径规范,使用绝对路径或..转义,导致跨沙箱提供方不可移植;未正确设置OPENAI_API_KEY或RUN_AGENT标志,导致无法运行实时代理。排查方法:检查RUN_AGENT和DISABLE_TRACING配置;验证Manifest条目路径为相对路径;使用session.run_pre_stop_hooks()确保记忆生成完成;检查memories/MEMORY.md内容是否符合预期。生产环境优化:使用自动压缩策略,设置合理的阈值;将记忆生成提示词明确限定为工作流经验;定期审查和清理记忆;使用LocalDir或GitRepo处理大规模数据集。

企业应用扩展

本模式可扩展至多种知识工作场景,如客户支持升级处理、安全事件摘要、财务异常对账、产品竞争分析、合同审查等。进一步优化方向包括:引入基于重要性评分的记忆过滤机制,实现更精细的压缩策略;结合向量数据库实现语义记忆检索;设计多代理协作架构,共享记忆与压缩状态;集成外部存储系统以持久化记忆和产物。此外,可探索使用StaticCompactionPolicy动态调整阈值,以及通过MemoryGenerateConfig定制记忆生成模板,以适应不同领域的工作流需求。

总结

本文面向有编程基础的开发者,深入讲解如何利用OpenAI Agents SDK构建具备记忆与压缩功能的可靠AI代理。通过合规调查证据审查的实战案例,演示了从简单沙盒代理到集成压缩与记忆的完整演进过程。文章详细解析了压缩与记忆的核心原理、适用场景及最佳实践,并提供了可执行的环境配置、代码实现与故障排查指南,帮助开发者掌握构建长时运行、可扩展AI代理的工程方法。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程