首页 / 文章

实践中的上下文工程:一个基于Claude Agent SDK的研究智能体

使用Claude Agent SDK,借助写入、选择、压缩和隔离模式构建一个Python研究智能体,然后在实际测试中将其与简单的基准模型进行对比。

发布者:REACTAPP.TOP阅读约 20 分钟4633 词

来源: Context Engineering in Practice: Building a Production AI Agent with the Claude Agent SDK

研究型智能体很少会遇到一个难以处理的指令。真正棘手的是大量指令的堆积。在一次调查过程中,上下文中会充斥着用户的指令、计划、搜索查询、下载的页面、工具结构、源文件元数据、未完成的陈述、助手的报告以及之前的对话内容。所有这些都在同时存在,导致有效证据不得不与智能体自身的待处理任务竞争注意力。

在这里,“编写、选择、压缩与隔离”框架便成了Claude Agent SDK上的实际可用Python架构。智能体通过一次次的决策逐步得到改进,而每一个边界都会引发同一个问题:模型此刻必须看到什么,又有什么可以排除在外?我们的目标是打造一个可实际运行的研究型智能体,而非仅仅列出SDK的功能清单。

构建过程分为十个步骤:

  1. 明确要构建的内容。
  2. 建立简单的基准版本。
  3. 撰写:将工作状态移出对话流程。
  4. 选择:构建有限的工作集。
  5. 压缩:使长时会话可恢复。
  6. 隔离:委托特定的调查任务。
  7. 在环境中隔离重型工具的输出结果。
  8. 在会话之间保留上下文信息。
  9. 组建经过上下文优化的智能体。
  10. 比较这两种架构。

将构建什么

面对某个问题时,该智能体会规划研究方案、检索原始资料、将具体的调查任务分配给辅助工具、记录证据,最终撰写一份附有引用的报告。示例任务是:比较实际应用中的各类长期记忆技术,并说明每种技术的成本。该任务需要相关论文、实现文档、代码库中的证据以及引用核查,因此难度足够大,能够真正起到测试作用。

主代理负责维护计划并撰写最终总结。经过整理的工作状态存储在文件中,检索机制限制了输入的知识量,而子代理则在各自的上下文中处理具体问题。这些功能均未内置在SDK中。SDK仅提供模型与工具循环,应用程序则决定状态存储的位置以及各边界之间允许传递的信息量。

第一张图展示了初始系统:一个主代理与工具、知识库、临时记录区以及子代理相连。后续的架构调整则改变了信息在这些连接之间的传递方式。

在运行任何模块之前,需先创建环境并安装指定的SDK版本。

# terminal
python3 -m venv .venv
source .venv/bin/activate
python -m pip install claude-agent-sdk==0.2.139
export ANTHROPIC_API_KEY="your-api-key"

该布局将代理配置与工作区操作、自定义工具、钩子以及测量功能分开。这种分离很重要,因为每种上下文操作只需修改某一边界,而无需创建全新的应用程序。

# code/
research_agent/
  config.py       # naive and engineered ClaudeAgentOptions
  hooks.py        # pre-compaction checkpoint
  metrics.py      # message-stream and context measurements
  runner.py       # repeated runs and comparison
  tools.py        # in-process MCP tools
  workspace.py    # scratchpad and bounded retrieval
knowledge/
  memory_approaches.json
tests/
CLAUDE.md

上下文工程始于确定责任归属。对话负责处理当前的决策,而文件、工具、检索系统以及子代理则负责管理那些暂时无需激活的内容。

简单的基准方案

最简版的 Agent SDK 程序会将控制流程交给 query() 函数。Claude 接收提示后判断是否需要使用工具,观察结果并重复该过程,直到得到答案或达到预设的限制次数。由于 SDK 已经提供了这样的循环结构,应用程序无需自行编写工具调用的解析与分发逻辑。

# research_agent/minimal.py
import asyncio
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query

QUESTION = "Compare approaches for long-term memory in production AI agents."

