跳到正文
AWS Machine Learning Blog· Yoonseo Kim·· 3 小时前精选AI 评分62

AWS 机器学习博客详解四层上下文工程,Deep Insight 展示生产级 Agent 实践

Four layers of context engineering for production AI agents: How Deep Insight applies them in working code

AI 导读

AWS Machine Learning Blog 发布教程,将上下文工程整理为四层:多智能体架构隔离上下文、用提示词限制输入、用文件系统外化内容、用验证智能体和保障机制兜底。

推荐理由

文章将上下文工程整理为四层实践,并用开源参考实现给出具体代码与量化效果,方法可直接迁移到生产级多智能体系统。

正文 · AI 翻译

当生产级 AI 智能体处理数据分析和报告生成等任务时,上下文管理可能成为主要瓶颈。20 万 token 的上下文窗口听起来很充裕,但一个真实任务可能几分钟内就消耗掉大部分。在数据分析和报告生成场景中,这会导致智能体中途失败、准确性下降或成本飙升。业界已发表了大量相关指导,但这些技术读起来容易,应用起来却难。将它们结合使用会带来可衡量的收益:在本文演示的参考实现中,仅一个“写入并执行”工具就让每个脚本添加到上下文的 token 减少了约 95%,而按需加载技能则将系统提示词缩减了约 97%。

本系列文章将这些指导整理为可以直接应用的实践。Anthropic 的工程博客文章,加上 Strands Agents SDK 中内置的功能,共同指向四个层次的上下文工程:在架构层面隔离上下文、通过提示词控制上下文、将上下文外置到文件系统,以及用独立智能体和防护机制验证上下文。在本文(第 2 部分)中,你将了解每一层的工作原理及其在可运行代码中的呈现方式。我们在 Deep Insight(一个基于 Strands Agents SDK 和 Amazon Bedrock AgentCore 构建的开源多智能体数据分析解决方案)中应用了这些技术,它也是贯穿全文的参考实现。

系列文章

第一篇文章定义了生产级多智能体系统必须解决的五个核心挑战,并通过代码示例介绍了 Strands Agents SDK、Amazon Bedrock 和 Bedrock AgentCore 以及 AWS Fargate 各自如何解决该问题的不同层面。这第二篇文章将介绍构建在该架构之上的四层上下文工程实践,并逐一展示其实现。第三篇文章将指导你在 Amazon Virtual Private Cloud (Amazon VPC) 的安全私有网络中部署和运维智能体,涵盖实际运营所需的一切内容。

什么是上下文工程?

上下文工程是一门优化大语言模型(LLM)有限上下文窗口中的信息,以使智能体发挥最佳性能的学科。Amazon Bedrock 上提供的 Anthropic Claude Sonnet 4.5 拥有 20 万 token 的上下文窗口,但复杂的生产任务很快就会超出这一限制。例如,一个看似简单的请求,如“分析销售数据并生成 DOCX 报告”,要经历多个阶段:数据加载和探索(5 分钟)、按品类进行销售分析(3 分钟)、生成五张时间序列分析图表(10 分钟)、统计验证(3 分钟)以及最终报告生成(5 分钟)。合计在 15 分钟内消耗超过 15 万 token。

在单一上下文中处理此类任务会导致问题:窗口在任务中途溢出,陈旧的上下文降低性能,而错误则要求重新运行整个流水线。随着上下文不断增长,对重要信息的注意力会下降,而低效操作在按 token 计费的模式下会推高成本。

“随着上下文窗口中的 token 数量增加,模型从该上下文中准确回忆信息的能力会下降(上下文腐化,Context rot)。”

—— Anthropic,《面向 AI 智能体的有效上下文工程》(2025.09)

上下文工程恰好解决这些问题,而且它不仅仅是一系列技巧的集合。它贯穿架构设计、提示词编写、工具实现与验证,是一门分层学科,每一层都弥补前面各层的不足。本文详细介绍可用于高效管理上下文的四层方法。

如何处理超出上下文窗口的任务

