压缩还不够?上下文卸载与隔离:Agent 长任务保持清醒的两个高阶策略

Write（写出去） — 把信息持久化到外部系统
Select（选进来） — 按需从外部系统检索
Compress（压下去）— 对已经在上文中的信息做精简

三者互相配合：Write 提供"信息库"，Select 决定"取什么"，Compress 控制"怎么用"。

# ❌ 错误做法：工具返回的完整内容占满上下文
result = web_search("Python async best practices 2026")
# 搜索结果 8000 tokens，全部注入上下文

# ✅ 正确做法：完整内容写文件，上下文只留引用
defsearch_and_offload(query: str, file_store: Path) -> str:
"""搜索结果卸载到文件系统"""
    result = web_search(query)  # 8000 tokens

    # 完整结果写入文件
    file_id = f"search_{hash(query)}.md"
    (file_store / file_id).write_text(result, encoding="utf-8")

    # 上下文中只保留摘要和文件路径（约 80 tokens）
    summary = llm.generate(f"用一句话概括以下搜索结果：\n{result[:2000]}")

    return f"""搜索结果摘要：{summary}
    完整内容存储在：{file_id}
    如需查看详情，使用 read_file("{file_id}") 读取。"""

# 压缩率：8000 → 80 tokens（100:1）

classScratchpadManager:
"""Agent 的草稿本：上下文卸载的最简实现"""

    def__init__(self, workspace: Path):
        self.workspace = workspace
        self.scratchpad_path = workspace / "scratchpad.md"

    definit_task(self, goal: str):
"""任务开始时初始化草稿本"""
        self.scratchpad_path.write_text(f"""# 任务目标
{goal}

## 进度
- 步骤 1: 进行中...
""", encoding="utf-8")

    defappend(self, note: str):
"""追加笔记到草稿本"""
        with open(self.scratchpad_path, "a", encoding="utf-8") as f:
            f.write(f"- {note}\n")

    defupdate_todo(self, todos: list[dict]):
"""更新待办列表——每次迭代后重写，确保目标在上下文末尾"""
        todo_section = "\n".join(
            f"- [{'x'if t['done'] else' '}] {t['task']}"
            for t in todos
        )
        # 将 todo 注入到上下文的末尾位置
        return f"""## 当前进度

{todo_section}

已完成 {sum(1for t in todos if t['done'])}/{len(todos)} 项任务。"""

    defget_summary(self) -> str:
"""获取草稿本摘要（不把全部内容放回上下文）"""
        content = self.scratchpad_path.read_text(encoding="utf-8")
        # 只返回最近 50 行
        lines = content.split("\n")
        return "\n".join(lines[-50:])

classFileSystemToolSpace:
"""文件系统作为 Agent 的无限上下文"""

    def__init__(self, sandbox_root: Path):
        self.root = sandbox_root
        # 预定义目录结构
        (self.root / "outputs").mkdir(exist_ok=True)
        (self.root / "temp").mkdir(exist_ok=True)
        (self.root / "knowledge").mkdir(exist_ok=True)

    defget_tool_manifest(self) -> list[dict]:
"""只暴露原子操作，不让模型看到具体工具"""
        return [
            {
                "name": "write_file",
                "description": "将内容写入文件",
                "parameters": {"path": "string", "content": "string"}
            },
            {
                "name": "read_file",
                "description": "读取文件内容",
                "parameters": {"path": "string", "offset": "int (optional)", "limit": "int (optional)"}
            },
            {
                "name": "run_shell",
                "description": "执行 Shell 命令（grep, cat, sed, awk 等）",
                "parameters": {"command": "string"}
            }
        ]

    defexecute(self, tool_name: str, params: dict) -> str:
"""执行工具调用，大结果只返回摘要"""
        if tool_name == "run_shell":
            result = self._run_shell(params["command"])
            # 大输出自动卸载到文件
            if len(result) > 2000:
                output_file = self.root / "temp" / "shell_output.txt"
                output_file.write_text(result, encoding="utf-8")
                return f"输出共 {len(result)} 字符，已写入 temp/shell_output.txt。" \
                       f"前 500 字符：\n{result[:500]}"
            return result

        # ... write_file, read_file 的实现

classSubAgentDispatcher:
"""主 Agent 调度子 Agent，只传递必要信息"""

    defdispatch(self, sub_agent: str, task: str, context: dict = None) -> str:
"""启动子 Agent，只给任务描述和必要上下文"""
        agent = self._get_agent(sub_agent)

        # 子 Agent 的上下文只包含：
        # 1. 子 Agent 专属的系统提示（短）
        # 2. 任务描述（用户输入）
        # 3. 必要的上下文片段（可选）
        system_prompt = self._get_role_prompt(sub_agent)
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"任务：{task}"}
        ]

        if context:
            messages.insert(1, {
                "role": "system",
                "content": f"相关上下文：\n{json.dumps(context, ensure_ascii=False)}"
            })

        result = agent.run(messages)
        return result  # 只返回最终结果给主 Agent

# 使用示例
dispatcher = SubAgentDispatcher()

# 主 Agent 分析用户请求后，分发子任务
code_result = dispatcher.dispatch(
    sub_agent="code_analyzer",
    task="分析 auth.ts 中的安全漏洞"
)
test_result = dispatcher.dispatch(
    sub_agent="test_writer",
    task="为 auth.ts 的 login 函数写单元测试"
)

# 两个子 Agent 的上下文完全隔离，互不干扰

classSharedMemoryDispatcher:
"""通过文件系统共享中间结果"""

    def__init__(self, workspace: Path):
        self.workspace = workspace
        self.shared = workspace / "shared"

    defrun_parallel_tasks(self, tasks: list[dict]) -> dict:
"""并行执行多个子任务，通过文件交换结果"""
        results = {}

        # 阶段 1：并行执行
        for task in tasks:
            agent = SubAgent(task["type"], self.workspace)
            output = agent.run(task["instruction"])
            # 结果写入共享文件
            result_file = self.shared / f"{task['id']}.json"
            result_file.write_text(json.dumps(output, ensure_ascii=False))
            results[task["id"]] = output

        # 阶段 2：汇总分析（子 Agent 可以读取其他 Agent 的结果）
        summary_prompt = f"""分析以下子任务结果，给出综合结论：

        {self._format_shared_results(results)}

        各任务结果存储在 shared/ 目录下，如需查看详情请读取对应文件。"""

        return self._run_summary_agent(summary_prompt)

# HuggingFace 的 CodeAgent 实践
# Agent 生成的代码在独立沙盒中执行，只有返回值和错误信息传回上下文

classSandboxExecutor:
"""在隔离环境中执行代码，只回传必要信息"""

    defexecute(self, code: str, timeout: int = 30) -> dict:
"""沙盒执行代码"""
        try:
            # 在独立进程中执行
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd="/tmp/sandbox"  # 受限的工作目录
            )

            # 只回传摘要信息
            return {
                "success": result.returncode == 0,
                "output_summary": result.stdout[:500],
                "error_summary": result.stderr[:200] if result.stderr else None,
                "output_size": len(result.stdout)
            }
        except subprocess.TimeoutExpired:
            return {"success": False, "error_summary": "执行超时 (>30s)"}

defbuild_isolated_prompt(agent_type: str, task: dict,
                          shared_context: dict) -> str:
"""构建带隔离区的系统提示"""

    prompt = f"""你是 {agent_type}，专注于特定领域。

=== 你的职责 ===
{task['role_instructions']}

=== 当前任务 ===
{task['instruction']}

=== 共享上下文（只读） ===
{json.dumps(shared_context, ensure_ascii=False, indent=2)}

=== 其他 Agent 的状态（不要修改） ===
{json.dumps(task.get('peer_states', {}), ensure_ascii=False, indent=2)}

仅基于"你的职责"和"当前任务"工作。共享上下文供参考，不需要全部使用。
不要修改其他 Agent 的状态区域。"""
    return prompt

# L1 原子函数（在上下文中）
FIXED_TOOLS = [
    "read_file(path)",      # 读取文件
    "write_file(path, content)",  # 写入文件
    "run_shell(cmd)",       # 执行 Shell 命令
    "search(query)",        # 网页搜索
]

# L2 沙盒工具（不在上下文中，通过 run_shell 调用）
# Agent 执行：run_shell("ffmpeg --help") 查看用法
# Agent 执行：run_shell("jq '.data[] | select(.score > 0.8)' results.json")

# L3 Python 脚本（不在上下文中，通过 execute_python 调用）
# Agent 执行：execute_python("import yfinance; ...")

隔离的收益 = 准确率提升 × 任务价值
隔离的代价 = Token 增加 × 单价 + 协调复杂度 × 开发成本

当收益 > 代价 时启用隔离。