async def main() -> None:
    options = ClaudeAgentOptions(
        model="sonnet",
        allowed_tools=["WebSearch", "WebFetch"],
        permission_mode="dontAsk",
        max_turns=20,
    )
    async for message in query(prompt=QUESTION, options=options):
        if isinstance(message, ResultMessage):
            print(message.result or "")

asyncio.run(main())

那个简短的程序隐藏了某些内容:query()负责执行循环,但具体进入循环的内容仍由应用程序决定。research_agent/config.py中的可运行基础代码详细列出了这些选择。

# research_agent/config.py
def naive_options(*, run_root, server, model, max_budget_usd):
    tools = ["Read", "Glob", "Grep", "WebSearch", "WebFetch"]
    return ClaudeAgentOptions(
        cwd=run_root,
        model=model,
        tools=tools,
        allowed_tools=[*tools, "mcp__research__*"],
        permission_mode="dontAsk",
        mcp_servers={"research": server},
        strict_mcp_config=True,
        setting_sources=[],
        system_prompt={
            "type": "preset",
            "preset": "claude_code",
            "append": NAIVE_PROMPT,
        },
        env={"ENABLE_TOOL_SEARCH": "false"},
        max_turns=20,
        max_budget_usd=max_budget_usd,
    )

在基准配置中,没有写入工具、没有智能体工具、没有agents设置、没有钩子功能,也没有项目指令文件。负责人必须在单次对话中完成全部调查工作。将ENABLE_TOOL_SEARCH=false设置后,可避免MCP模式被延迟加载。系统提示会进一步明确要求智能体调用全语料库工具,将调查结果保留在对话中,既不写入工作区笔记也不进行任务委托。

全语料库工具被刻意设计得非常简单,因为问题在于它返回的内容,而非实现方式。

# research_agent/tools.py
@tool(
    "load_knowledge_corpus",
    "Return the entire local memory-research collection. Intended only for the naive baseline.",
    {},
    annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False),
)
async def load_knowledge(_: dict[str, Any]) -> dict[str, Any]:
    return _text_result(load_corpus(corpus_path))

当 Claude 发出请求时,SDK 会以包含 ToolResultBlock 的 UserMessage 形式输出结果。这里的“用户”指的是协议角色,而非真实的人类消息。该数据块会被加入传递给后续模型步骤的历史记录中,因此一次网络搜索、获取的三个页面以及完整的语料库都会生成类似这样的重复序列。

UserMessage(question)
AssistantMessage(ToolUseBlock: load_knowledge_corpus)
UserMessage(ToolResultBlock: entire corpus)
AssistantMessage(ToolUseBlock: WebSearch + WebFetch)
UserMessage(ToolResultBlock: raw search and page content)
AssistantMessage(final report)

那个流程本身并没有什么问题。它显得过于简单,是因为该应用从未将原始结果转化为经过整理的外部状态,也从未将其从当前处理流程中移除。

第二张图展示了这种简单设计如何在单个窗口中收集指令、工具定义、原始页面内容、分析结果以及对话历史。其扩展性源自应用自身的策略,而非SDK循环中的任何缺陷。

对于大多数Agent SDK配置而言,默认情况下工具搜索功能是开启的,也有相关文档记载了例外情况。因此,如果实验需要预先加载所有工具信息,就必须刻意关闭该功能作为基准对比。作为一种对照,这种简单设计很有用,因为它去除了后续步骤将要测试的所有上下文管理决策。

在花费时间和资源进行比较之前,先通过相同的配置发送了一个受限请求作为测试。测试结果显示,认证、路由以及模型解析功能均正常运行。

# Captured output: python scripts/smoke_test.py
subtype=success
result=OPENROUTER_OK
models=claude-sonnet-5

完整的基准测试共进行了三次,其结果与经过设计的试验结果一同出现在最终的对比中。第一次试验用于将累积效果具体化。

# Captured output: python scripts/show_naive_trial.py ../measurements/comparison.json
Naive trial 1
SDK result: success
Assistant steps: 4
Tool calls: 6
  WebFetch: 3
  WebSearch: 1
  mcp__research__load_knowledge_corpus: 1
  mcp__research__search_knowledge: 1