Anthropic 已发布过多种处理长周期任务的技术。Deep Insight 将这些技术归为四层,每层弥补其他层的局限。下表列出了这四层。

层 实践 核心原则 局限
第 1 层:架构 通过多智能体架构隔离上下文 每个智能体在独立的上下文中运行 当仅靠隔离还不够时
第 2 层:提示词 输出控制与行为规则 通过提示词限制上下文的输入 当仅靠提示词还不够时
第 3 层:工具 基于文件的上下文外置 将重量级内容保存在上下文之外的文件中 当问题仍然出现时
第 4 层:验证智能体与保障措施 独立验证与保障措施 捕获错误并防止上下文溢出 最后一道防线

接下来的章节将依次介绍每一层,说明该层规定了什么、为什么仅靠前面几层不够,以及该实践在 Deep Insight 中是如何实现的。

1. 通过多智能体架构隔离上下文

给每个专项智能体分配自己的上下文窗口,只把压缩后的结果传给编排器。Deep Insight 通过 Coordinator、Planner、Plan Reviewer、Supervisor 和 Tool Agents(Coder、Validator、Reporter、Tracker、Auditor)的层级结构来实现这一点。

如果只用单个智能体,从用户请求到数据探索、分析、图表生成、验证和报告的所有信息都会堆积在一个上下文中,很容易超出窗口。给每个智能体分配各自的上下文,使得即使是用量最大的 Coder,总量也保持在 25K token 以内。每个智能体只接收它所需的内容,并通过完成通知和简要摘要向其他智能体发出信号。

隔离的关键是最小化共享信息。Deep Insight 使用一个 Python 字典 shared_state,它只存储四样东西:

  • messages:上一个智能体的最终消息(而非完整的对话历史)。
  • clues:从每个智能体收集的压缩摘要。
  • full_plan:由 Planner 生成、由 Tracker 更新的清单。
  • history:记录智能体名称和消息的简要日志。

线索采用固定的 CLUES_FORMAT:传递的不是每个智能体的完整对话(数万 token),而是压缩后的核心结果。

# coder_agent_tool.py -- context passing format
CLUES_FORMAT = "Here is clues from {}:\n\n<clues>\n{}\n</clues>\n\n"
RESPONSE_FORMAT = "Response from {}:\n\n<response>\n{}\n</response>\n\n*Please execute the next step.*"

# After Coder completes: add only compressed result to clues
clues += CLUES_FORMAT.format("coder", response["text"])

这一模式正是 Anthropic 在《我们如何构建多智能体研究系统》一文中所推荐的模式,该文描述了一种编排者-工作者模式:由一个主智能体将任务委派给专项子智能体,每个智能体只在自己的上下文中工作。你无需自己构建:Strands Agents SDK 的 agents-as-tools 模式已在框架层面提供了这种隔离。

2. 通过提示词控制上下文的流入

仅靠第一层隔离并不足够:即使拥有独立上下文,一个生成长回复的智能体也会撑大 Supervisor 的上下文。第二层通过提示词限制每个智能体的输出。

对于 Deep Insight,我们使用提示词限制每个智能体返回给 Supervisor 的信息量。三条规则起了大部分作用,正如 Deep Insight 的提示词所示。

“从最强的模型和简单的提示词开始。然后基于实际观察到的失败案例,逐步添加具体指令。”

—— Anthropic,《Effective Context Engineering for AI Agents》(2025.09)

实现细节 2-1:明确的单智能体输出 token 预算

每个智能体的提示词都规定了最大回复 token 数,防止长回复污染 Supervisor 的上下文。

智能体 输出 token 预算 原因
Coder 1,000–1,500 tokens 分析代码被保存到文件中,因此智能体只返回状态、已完成任务和洞察的摘要
Validator 800 tokens 验证结果保存到 citations.json。智能体只返回摘要
Reporter 1,000 tokens 智能体将报告生成为 DOCX 文件,只返回结果摘要

Anthropic 的“Writing Effective Tools for Agents”一文建议为工具提供 CONCISE 或 DETAILED 的响应选项,以提高上下文效率。把这个思路向上应用一层,写入每个智能体的系统提示词中。在 Deep Insight 中,所有智能体共享一个标准化格式:

## Status
[SUCCESS | PARTIAL_SUCCESS | ERROR]
## Completed Tasks
- [Task 1 from FULL_PLAN -- use exact wording so Tracker can match]
- [Task 2 -- be specific, not generic]
## Key Insights
- [Key finding 1 with numbers and ratios]
- [Key finding 2 with business implications]
## Generated Files
- ./artifacts/chart.png -- brief description
- ./artifacts/calculation_metadata.json -- N calculations

实现细节 2-2:自包含代码原则

Coder 智能体的提示词明确声明“变量不会在脚本之间持久保存”。每个脚本必须包含所有必要的 import 并直接加载数据。提示词同时提供了错误和正确的示例:

# WRONG -- Attempting to use a variable from Turn 1 in Turn 2
# Turn 1:
df = pd.read_csv('data.csv')
total = df['Amount'].sum()

# Turn 2:
print(f"Total: {total}")  # NameError: name 'total' is not defined

# CORRECT -- Every script is self-contained
import pandas as pd
df = pd.read_csv('./artifacts/cache/df_main.csv')  # Load from cache
total = df['Amount'].sum()
print(f"Total: {total}")

这条规则与上下文工程相关的原因在于:引用之前脚本的变量需要智能体“记住”先前的对话,这意味着过去的代码必须保留在上下文中。自包含原则意味着过去的代码可以安全地从上下文中移除。

实现细节 2-3:Supervisor 的分节完成规则

Supervisor 的提示词包含这样一条规则:“在当前智能体分节中的所有清单项都被标记为 [x] 之前,不要移动到下一个智能体。”我们把这条规则写在提示词里,因为判断 Coder 的工作是否完成,需要 Supervisor 利用 Tracker 的更新信息来解读计划。这种判断很难仅用代码表达。

3. “上下文中只放指针,实际数据放文件”

第一层在智能体之间隔离上下文,第二层限制智能体内部的输出。但当 Coder 智能体生成 100 行 Python 代码时,这些代码本身会进入上下文。即使有“简短回复”的提示词指令,有些代码也无法写得简短。第三层通过工具设计把繁重内容移出上下文,转移到文件系统中。四项技术带来了大部分收益,正如下面的实现细节所示。

实现细节 3-1:用 Write 和 Execute Tool 把代码移出上下文

Coder 读取分析计划、编写 Python 并执行以生成可视化。但当代码变长时,传统做法(分离的文件写入工具和执行工具)会把 100 行代码按脚本直接加入上下文,累计超过 1,000 行。

解决方案是把两个步骤合并为一个工具,这样代码就永远不会作为消息进入上下文。Deep Insight 的 Coder 智能体为此使用一个 write_and_execute_tool。

以下是 write_and_execute_tool 的一部分。写入并执行文件后,它只返回一个摘要:“已写入 105 行,执行成功,已完成五个类别的分析。”代码保存在文件中,上下文里只留下约三行摘要。根据这个示例估算,这样可节省大约 95% 的 token。

# write_and_execute_tool.py
def _handle_write_and_execute_tool(file_path, content, timeout=300):
    """Write a Python script and immediately execute it."""
    # Step 1: Write file
    with open(file_path, 'w', encoding='utf-8') as f:
        f.write(content)
    # Step 2: Execute immediately
    result = subprocess.run(
        f"python {file_path}",
        shell=True,
        capture_output=True,
        timeout=timeout
    )
    # Return only summary
    return f"Written {len(content.split(chr(10)))} lines to {file_path}\n" \
        f"Execution successful\n" \
        f"Output: {result.stdout[:500]}"  # Only first 500 chars

“服务端过滤:在执行层先行过滤数据,再让其进入上下文,token 减少 98.7%(从 150K 到 2K)。”

—— Anthropic,《Code Execution with MCP》(2025.11)

实现细节 3-2:在 Coder 智能体中将通用逻辑作为模块复用

通用模块会在多次分析中重复使用。与其每次重新生成,不如将常用函数一次性定义为模块再导入,这样可以避免 token 浪费和上下文重复。