Final active context: 16,478 tokens
Final tool-result payload: 6,851 tokens
Cumulative tree input: 138,182 tokens
Estimated cost: $0.754

最后三个数值衡量的是不同的指标。16,478这个数字代表最终的活跃上下文长度,其中6,851个标记来自工具输出结果。138,182则是四个助手步骤中的累计输入量,包括了缓存创建和缓存读取的次数,因此并不表示智能体曾同时拥有138,182个标记的上下文窗口。当子智能体被纳入比较时,这种区别就显得尤为重要。

只有当工具搜索等便利功能被明确关闭时,简单的智能体才能作为合理的对照组。否则,当前SDK的默认设置可能会让基准测试的结果比预期更优秀。

写入:临时文件

第一项改动是增加了由文件系统支持的临时笔记区。plan.md用于记录问题、研究维度以及已完成和尚未完成的工作。research_notes.md用于整理各种观点、证据及矛盾点,sources.json存储引用相关的元数据,而open_questions.md则用于展示尚未解决的问题。体积较大的原始资料则存放在artifacts文件夹中。

# research_agent/workspace.py
def initialize_workspace(root: Path, question: str) -> Path:
    workspace = root / "workspace"
    workspace.mkdir(parents=True, exist_ok=True)
    (workspace / "artifacts").mkdir(exist_ok=True)
    (workspace / "checkpoints").mkdir(exist_ok=True)

    for name, template in WORKSPACE_FILES.items():
        path = workspace / name
        if not path.exists():
            path.write_text(template.format(question=question), encoding="utf-8")
    return workspace

由于这些文件具有持久性,代理可以在每个研究步骤之后更新它们,并仅读取下一步所需的文件。即使发生崩溃或数据压缩,精心整理的计划和证据依然完好无损,人们可以直接查看当前状态,而无需从记录中拼凑信息。

第三个图表来自一个离线脚本,该脚本用于初始化工作空间,并输出工程化代理用作临时存储区的具体文件列表。

会话还有另一个作用。会话文档将其描述为存储提示、工具调用、工具结果以及继续对话所需响应的场所。而临时存储区则保存着应用程序自身整理好的研究状态。恢复会话可以找回之前的处理路径,而读取plan.md则能获取人为维护的作业视图。

这种区分有助于避免两种常见错误。如果仅以转录文本作为数据库,恢复工作将依赖于越来越长的历史记录。而如果临时存储区变成了转录文本的存储库,同样的上下文问题就会在磁盘上重现。因此,每个文件都有较为有限的结构,用于存储结论、证据及未完成的工作,而非复制的对话内容。会话历史记录了智能体是如何达到当前状态的,而临时存储区则保存了决定下一步行动所需的小规模精选状态。

选择:有限的工作集

外部状态可能会超出单次模型调用所能处理的范围。通过按阶段加载可以解决这一问题:规划阶段读取plan.md,综合阶段读取research_notes.md,构建引用时则读取sources.json。工具与知识也遵循相同原则,因为就连它们的元数据也可能占据提示词中相当大的部分。

在工具方面,该设计实现了工具搜索功能。Anthropic的文档指出,一个目录最多可包含10,000个工具,除非另有设置,否则每次搜索会返回最多五个相关结果。它涵盖了MCP工具以及自定义的SDK MCP工具,因此庞大的目录可以在Claude需要特定结构时才被加载到上下文中。文档称,当工具数量在10个或更少时,预先加载所有工具通常速度更快,这使得延迟加载更多工具更多是出于扩展需求而非必然的优化手段。

在知识处理方面,有一个基于SQLite FTS5实现的自定义search_knowledge工具。该处理程序接收查询及有限的top_k参数,返回被引用的段落而非全部内容。SDK中的自定义工具则通过进程内的MCP服务器运行,因此这类示例虽保持自包含性,但仍需遵循与外部服务相同的边界限制。

# research_agent/tools.py
@tool(
    "search_knowledge",
    "Search the local memory-research collection and return only the most relevant cited passages.",
    {
        "type": "object",
        "properties": {
            "query": {"type": "string", "minLength": 1},
            "top_k": {"type": "integer", "minimum": 1, "maximum": 10},
        },
        "required": ["query", "top_k"],
        "additionalProperties": False,
    },
    annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False),
)
async def search_knowledge(args: dict[str, Any]) -> dict[str, Any]:
    try:
        return _text_result(
            search_corpus(corpus_path, args["query"], args["top_k"])
        )
    except (KeyError, TypeError, ValueError, sqlite3.Error) as error:
        return _error_result(error)

提示词要求设置top_k=5,而架构允许的值范围为1到10。通过这种方式将策略与硬性限制分开:日后可调整检索质量而不需修改工具契约,同时那些请求数千条记录的异常请求也会在淹没上下文之前被拒绝。

第四幅图展示了文件读取、工具搜索以及知识检索这三种获取不同类型上下文的方式,最终都会得到相同的结果:即为当前步骤限定一个固定的工作集。

离线测试会运行实际的搜索代码,确认数量上限为五项、每条返回结果都会保留其来源URL,以及会过滤掉无效的top_k值。由于测试用例规模极小,因此无法评估语义检索的质量。在实际应用中,这类判断应由检索系统依据经过评估的查询结果和召回率指标来完成,而非仅凭最终生成的文本来推测。将数据从对话中移除虽能释放空间,但关键在于如何选择,才能确保在相关内容出现时智能体仍能找到正确的五项结果。

压缩:可恢复的长会话

筛选机制决定了哪些内容会被纳入,但长时间的对话会不断积累旧的消息记录。当上下文长度接近限制时,Agent SDK会自动压缩历史记录:较旧的消息会被简化为摘要,而最近的对话内容及重要状态则会被保留。此时流中会出现一个子类型为compact_boundary的SystemMessage,以便统计代码能够记录其出现次数。

通用的摘要可能会遗漏研究工作所依赖的关键信息。因此,该项目在CLAUDE.md中制定了针对研究的特定保留规则:需保留原始问题、已验证的发现、未解决的问题、矛盾点、来源URL、重大决策以及当前计划。该文件决定了摘要应侧重哪些内容,从而无需在每次调用时都携带完整的临时记录。

在进行任何自动或手动的压缩操作之前,PreCompact钩子会先保存一个检查点。当SDK转录内容可用时,它会被复制下来,同时元数据会记录触发操作的时间、会话信息、工作目录以及转录内容的来源。

# research_agent/hooks.py

def build_precompact_hook(workspace: Path):
    async def archive_before_compaction(
        input_data: dict[str, Any],
        tool_use_id: str | None,
        context: Any,
    ) -> dict[str, Any]:
        del tool_use_id, context

        checkpoint_dir = workspace / "checkpoints"
        checkpoint_dir.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        session_id = input_data["session_id"]
        trigger = input_data["trigger"]
        stem = f"{timestamp}-{session_id}-{trigger}"
        transcript = Path(input_data["transcript_path"])

        metadata = {
            "session_id": session_id,
            "trigger": trigger,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "source_transcript": str(transcript),
            "custom_instructions": input_data.get("custom_instructions"),
            "archived": transcript.is_file(),
        }
        if transcript.is_file():
            shutil.copy2(transcript, checkpoint_dir / f"{stem}.jsonl")
        (checkpoint_dir / f"{stem}.json").write_text(
            json.dumps(metadata, indent=2) + "\n", encoding="utf-8"
        )
        return {}

    return archive_before_compaction

第五幅图展示了压缩功能如何将累积的对话内容缩减为符合保留规则的摘要,而检查点与临时存储区则作为独立记录保留在系统中。

临时存储区和压缩功能满足不同的需求。临时存储区是智能体在处理过程中有意写入的状态记录,便于日后查阅;而压缩功能则是在活跃对话历史过长时对累积内容进行的有损压缩处理。两者结合使用后,若摘要丢失了某些信息,压缩后的会话仍可依赖sources.json或research_notes.md来补充内容。