如果不这样做,在每次分析中都要定义诸如 to_python_type() 或 track_calculation() 之类的函数,10 次分析会产生 200–300 行重复代码。Deep Insight 会在第一个任务之前生成一个名为 coder_analysis_utils.py 的模块,其中包含用于转换 numpy 类型、追踪计算元数据以及保存为 JSON 的函数。后续分析直接导入该模块,用两行 import 语句替代 20–30 行重复生成的代码,既节省 token,又提升了一致性和可调试性。

# coder_analysis_utils.py -- common module generated before first task
import numpy as np
import json
from datetime import datetime

_calculations = []

def to_python_type(value):
    """Convert numpy/pandas types to Python native types"""
    if isinstance(value, (np.integer, np.int64)): return int(value)
    elif isinstance(value, (np.floating, np.float64)): return float(value)
    elif isinstance(value, np.ndarray): return value.tolist()
    return value

def track_calculation(calc_id, value, description, formula,
                      source_file, source_columns, importance):
    """Track calculation metadata -- Validator independently verifies later"""
    _calculations.append({
        "id": calc_id, "value": to_python_type(value),
        "description": description, "formula": formula,
        "source_file": source_file, "source_columns": source_columns,
        "importance": importance
    })

def save_calculation_metadata(path="./artifacts/calculation_metadata.json"):
    """Save all tracked calculations to JSON"""
    with open(path, 'w', encoding='utf-8') as f:
        json.dump({"generated_at": datetime.now().isoformat(),
                   "calculations": _calculations}, f, ensure_ascii=False, indent=2)

实现细节 3-3:通过结构化笔记保存和共享分析结果

我们使用结构化笔记来共享详细的分析结果,而无需让结果经过每个智能体的上下文。Deep Insight 将这些结果存储在一个名为 all_results.txt 的文件中,各智能体按需读取。

例如,Coder 智能体会在每次分析任务完成后将结果追加到 all_results.txt 中:

with open('./artifacts/all_results.txt', 'a', encoding='utf-8') as f:
    f.write(f"""
## Category Sales Analysis (completed at: 14:32)
- Top revenue category: Fruit (417,166,008 KRW, 45% of total)
- Top 3 categories: Fruit, Vegetables, Meat (78% of total)
- Generated file: ./artifacts/category_sales_chart.png
""")

Coder 在每次分析后把详细结果记录到该文件中,Reporter 在撰写最终报告时读取它。其他智能体也通过 all_results.txt 以同样的方式协作。

如果没有笔记机制,Coder 在每次任务后都要把详细结果传给 Supervisor,不断累积在其上下文中。采用笔记机制后,只需传递一条约 30 token 的消息,例如“品类销售分析完成,发现 3 条关键洞察,详细结果已保存到 all_results.txt”,Reporter 在需要时再读取该文件。

这种模式可以泛化到单个文件之外。文件系统本身就可以充当智能体之间的通信总线,Deep Insight 正是这样做的:

文件名 生产者 消费者 作用
all_results.txt Coder Reporter 分析结果的累计日志
calculation_metadata.json Coder Validator 数值计算记录:来源、公式和数值(用于验证)
citations.json Validator Reporter 已验证的引用数据(以 [1]、[2] 形式插入报告)
audit_findings.json Auditor Reporter 在最终报告中发现的引用缺陷(驱动 Reporter 有针对性的重试)
coder_analysis_utils.py Coder 所有脚本 共享的实用函数(编写一次,反复导入)
full_plan (shared_state) Planner/Tracker Supervisor 清单状态(更新 [x])

“‘即时’(just in time)方法在运行时动态地将数据加载到上下文中,而不是预先处理所有相关数据,从而保持上下文的新鲜度、相关性,并能适应不断演变的任务状态。”

—— Anthropic,《Effective Context Engineering for AI Agents》(2025.09)

实现细节 3-4:通过 Skills 只加载所需的知识