在离线测试中,系统向该钩子提供了临时转录文本,并验证了三点:检查点已生成,转录文本已被复制,触发元数据也得以保留。在六次实时对比测试中,从未出现compact_boundary,因为所观察到的最大上下文长度仅为34,309个标记,而预期窗口应为200,000个标记。因此该实验仅能证明该钩子功能正常,无法反映实时压缩效果的好坏。摘要应包含决策结果及未完成的工作,而临时存储区则应以可读取的形式保存相关证据。

隔离:专注的子代理

当多个调查方向没有共同的、不断发展的轨迹时,管理起来会更为简单。在首席调查员制定计划后,三名指定的子调查员负责纸质资料分析、证据库检索以及引用核查。每项职责都限定了任务范围、使用的工具以及工作交接的方式。

# research_agent/config.py
"paper-researcher": AgentDefinition(
    description="Analyzes primary research papers for memory mechanisms and trade-offs.",
    prompt=(
        "Investigate only the assigned paper question. Use primary sources. "
        "Return at most five findings, each with a URL and an explicit limitation."
    ),
    tools=["WebSearch", "WebFetch", "mcp__research__search_knowledge"],
    model=model,
),

在代码中,子代理是通过AgentDefinition来声明的。根据Anthropic关于子代理的文档,每个子代理都会在全新的对话中开始工作,并拥有自己的上下文;它会将中间阶段的工具调用及结果保存在该上下文中,仅向父代理发送最终消息。因此,这种信息传递构成了真正的上下文边界,而不仅仅是主控者在现有界面内的某种惯例。

第六幅图显示了每个专家都会收到一项特定的任务以及一个全新的界面。主控者只能看到简化的信息传递结果,而看不到生成该结果的任何搜索过程、获取的页面内容或提取步骤。

隔离机制需要付出一定成本。三个子代理可能会重新发现相同的来源,消耗更多标记,并使用相互冲突的术语。为避免这种情况,提示要求提供原始网址、明确限制条件以及不超过五项的查找结果。事实核查员只能看到有争议的声明,而无法查看全部内容,这既保证了其工作的可审计性,也降低了其擅自修改综合结果的风险。

上级系统必须能够评估返回的内容。没有来源标识的五行文字虽然简洁,但无法使用;而完整的子对话记录则会模糊界限。此处的合同格式会返回查找结果、引用信息、版本限制及约束条件,这些都是负责人在对比不同声明时所需的要素。仅隐藏子对话内容是不够的,因为只有当返回结果包含证据与不确定性时,隔离机制才能发挥作用。

隔离复杂工具的输出

某些负载内容仅应作为引用跨工具边界传递。materialize_source会将选定的文档写入workspace/artifacts/目录,并返回其路径、字节数以及简短预览。之后Claude可通过文件工具打开相应范围,这样整个文档就不会全部出现在某个工具的输出结果中。

# research_agent/workspace.py (full function; use the Gist when publishing)
def materialize_source(corpus_path: Path, workspace: Path, source_id: str) -> dict:
    document = next(
        (item for item in load_corpus(corpus_path) if item["source_id"] == source_id),
        None,
    )
    if document is None:
        raise ValueError(f"unknown source_id: {source_id}")

    path = safe_artifact_path(workspace, f"{source_id}.txt")
    content = (
        f"Title: {document['title']}\n"
        f"URL: {document['url']}\n"
        f"Published: {document['published']}\n\n"
        f"{document['text']}\n"
    )
    path.write_text(content, encoding="utf-8")
    return {
        "path": str(path),
        "characters": len(content),
        "preview": content[:240],
    }

SDK 对 MCP 的输出结果也有类似的安全机制。Anthropic 的 MCP 文档规定,输出内容超过 10,000 个标记时会触发限制,而默认上限为 25,000 个标记;过长的文本可被写入磁盘,并以文件引用的形式在上下文中使用。这一默认值起到了最后保障的作用。在达到紧急限制之前,通过返回路径和元数据,就能为应用程序提供稳定的行为规范。

第七张图来自一项故意设计的故障测试,该测试试图使用 ../outside.txt 来绕过输出目录的限制。工作区会在任何内容被写入之前就拒绝该路径。