这种方法不是将所有专业知识都放进系统提示词(system prompt),而是在需要时才加载。若包含每一份技能指南(数据结构检查、README 生成、技能创建、系统提示词编写、工具创建),将增加超过 20K token,使 Coder 智能体在每次调用时都加载不必要的信息,并且每添加一个技能都需要重新部署。

Deep Insight 的 Skills 系统在启动时只加载约 500 token 的技能目录,技能列表中仅包含名称和简要描述。当用户请求“创建一个 README 文件”时,智能体会判断“我需要 readme-generator 技能”,并调用 skill_tool(skill_name="readme-generator")。SkillLoader 读取 readme-generator/SKILL.md,并返回一份约 3K token 的详细指南。

# 1. Skill Discovery (at initialization)
from src.utils.skills.skill_utils import initialize_skills
available_skills, skill_prompt = initialize_skills(
    skill_dirs=["./skills"],
    verbose=True
)

# 2. Add to System Prompt (~500-token catalog)
system_prompt = base_prompt + skill_prompt

# 3. Runtime -- Agent loads skill when needed
# Agent decides: "This is README generation, I need the readme-generator skill"
# -> skill_tool(skill_name="readme-generator")
# -> Returns full SKILL.md content (~3K tokens)

每个技能被定义为一个 SKILL.md 文件,由 YAML frontmatter 和详细的 markdown 指南组成。其中包含名称、描述、许可证和允许使用的工具(allowed-tools)等元数据,以及 Overview、When to Use This Skill、详细说明和最佳实践(Best Practices)。要添加新技能,只需在 skills/ 目录下新建一个目录并编写 SKILL.md 文件,它就会被自动发现。

启动时,智能体加载约 500 token 的技能目录,而不是在完整指南上浪费超过 20,000 个 token(减少约 97%)。需要时,它还可以再加载几千 token 的技能详细说明。

4. 使用验证智能体和安全保障机制来验证质量

即使有了第 1–3 层,仍存在一个风险:相互隔离的智能体无法看到彼此的工作成果,因此如果 Coder 计算有误,Reporter 会把这个错误带进报告。第 4 层是最后一道防线。

“提供测试工具,让智能体能够自行验证功能,并修复那些仅从代码中难以察觉的 bug。让模型以描述性的提交信息将进度提交到 git,并在进度文件中撰写进度摘要。”

— Anthropic,《Effective Harnesses for Long-Running Agents》(2025.11)

实现细节 4-1:Tracker、Validator 和 Auditor 智能体:进度跟踪与质量保障

Tracker 实时更新计划中的检查清单。当 Coder 完成数据加载后,Supervisor 会调用 Tracker,Tracker 根据“数据加载完成(1,000 行,5 列)”等信息,将计划项从 [ ] 更新为 [x]。这让用户和系统都能准确了解进度处于什么位置。

Validator 独立重新执行 Coder 的计算以验证准确性。它读取 calculation_metadata.json,从源数据出发重新执行每项计算,并进行比对:误差在 0.01 以内视为通过,不匹配的则被标记出来。经过验证的计算会进入 citations.json,供 Reporter 引用。Reporter 完成后,Auditor 会对照 citations.json 和 calculation_metadata.json 检查最终的 DOCX 文档,但不修改它们。它将引用缺陷记录在 audit_findings.json 中,并最多可要求 Reporter 进行两次有针对性的返工。

实现细节 4-2:确定验证计算的优先级

验证每一项计算效率低下,因此应按重要性界定验证范围。Deep Insight 的 OptimizedValidator 采用以下阈值:

  • 计算项超过 50 个:激进过滤,验证所有高优先级项,外加仅 10 个中优先级项。
  • 计算项超过 20 个:适度过滤,验证所有高优先级项,外加 15 个中优先级项。
  • 计算项不超过 20 个:验证大多数,所有高优先级项加所有中优先级项再加 5 个低优先级项。

它还采用类型安全的比较方式:为避免 numpy 类型与 Python 类型不一致的问题,所有值都会被转换为 float(),并检查绝对差值是否小于 0.01:

# validator.md -- type-safe numerical comparison
try:
    match = abs(float(expected) - float(actual)) < 0.01