环境隔离同样构成了安全边界。由模型控制的工件名称绝不能变成任意的文件系统路径。safe_artifact_path会接受一个经过筛选的文件名,并检查其最终路径是否仍在workspace/artifacts/目录内。实际出现的故障情况如下。

# Captured output: python scripts/demonstrate_failure.py
Blocked artifact path: artifact name must use only letters, numbers, dots, underscores, or hyphens

该检查并未提及操作系统沙箱机制或多租户隔离功能。它仅确认了一条严格规则:工具参数不得通过路径遍历方式在目标目录之外写入数据。实际部署仍需要进程级权限,以及为每个用户或运行任务分配独立的工作空间。而用路径替换大型数据包仅能在路径受到限制、代理能够稍后获取有限数据片段的情况下节省上下文开销。

在会话之间保留上下文

长时间运行的任务会比进程和上下文窗口的生命周期更长。SDK提供了三种选择:继续当前的对话、通过编号恢复特定会话,或将会话分叉为新的分支。选择“继续”适用于仍在处理的任务,“恢复”则是明确的重启操作,而“分叉”则适合需要继承先前上下文但不影响原会话后续进展的后续任务。

# examples/session_modes.py
def session_options(session_id: str) -> dict[str, ClaudeAgentOptions]:
    return {
        "continue": ClaudeAgentOptions(continue_conversation=True),
        "resume": ClaudeAgentOptions(resume=session_id),
        "fork": ClaudeAgentOptions(resume=session_id, fork_session=True),
    }

每个选项都会保留对话状态:提示语、工具调用、结果以及响应。创建分支时会生成新的会话ID,而原始会话则保持不变。这很适合诸如“现在专门针对编程智能体比较这些研究结果”这样的请求,因为新分支可以基于现有的研究方向继续推进。

第八幅图展示了不同模式的特点:continue用于延续当前的研究方向,resume用于恢复之前选择的方向,而fork则从同一个检查点开始新的分支。

磁盘上的文件需要单独处理。通过分叉创建的各个分支会共享同一个工作目录,除非应用程序为它们指定其他路径,否则这两个分支都可能覆盖 plan.md 或 final_report.md。为避免冲突,运行工具会使用 UTC 时间戳和随机后缀来生成唯一的运行根目录;而在实际使用中,应在任一分支写入内容之前,将临时文件复制或版本控制到新的运行根目录中。需注意,会话 ID 用于标识对话历史记录,它并非代理所访问的外部系统中的事务 ID,因此数据库、对象存储、搜索索引和工作空间各自都需要独立的追溯机制和并发策略。文件结构使得这些信息一目了然,而不会被恢复功能所掩盖。分叉仅用于分隔对话内容,而外部状态则会在工作空间被版本控制或拆分之前保持共享状态。

构建基于上下文工程的智能体

最终的架构整合了全部四种操作,且无需额外的编排框架。“写入”功能用于构建结构化工作空间;“选择”功能则负责按阶段读取文件、延迟加载工具架构以及进行有限范围的知识检索;“压缩”功能则将SDK数据与保留指令相结合,并在操作前生成检查点;“隔离”功能则意味着将智能体的各个组件限制在特定范围内,同时将工具输出存储在文件中。

# research_agent/config.py
tools = ["Read", "Write", "Edit", "Glob", "Grep", "WebSearch", "WebFetch", "Agent"]
return ClaudeAgentOptions(
    cwd=run_root,
    model=model,
    tools=tools,
    allowed_tools=[*tools, "mcp__research__*"],
    permission_mode="dontAsk",
    mcp_servers={"research": server},
    strict_mcp_config=True,
    setting_sources=["project"],
    system_prompt={"type": "preset", "preset": "claude_code", "append": ENGINEERED_PROMPT},
    env={"ENABLE_TOOL_SEARCH": "true"},
    hooks={"PreCompact": [HookMatcher(hooks=[build_precompact_hook(workspace)])]},
    agents=research_subagents(model),
    max_turns=30,
    max_budget_usd=max_budget_usd,
)

设置 setting_sources=[“project”] 可让针对特定运行的 CLAUDE.md 文件加入配置,而 strict_mcp_config=True 则会将 SDK 限制在当前应用程序提供的服务器上。permission_mode 设为 dontAsk 是因为该示例针对的是无人值守的独立运行环境,这并非通用的权限设置建议。高级审批流程和部署策略有意未被纳入考虑范围。

在第九幅图中,完整的架构涵盖了四个关键方面:持久化写入、选择性读取、对话内容压缩以及独立执行。主模型仅会接收到经过筛选的活跃上下文信息。

最终版本依赖于 ClaudeSDKClient,而省略了更轻量的 query() 辅助函数。二者使用相同的代理循环,但客户端提供了明确的响应边界以及一个结束用的 get_context_usage() 调用。这一设计选择是由工具化需求推动的,整体架构保持不变。

最初仅依靠提示词设计的尝试又暴露出了另一个生产环境中的问题:虽然三个 SDK 实例都运行完毕,但只有其中一个生成了 final_report.md;另外两个主导代理在委托的任务实际上尚未完成时就返回了。因此,SDK 的正常退出并不意味着应用程序的任务已经结束。解决办法是让子代理的调用显式地为同步操作,并为控制器提供独立的完成机制:控制器会查找包含至少两个 URL 的完整报告,并允许在同一会话中发起一次专注的修复查询。

# research_agent/runner.py
while True:
    async for message in client.receive_response():
        metrics.observe(message)

    report_path = workspace / "final_report.md"
    if mode == "naive" or _report_meets_contract(report_path):
        break
    if metrics.completion_retries >= MAX_COMPLETION_RETRIES:
        break

    metrics.completion_retries += 1
    await client.query(COMPLETION_REPAIR_PROMPT)

失败的仅提示结果会保留在 measurements/initial-prompt-only-comparison.json 中,因为控制器应当验证持久的结果,而非从模型的最终状态推断结果。

离线测试套件负责处理所有无需模型参与的任务:选项集管理、工作空间创建、有限范围检索、错误路径的拒绝处理、压缩检查点以及消息计数。这正是它在所述环境中报告的内容。

# Captured output: python -m unittest discover -s tests -q
----------------------------------------------------------------------
Ran 12 tests in 0.048s

OK

此时该框架已可正常运行:每个动词都对应一个具有独立代码的边界,该边界有状态管理负责人,同时存在可供测试验证的不变量。

两种架构的比较

关于架构的描述应当转化为具体数字。RunMetrics会监控SDK消息流,通过消息ID去除顶层重复的助手操作步骤,统计每次工具调用及每个压缩边界的数量,计算工具输出中的字符数,并获取最终的上下文快照。包括子代理在内的整个智能体树结构的累计模型使用量来自ResultMessage.model_usage,而total_cost_usd仅为本地估算值,并非正式账单。

# research_agent/metrics.py
if isinstance(message, ResultMessage):
    self.query_results += 1
    self.session_id = message.session_id
    self.result_subtype = message.subtype
    result = message.result or ""
    self.sdk_success = (
        message.subtype == "success"
        and bool(result.strip())
        and "not logged in" not in result.lower()
    )
    self.estimated_cost_usd = (
        (self.estimated_cost_usd or 0.0) + (message.total_cost_usd or 0.0)
    )
    self.result_usage = message.usage
    self.model_usage = self._merge_model_usage(self.model_usage, message.model_usage)
    self.total_tree_input_tokens = self._tree_input_tokens(self.model_usage)

if isinstance(message, SystemMessage) and message.subtype == "compact_boundary":
    self.compactions += 1

针对同一个问题,每种架构均使用完全相同的 SDK、模型别名、提供方以及本地知识配置,独立运行了三次。每次运行的记录及计算出的均值均保存在 measurements/comparison.json 文件中,有一个简单的显示脚本会从该文件重新生成紧凑的表格。

# terminal
python scripts/show_comparison.py ../measurements/comparison.json

第十张图表展示了三分值的均值,并揭示了明显的权衡关系。隔离机制减少了领头者所能看到的最终工具结果数据量,但子代理树却消耗了更多的标记数和预估成本。以下为显示脚本的捕获输出。