except:
    match = str(expected) == str(actual)

实现细节 4-3:SummarizingConversationManager:最后一道安全网

在极端情况下上下文仍可能溢出,因此要在其他层之下保留一道安全网。为此,Deep Insight 使用了 Strands Agents SDK 中的 SummarizingConversationManager。

# strands_sdk_utils.py -- safety net configuration at agent creation
agent = Agent(
    model=llm,
    system_prompt=system_prompts,
    tools=tools,
    conversation_manager=SummarizingConversationManager(
        summary_ratio=0.5,  # summarize oldest 50% on overflow
        preserve_recent_messages=10,  # always preserve 10 most recent messages
        summarization_system_prompt=apply_prompt_template("summarization", {})
    )
)

发生溢出时,LLM 会将最早 50% 的消息摘要为一条,同时始终保留最近的 10 条消息,以维持当前任务的上下文。摘要提示词(summarization.md)要求 clues、plans 和 tracking status 必须逐字保留。

实现细节 4-4:选择性提示词缓存以优化成本

上下文工程不仅关乎节省令牌,还包括成本优化。根据每个智能体使用相同提示词被调用的频率,为它们分别启用提示词缓存。Deep Insight 中各智能体的设置如下:

智能体 提示词缓存 工具缓存 原因
Supervisor 是 是 重复调用频率高,且系统提示词和工具定义规模较大
Coder 是 是 按顺序运行多个分析脚本,每次使用相同的提示词
Reporter 是 是 逐节递增式生成报告,提示词复用频率高
Auditor 是 是 在多轮对话中使用相同的提示词和工具检查 DOCX 与 JSON 文件,并在 Reporter 每次重试后重新运行
Coordinator 否 否 一次性单次调用,缓存无收益
Planner 否 否 每个请求仅调用一次,且仅在用户请求修订计划时再次调用
Validator 否 否 每次使用不同数据调用,缓存命中率低
Tracker 否 否 轻量级任务,无工具,无需缓存

启用缓存的智能体会在系统提示词和工具定义中插入缓存点,从而在重复调用时将输入令牌成本降低最多 90%。这是 Strands Agents SDK 原生支持的功能。Plan Reviewer 未出现在表格中,因为它是一个人在回路中的环节,用于收集用户的批准或反馈,而不是调用模型。

结论

本文详细介绍了四层工程方法,可用于高效管理超出上下文窗口的复杂任务的上下文。

  • 第 1 层(架构级隔离):通过多智能体架构让每个智能体在独立的上下文中工作,只传递压缩后的摘要。
  • 第 2 层(基于提示词的控制):输出令牌预算、标准化的响应格式以及自包含代码理念,主动限制上下文的流入。
  • 第 3 层(文件系统运用):write_and_execute_tool 将代码排除在上下文之外,文件系统充当智能体间的通信总线,知识仅在需要时懒加载。
  • 第 4 层(校验与保障):Validator 的独立计算复核、Auditor 对最终报告的只读引用审计、Tracker 的进度跟踪、SummarizingConversationManager 的溢出防护,以及选择性提示词缓存,共同管理质量与成本。

每一层构成一套防御体系,弥补前一层遗漏之处。当架构隔离不足时,由提示词限制输出。当提示词不足时,由工具将内容保持在上下文之外。而当问题仍然出现时,校验机制将其捕获。

这些层次中的每一层都在 Anthropic 的工程博文中有所记载:“Effective Context Engineering for AI Agents”、“Code Execution with MCP” 和 “How We Built Our Multi-Agent Research System”。他们建议的许多内容已经在 Strands Agents SDK 中内置,无需自行构建:SummarizingConversationManager、提示词缓存,以及 agents-as-tools 模式。

在构建生产级智能体时,仅靠 LLM 的上下文窗口往往是不够的。请应用这些上下文工程技术来高效管理上下文,并将你的智能体扩展到生产级工作负载。

Deep Insight(此处使用的参考实现)是开源的。你可以查看代码并针对自己的领域进行扩展。


关于作者

来源: AWS Machine Learning Blog · aws.amazon.com