# Captured output: python scripts/show_comparison.py ../measurements/comparison.json
Measured comparison - 3 runs per architecture
Metric                               Naive      Engineered
Artifact success                       3/3             3/3
SDK success                            3/3             1/3
Mean tree input                    102,210         737,036
Mean peak context                   17,184          32,668
Final tool-result tokens             7,351           6,202
Mean subagents                           0               3
Mean estimated cost                 $0.663          $3.270
Compactions                              0               0

成功有两种表现形式。所谓“成果成功”,指的是架构的输出符合预期要求:对于简单智能体,表现为返回的报告中至少包含两个不同的网址;而对于经过设计的智能体,则表现为生成一份长度在500字以上、包含两个网址的持久性final_report.md文件。六次测试均达到了这一标准。

SDK运行成功的前提是最终的ResultMessage子类型为success。简单模式下三次尝试中有三次成功,而优化模式下三次尝试中仅有一次成功。两次优化模式下的运行虽然生成了报告,但最终结果为error_max_budget_usd,其中一次还使用了控制器的单次修复查询功能。Anthropic的预算示例说明,在API调用后会检查支出情况,因此即使运行已完成结果生成并在最终结果出现之前就满足了预设的预算限制,也仍算成功。应将这一上限视为保护机制,而非应用运行结束的标志。

具体数据如下:

  • 引导信息减少。在基准情况下,可见的工具结果字符平均数为28,572,而分离后则为23,428,下降了18.0%;最终上下文快照中工具结果所占的token数从7,351降至6,202,减少了15.6%。每次实验中恰好有三个子代理参与运行,且它们的中间搜索结果从未进入引导信息的轨迹。
  • 总体成本急剧上升。整个处理流程的输入token数最初平均为102,210,之后变为737,036,增长了621.1%。引导信息与上下文组合的最大输入token数平均为17,184,之后上升至32,668,增长了90.1%;SDK的成本估算值则从0.663美元上升至3.270美元,增长了393.4%。
  • 工具的使用频率也有所上升:整个系统中共有约30.3次调用,而基准情况下仅为6次。对于这类有明确限制的任务而言,三名专家、他们各自的处理方案以及彼此之间的协作成本,已经超过了为任务负责人节省下来的资源。
  • 由于每次运行都显示边界值为零,因此从未进行过压缩操作。这并不能说明具备上下文保留意识的压缩方法无效,只能说明该任务的数据量从未接近20万标记的阈值。此外,由于MCP目录规模较小,延迟发现机制也几乎没有机会弥补其带来的选择开销,因此该实验无法单独评估工具搜索的价值。

    结果能说明什么,又不能说明什么

    这里并没有任何内容可以作为全面反对上下文工程的依据。数据表明为何应选择性地使用这四种操作:写入文件和有限范围检索能够生成可被检查的持久状态。当某个子任务的执行路径可能污染本已相当长的主上下文时,隔离机制能起到作用,但这种做法并非毫无成本。在当前的任务规模下,除非专业质量上的提升能够弥补额外的开销,否则使用带有临时存储区的单个主上下文可能是更优的设计方案。

    最终剩下的是一个具备上下文架构、能在本地检测故障并拥有明确评估边界的可用成果。一个合理的后续问题是关于质量:所编写、筛选、压缩和隔离的材料真的是合适的材料吗?要回答这个问题,需要通过任务级评估来考察证据覆盖度、矛盾处理能力、引用有效性以及压缩后的恢复情况。在此过程中,任务隔离虽然减少了引导信息的工具结果负载,却增加了整个系统的标记数量与处理成本,因此必须将上下文工程视为一个整体系统来衡量,而不能简单地认为它就是一种优化手段。

    相关阅读

  • 为 .NET 10 中的 AI 智能体构建安全上下文编译器 — 在 .NET 10 中通过信任标签、新鲜度检查、内容屏蔽、注入隔离、确定性令牌预算以及可测试的 SHA-256 清单来编译模型上下文。