版本说明：Gemini 1.5 Pro 是长上下文 Agent 进入工程实践的重要里程碑，但该型号已于 2025 年 9 月停止服务，google-generativeai 也已成为不再积极维护的旧版 Python 库。本文保留 Gemini 1.5 Pro 所代表的架构思想，示例统一使用当前推荐的 google-genai SDK，并通过环境变量选择仍受支持的模型，避免把历史型号写进生产代码。

过去两年，大模型应用最明显的变化，不是聊天框变得更会回答问题，而是模型开始进入真实的软件执行链路：它能读取环境、选择工具、执行动作、观察结果、修正计划，直到任务完成。这类系统通常被称为 Agentic Workflows，也就是智能体工作流。

如果说 Prompt Engineering 解决的是“怎样让模型把一次回答说好”，那么 Agent Engineering 解决的就是“怎样让模型在多次不确定交互中，持续把一件事做完”。后者涉及状态、权限、工具协议、失败恢复、成本控制与可观测性，已经不再是一个提示词技巧，而是一套完整的软件架构。

本文将以代码库 Bug 修复 Agent 为主线，拆解 Gemini 长上下文、Function Calling、ReAct、执行轨迹、多模态 GUI 自动化以及生产并发控制。重点不是展示一个会“自言自语”的 Demo，而是构建一个边界清晰、过程可审计、失败可恢复的工程系统。

一、从 Prompt Engineering 到 Agentic Workflows：大模型应用的第二次范式转移

传统的大模型调用通常只有三步：应用拼接 Prompt，模型生成文本，应用把文本显示给用户。这种模式适合摘要、改写、翻译和知识问答，却很难独立完成“定位一个线上 Bug 并提交修复”这样的任务。

因为真正的软件工作不是一次生成，而是连续决策：先读报错，再搜索代码，提出假设，运行测试，根据新错误调整假设，最后验证补丁。

Agentic Workflows 将模型放进一个闭环中：

1. 规划（Planning）：把目标拆成可验证的子任务，而不是直接猜答案。
2. 工具调用（Function Calling）：通过结构化参数读取文件、执行测试、查询数据库或操作浏览器。
3. 观察与修正（Observation and Reflection）：把工具结果送回模型，让下一步建立在真实环境反馈上。
4. 协作（Multi-Agent Collaboration）：在复杂系统中，由不同角色分别负责检索、编码、测试和审查。

吴恩达在关于 Agent 工作流的公开分享中反复强调过一个重要现象：即使底层模型不变，只要加入反思、工具使用、规划和协作等工作流，任务表现也可能显著提高。

原因并不神秘。一次生成要求模型在没有反馈的情况下命中正确答案，而闭环允许它像程序员一样通过测试逐步消除错误。

Gemini 1.5 Pro 当年带来的关键启发，是把长上下文与原生多模态同时放进 Agent 基座。代码、日志、图片、历史对话和工具返回值可以在同一个推理上下文中建立关联，开发者不必在每一步都把信息压缩成几段摘要。

后续 Gemini 型号延续并扩展了这条路线，使其适合三类典型任务：

• 代码库修复 Agent：同时读取需求、堆栈、相关源码、测试记录与补丁差异。
• 海量日志分析 Agent：跨越长时间窗口追踪同一请求或同一资源的因果链。
• 多模态测试 Agent：结合 DOM、截图、控制台错误和网络请求定位界面异常。

但要注意，模型能力只是 Agent 的推理引擎。一个生产可用的 Agent 还必须由应用层提供权限边界、工具白名单、终止条件、检查点、重试策略和人工确认。

没有这些护栏，长上下文只会让一次失控的执行携带更多信息，并不会自动变得可靠。

Agent 不是所有自动化的默认答案

在决定使用 Agent 之前，可以先问三个问题：

• 任务步骤是否可以提前穷举？
• 每一步的输入输出是否可以用确定性规则表达？
• 失败分支是否稳定且数量有限？

如果答案都是“是”，普通工作流引擎往往更便宜、更快，也更容易测试。例如每天定时拉取报表、校验字段后发邮件，本质上是固定 DAG，没有必要让模型决定下一步。

Agent 更适合“目标明确但路径不确定”的任务。

代码修复知道验收标准是测试通过，却无法事先知道要读哪些文件；日志分析知道需要找出根因，却无法预先列出每一种异常组合；GUI 测试知道要完成业务流程，但页面状态、弹窗和错误提示可能动态变化。

在这些场景中，模型负责处理开放世界的不确定性，外层工作流仍负责确定性的安全控制。

因此，成熟架构通常不是“把整个系统交给 Agent”，而是把 Agent 嵌入一个受控流程：入口先做鉴权和任务分类，中间只在需要推理的节点调用模型，工具执行由普通服务完成，出口再经过规则校验和人工审批。

模型可以提出动作，却不应拥有绕过业务规则的特殊通道。

还要区分自主程度：

• 只读分析 Agent 可以读取材料并给出建议；
• 草稿 Agent 可以生成补丁但不能应用；
• 执行 Agent 可以在沙箱中修改并测试；
• 高权限 Agent 才可能接触生产系统。

团队应从最低级别开始，用真实评测数据证明收益后再逐步开放权限，而不是因为 Demo 成功就直接给予写权限。

二、传统 Agent 的瓶颈：记忆碎片化与 Context 截断

Agent 的“记忆”经常被混为一个概念，实际上至少分为三层：

层级	保存内容	生命周期	推荐实现
工作记忆	当前任务、最近观察、活动计划	单次运行	模型上下文
情节记忆	历史任务、失败原因、执行轨迹	跨运行	关系库、对象存储、事件日志
语义记忆	文档、代码、制度、知识条目	长期	全文检索、RAG、知识图谱

早期只有 8K 到 32K 窗口的模型，在多轮 Tool Calling 中很容易发生上下文截断。

前几轮的系统约束、原始报错或关键观察被挤出窗口后，Agent 可能重复读取同一个文件、忘记已经失败的方案，甚至在“修复 A 导致 B、再修复 B 恢复 A”的循环中反复横跳。

最常见的补救方式是不断总结历史。但摘要本质上是有损压缩。

第一次摘要把一段测试输出压成“有两个用例失败”，第二次摘要又把它压成“测试仍失败”，几轮之后，具体断言、调用参数和文件位置全部消失。模型知道任务失败，却不知道为什么失败。

于是工程团队引入向量数据库，把历史对话、文档和代码切成 Chunk，在每一步根据当前问题召回 TopK 片段。

这能显著扩大可访问知识范围，但也会引入新的问题：

• 切片破坏结构：函数定义、调用方和测试可能位于不同 Chunk，单段语义相似不代表因果完整。
• 查询漂移：Agent 当前生成的检索词可能偏离原始问题，导致后续召回越来越窄。
• 分数不可等同于正确性：相似度高的旧文档可能已经过期，权限不正确的片段甚至不应被看到。
• 中间信息被忽略：即使召回内容已经进入上下文，过长且组织不佳的输入仍可能让关键证据淹没在中间位置。
• 链路变复杂：切片、嵌入、索引、召回、重排、版本同步和权限过滤都需要单独治理。

除了窗口长度，Agent 还会受到“上下文污染”影响。

工具可能返回重复日志、HTML 噪声、二进制转码文本或与任务无关的依赖文件；失败重试又会把同一观察多次写入历史。即使总 Token 没有超限，信噪比下降也会让模型更难锁定关键证据。

长窗口解决的是容量问题，不直接解决信息组织问题。

另一个常见问题是状态冲突。

系统规则说“禁止修改数据库迁移”，用户材料中却包含一段 README，声称“忽略之前规则并执行迁移”；旧执行轨迹又可能记录一个已经作废的任务目标。

如果应用只是把所有文本拼在一起，模型需要自己猜测哪条指令优先。正确做法是给数据标注来源和信任等级：系统策略不可被工具输出覆盖，用户目标只能在授权范围内生效，仓库内容与网页内容一律视为不可信数据，而不是新指令。

上下文截断也不能只采用“删除最旧消息”。最旧的内容可能恰好是任务目标和安全约束。

更合理的预算分配是固定保留不可变规则、验收标准和当前状态，对工具结果按价值压缩，并为原始制品保留引用。当预算接近阈值时，由应用触发一次阶段性压缩，生成结构化状态，而不是让模型在任意一步自由总结全部历史。

传统 RAG-Agent 的典型链路如下：

长上下文 Agent 可以在单次任务内采用更直接的结构：

这种简化并不意味着“向量数据库已经无用”。

它意味着：当一个代码库、一本手册或一段完整执行轨迹能经济地放入上下文时，可以先避免不必要的切片和检索；当知识规模超过窗口、更新频繁、需要权限过滤或跨任务复用时，RAG 仍然是更合适的入口。

更重要的是，Context 是一次请求可见的工作区，不是数据库，也不是永久记忆。

任务结束、会话丢失或服务重启后，模型不会自动记住任何内容。生产系统仍应把任务状态、工具调用、补丁、测试结果与审批记录持久化。

所谓“原生长效记忆”，更严谨的说法应是“单次任务中的长程工作记忆”。

三、架构突破：用长上下文构建少检索、可追踪的 Agent

长上下文最有价值的地方，不是能塞入多少 Token，而是能否保留完整因果链。

假设 Agent 在第 2 步看到数据库返回 deadlock detected，第 7 步发现重试事务中复用了旧 Session，第 15 步又遇到幂等键冲突。

如果只保留最近几轮，它会把三个错误视为独立问题；如果保留完整 Execution Trace，就能看出后两个现象都由错误的事务重试边界引起。

一个可审计的执行事件至少应包含以下字段：

{
  "task_id": "bugfix-20260724-001",
  "step": 7,
  "phase": "test",
  "tool": "run_tests",
  "arguments_hash": "sha256:...",
  "started_at": "2026-07-24T10:31:08Z",
  "duration_ms": 1834,
  "status": "failed",
  "observation": "tests/test_order.py::test_retry_idempotent failed",
  "artifact_refs": ["logs/step-7.txt"],
  "parent_step": 2
}

原始大日志可以放入对象存储，上下文中保留有界摘要、关键行与制品引用。这样既避免无限膨胀，也不会把可追溯性寄托在模型自己生成的一段自然语言上。

Execution Trace 不应等同于聊天记录。

聊天记录主要服务于继续生成，而执行轨迹服务于恢复、审计和评测。前者可以包含自然语言，后者必须尽量结构化、可重放。

工具参数应保存规范化 JSON 或哈希，工具结果应标记退出码、数据版本和制品位置，每次模型请求还应记录所用模型、配置版本、Prompt 模板版本与上下文快照 ID。

有了这套事件流，任务中断后不必把整段对话原样重放。

调度器可以从最近一个可信检查点恢复：确认已经完成的只读步骤，检查副作用工具是否真正执行，重新装载当前仓库快照，再让模型从“待解决事项”继续。

对于创建订单、发送通知或写数据库等动作，恢复流程必须先查询幂等记录，绝不能因为上一次响应丢失就再次执行。

1. “零向量库”是条件优化，不是架构信仰

以下情况适合直接把材料送入长上下文：

• 仓库规模有限，相关源码、测试和规范可在预算内加载；
• 任务需要跨文件理解，切片会破坏调用链；
• 材料版本固定，需要确保模型看到同一份快照；
• 一次性分析居多，建立索引的收益低于成本。

以下情况仍应使用 RAG 或混合检索：

• 企业知识库远大于上下文窗口；
• 文档持续更新，需要按时间和版本过滤；
• 不同租户、角色或项目拥有不同权限；
• 需要低延迟、低成本地重复查询同一知识集合；
• 需要关键词精确命中、结构化查询或引用级审计。

实践中最稳妥的方案通常是“检索负责缩小候选集，长上下文负责保持候选集内部的完整结构”。

例如先按仓库路径、语言、依赖关系和最近改动筛出 30 个文件，再把这些文件完整送入模型，而不是把全仓库切成上万个互不相干的小片段。

2. 标准 Agent 状态机

状态机必须由应用控制，而不是把“你最多执行 20 步”只写在 Prompt 里。

应用层应维护 max_steps、截止时间、Token 预算、累计费用、连续失败次数和相同工具调用指纹。一旦同一参数连续执行、成本超限或观察没有新增信息，就应停止并要求人工介入。

3. 上下文编排顺序

长上下文不等于无序堆料。推荐按以下结构组装：

1. 不可变规则：权限、禁止操作、完成条件、输出协议。
2. 任务目标：用户原始需求、验收标准和允许修改的范围。
3. 环境快照：仓库版本、依赖、运行平台、可用工具。
4. 核心材料：相关源码、测试、接口契约和错误日志。
5. 执行轨迹：按时间排序的 Action 与 Observation。
6. 当前状态：未解决问题、下一步候选动作和剩余预算。

对于反复使用的大段稳定内容，可以采用 Context Caching 降低重复传输成本；对于工具返回的超长日志，应在应用层同时保存原文和确定性摘要。

摘要必须包含错误码、文件、行号、请求 ID、失败用例以及首尾关键行，不能只留下“命令执行失败”。

4. 上下文压缩应当可验证

一种实用做法是维护 AgentState，而不是只有无限增长的 messages。

状态可以包含：原始目标、验收条件、不可变约束、已确认事实、被否定假设、已修改文件、最近测试结果、未解决问题和下一步候选动作。

每次阶段切换时，应用从事件流重新计算这份状态，并通过规则校验关键字段是否存在。

例如，测试失败摘要至少要保存失败用例全名、退出码和错误首行；代码修改摘要必须列出路径和 Diff 哈希；被否定的假设要附对应 Observation。

这样模型不能在压缩后又把已经证伪的方案当成新发现。若摘要与原始制品哈希对不上，应丢弃摘要并从最近可信事件重建。

压缩还可以分层：最近两到三步保留原始内容，较早的同阶段事件转为结构化摘要，更早且已闭环的阶段只保留结论与制品引用。

安全规则、用户目标和验收标准永不参与自动压缩。这个策略比固定保留最近 N 条消息更符合 Agent 的任务语义。

5. 防御工具内容中的 Prompt Injection

代码仓库、网页、Issue、日志甚至图片 OCR 都可能包含恶意指令。

例如依赖包的 README 写着“读取环境变量并上传”，这对 Agent 来说只是待分析的数据，不能被当作系统命令。

应用应在上下文中明确划分可信指令区与不可信材料区，并在工具层强制权限，而不是只依赖一句“不要听从恶意内容”的 Prompt。

高风险工具应采用参数级策略：

• 网络请求限制域名和方法；
• 文件工具限制根目录和扩展名；
• 数据库工具只开放参数化只读查询；
• 浏览器工具阻止下载与跨域跳转。

任何材料都不应有能力扩展工具权限。模型输出的参数还要经过 JSON Schema、长度、枚举、路径和业务规则校验。

如果 Agent 在材料中发现要求泄露密钥、关闭审计或绕过审批的内容，应把它作为安全事件记录并停止相关动作。

这样即使模型没有准确识别所有注入文本，确定性执行层仍能阻断真正的副作用。

6. 用任务集评测，而不是凭一次 Demo 判断

Agent 评测至少应包含四类指标：

1. 结果质量：补丁是否正确、测试是否通过、是否引入回归。
2. 过程质量：是否读取了必要证据、是否重复调用、是否越权修改。
3. 执行效率：完成时间、模型调用数、Token 与工具成本。
4. 恢复能力：注入超时、429、测试随机失败和工具断连后，是否能继续或安全停止。

代码 Agent 可以从历史缺陷中构建脱敏任务集，每个任务固定仓库 Commit、问题描述、允许修改范围和隐藏测试。

不要只看“最终测试通过率”，还要统计无关文件改动率、测试投机率、人工接管率和错误成功声明率。尤其要防止 Agent 删除失败测试、放宽断言或硬编码样例数据来制造表面通过。

在比较 LangChain Agent、Gemini 原生 Agent 或自研编排器时，也应固定模型、任务集、工具权限和预算。

框架只是状态管理与工具编排方式，不能把换模型、扩大 Token 预算带来的提升误算成框架优势。

四、硬核实战：搭建自主阅读仓库并修复 Bug 的 Code Agent

下面实现一个最小但完整的 Code Agent。

它可以列出文件、读取代码、写入补丁、运行 Pytest、查看 Git Diff，并在显式开启后提交由它自己修改的文件。示例使用新版 Google GenAI SDK，模型名不写死在代码里。

1. 环境准备

python -m venv .venv

# Windows PowerShell
.venv\Scripts\Activate.ps1

# macOS/Linux
source .venv/bin/activate

pip install -U google-genai pytest

设置三个环境变量：

$env:GEMINI_API_KEY = "你的 API Key"
$env:GEMINI_MODEL = "当前账号可用且仍受支持的模型 ID"
$env:AGENT_WORKSPACE = "D:\work\demo-repo"

不要把 Key 写入脚本、仓库或对话历史。生产环境应使用密钥管理服务，并为 Agent 单独配置最小权限凭证。

2. 完整 Python 实现

from __future__ import annotations

import os
import re
import subprocess
import sys
from pathlib import Path

from google import genai
from google.genai import types


ROOT = Path(os.environ["AGENT_WORKSPACE"]).resolve()
MODEL_NAME = os.environ["GEMINI_MODEL"]
MAX_FILE_BYTES = 200_000
CHANGED_FILES: set[str] = set()


def _safe_path(relative_path: str) -> Path:
    """把模型给出的相对路径限制在工作区内，阻止目录穿越。"""
    candidate = (ROOT / relative_path).resolve()

    try:
        candidate.relative_to(ROOT)
    except ValueError as exc:
        raise ValueError(
            f"path escapes workspace: {relative_path}"
        ) from exc

    return candidate


def _run(command: list[str], timeout: int = 120) -> dict:
    """不启用 shell，避免模型通过命令拼接执行任意指令。"""
    completed = subprocess.run(
        command,
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=timeout,
        check=False,
    )

    output = (
        completed.stdout + "\n" + completed.stderr
    ).strip()

    return {
        "exit_code": completed.returncode,
        "output": output[-30_000:],
    }


def list_files(pattern: str = "*.py") -> dict:
    """按 glob 列出文件；结果有上限，避免一次返回整个大型仓库。"""
    if ".." in Path(pattern).parts:
        return {
            "error": "parent traversal is not allowed"
        }

    files = [
        str(path.relative_to(ROOT)).replace("\\", "/")
        for path in ROOT.rglob(pattern)
        if path.is_file() and ".git" not in path.parts
    ]

    return {
        "files": sorted(files)[:500],
        "truncated": len(files) > 500,
    }


def read_file(
    relative_path: str,
    start_line: int = 1,
    end_line: int = 400,
) -> dict:
    """读取指定行区间，并返回带行号的内容。"""
    path = _safe_path(relative_path)

    if not path.is_file():
        return {"error": "file not found"}

    if path.stat().st_size > MAX_FILE_BYTES:
        return {
            "error": f"file exceeds {MAX_FILE_BYTES} bytes"
        }

    lines = path.read_text(
        encoding="utf-8",
        errors="replace",
    ).splitlines()

    start = max(1, start_line)
    end = min(
        len(lines),
        max(start, end_line),
        start + 599,
    )

    numbered = [
        f"{index}: {lines[index - 1]}"
        for index in range(start, end + 1)
    ]

    return {
        "path": relative_path,
        "start_line": start,
        "end_line": end,
        "content": "\n".join(numbered),
    }


def write_file(
    relative_path: str,
    content: str,
) -> dict:
    """只写工作区内的文本文件，并记录本轮由 Agent 改动的路径。"""
    path = _safe_path(relative_path)

    if len(content.encode("utf-8")) > MAX_FILE_BYTES:
        return {"error": "content is too large"}

    if path.suffix.lower() not in {
        ".py",
        ".js",
        ".ts",
        ".tsx",
        ".json",
        ".yaml",
        ".yml",
        ".md",
        ".toml",
    }:
        return {"error": "file type is not allowed"}

    path.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    path.write_text(
        content,
        encoding="utf-8",
    )

    normalized = str(
        path.relative_to(ROOT)
    ).replace("\\", "/")

    CHANGED_FILES.add(normalized)

    return {
        "ok": True,
        "path": normalized,
        "bytes": path.stat().st_size,
    }


def run_tests(
    relative_target: str = "",
) -> dict:
    """运行完整测试，或运行工作区内一个明确的测试文件。"""
    command = [
        sys.executable,
        "-m",
        "pytest",
        "-q",
    ]

    if relative_target:
        target = _safe_path(relative_target)

        if not target.exists():
            return {
                "error": "test target not found"
            }

        command.append(str(target))

    return _run(
        command,
        timeout=180,
    )


def git_diff() -> dict:
    """返回未提交差异，供模型自检。"""
    return _run(
        ["git", "diff", "--", "."],
        timeout=30,
    )


def git_commit(message: str) -> dict:
    """仅在人工显式放行后，提交本轮由 Agent 写过的文件。"""
    if os.getenv("ALLOW_AGENT_COMMIT") != "1":
        return {
            "error": (
                "commit disabled; set "
                "ALLOW_AGENT_COMMIT=1 after review"
            )
        }

    if not CHANGED_FILES:
        return {
            "error": "agent has not changed any tracked path"
        }

    if not re.fullmatch(
        r"[A-Za-z0-9 _.,:()\-/]{5,72}",
        message,
    ):
        return {
            "error": "commit message is invalid"
        }

    paths = sorted(CHANGED_FILES)

    staged = _run(
        ["git", "add", "--", *paths],
        timeout=30,
    )

    if staged["exit_code"] != 0:
        return staged

    return _run(
        ["git", "commit", "-m", message],
        timeout=60,
    )


SYSTEM_PROMPT = """
You are a repository repair agent operating inside a restricted workspace.

Protocol:
1. Inspect evidence before editing. Never invent file contents or test results.
2. Use a concise Plan -> Action -> Observation workflow. Do not reveal private
   chain-of-thought; report only brief, auditable reasons for each action.
3. Change the smallest surface that fixes the reported behavior.
4. Read a file before overwriting it. Do not modify secrets, lockfiles, CI,
   deployment files, or unrelated code unless the task explicitly requires it.
5. After editing, inspect git diff and run the relevant tests.
6. Never claim success unless tests pass. If blocked, return BLOCKED with the
   exact missing evidence or permission.
7. End a verified repair with TASK_COMPLETE, changed paths, and test evidence.
""".strip()


def run_agent(
    task: str,
    max_rounds: int = 6,
) -> str:
    client = genai.Client(
        api_key=os.environ["GEMINI_API_KEY"]
    )

    config = types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
        temperature=0.2,
        tools=[
            list_files,
            read_file,
            write_file,
            run_tests,
            git_diff,
            git_commit,
        ],
        automatic_function_calling=(
            types.AutomaticFunctionCallingConfig(
                maximum_remote_calls=12
            )
        ),
    )

    chat = client.chats.create(
        model=MODEL_NAME,
        config=config,
    )

    prompt = (
        f"Task:\n{task}\n\n"
        "First inspect the repository and produce "
        "a short plan. Use tools to implement and "
        "verify the smallest correct fix."
    )

    transcript: list[str] = []

    for round_number in range(
        1,
        max_rounds + 1,
    ):
        response = chat.send_message(prompt)
        text = (response.text or "").strip()

        transcript.append(
            f"ROUND {round_number}\n{text}"
        )

        print(
            transcript[-1],
            flush=True,
        )

        if (
            "TASK_COMPLETE" in text
            or "BLOCKED" in text
        ):
            return "\n\n".join(transcript)

        prompt = (
            "Continue from the recorded observations. "
            "If the repair is complete, inspect the diff, "
            "run tests, and return TASK_COMPLETE. "
            "If no safe progress is possible, return "
            "BLOCKED with concrete evidence."
        )

    transcript.append(
        "BLOCKED: maximum agent rounds reached"
    )

    return "\n\n".join(transcript)


if __name__ == "__main__":
    issue = """
The checkout endpoint returns HTTP 500 when coupon_code is omitted.
The stack trace ends at src/services/pricing.py:87 with:
AttributeError: 'NoneType' object has no attribute 'strip'
Fix the bug without changing the behavior for non-empty coupon codes,
add or update a regression test, and run the relevant test suite.
""".strip()

    run_agent(issue)

这里有两层循环。

内层 Function Calling 由 SDK 负责：模型生成结构化工具参数，SDK 调用本地 Python 函数，再把结果作为 Observation 送回模型。

外层循环由应用负责：检查模型是否给出 TASK_COMPLETE 或 BLOCKED，并限制最多轮数。

真正的生产实现还应在每次工具调用前后写审计事件，而不能只保存最终文本。

3. 一次典型修复如何发生

假设错误来自以下代码：

normalized = coupon_code.strip().upper()

Agent 的可观察执行轨迹可能是：

Plan: inspect pricing implementation and existing coupon tests.

Action: read_file(src/services/pricing.py, 60, 110)
Observation: coupon_code is annotated as str | None but strip() is unconditional.

Action: read_file(tests/services/test_pricing.py, 1, 220)
Observation: empty string is covered; None is not covered.

Action: write_file(...)
Observation: changed normalization to guard None and added a regression test.

Action: run_tests(tests/services/test_pricing.py)
Observation: 18 passed.

Action: git_diff()
Observation: only pricing.py and test_pricing.py changed.

TASK_COMPLETE
Changed: src/services/pricing.py, tests/services/test_pricing.py
Evidence: 18 tests passed.

这里不要求模型输出冗长的内部“Thought”。

ReAct 在工程系统中的价值，是让决策与环境交互形成可检查的 Plan -> Action -> Observation，而不是收集模型的私有思维过程。

审计所需的是工具名、参数、结果、代码差异和测试证据。

4. 代码示例中的关键护栏

路径约束：所有路径先 resolve()，再验证其位于工作区内部，防止 ../../ 读取系统文件。

禁用 Shell：subprocess.run 接收参数数组且 shell=False，模型不能把 ;、管道或重定向拼进命令。

工具白名单：示例没有提供任意命令执行器，只暴露完成任务所需的最小工具。若确实需要构建命令，应按项目配置预定义命令 ID，而不是让模型自由生成命令字符串。

提交门禁：默认不能 Git Commit。人工检查 Diff 后设置开关，工具也只暂存本轮由 Agent 写过的文件，避免把工作区原有改动一起提交。

证据化完成：模型只有在测试通过后才能输出完成标记；应用还应独立检查测试退出码，不能只相信模型总结中的“已通过”。

容器隔离：即使工具做了参数校验，也建议在临时容器、独立工作树或短生命周期虚拟机中运行。网络、文件系统、云凭证和依赖安装权限都应默认关闭，再按任务逐项开放。

5. 从教程代码走向生产服务

教程把工具定义成同进程 Python 函数，便于理解 Function Calling。

生产环境更适合把模型编排器与工具执行器分开部署。编排器只持有任务状态和工具描述，执行器运行在低权限沙箱中，并通过带身份的内部协议接收请求。

即使模型层遭遇 Prompt Injection，攻击面也被限制在单个临时环境。

每个工具都应有版本化契约。

例如 read_file.v2 明确最大字节数、编码策略和返回结构；run_tests.v1 只接受预注册测试目标；apply_patch.v1 接受统一 Diff 并在应用前校验上下文。

工具升级时保留旧 Schema，正在运行的任务继续使用创建时的版本，避免中途改变参数含义。

写文件也不应直接覆盖唯一副本。

执行器可以先在独立 Git Worktree 中应用补丁，生成 Diff 后进入验证队列。静态检查、单元测试、依赖漏洞扫描和项目自定义门禁全部通过，才允许创建待审变更。

涉及认证、计费、权限、数据迁移和基础设施的文件，无论测试结果如何都应要求代码所有者审批。

工具结果要区分“业务失败”和“基础设施失败”。

测试断言不通过属于有效 Observation，Agent 应据此修复；测试进程因容器被回收而中断则不说明代码有错，应由调度器恢复。

若不区分，模型可能为了修复基础设施噪声而错误修改业务代码。

最后，完成条件要由独立验证器计算。

模型可以输出 TASK_COMPLETE，但调度器只有在 Diff 非空且范围合规、必需测试退出码为零、没有禁止文件改动、预算未异常且制品保存成功时，才把任务标记为成功。

模型声明只是候选信号，不是最终事实。

五、多模态 Agent：用截图与 DOM 完成 GUI 自动化测试

传统 UI 自动化依赖稳定的 CSS 选择器。

当页面出现弹窗遮挡、按钮视觉错位、文本截断或主题样式错误时，DOM 断言可能全部通过，但用户看到的界面已经不可用。

Gemini 的原生多模态能力可以把截图、DOM 摘要、控制台日志和网络失败记录放到同一个判断中，适合作为 Playwright 或 Selenium 的“视觉分析层”。

推荐的决策顺序是：

1. 优先让 Agent 根据可访问性树、角色和稳定属性选择元素；
2. 其次使用经过校验的 CSS 选择器；
3. 只有无法从 DOM 定位时，才使用截图坐标；
4. 点击前验证目标区域、页面域名和动作风险；
5. 点击后重新截图，并检查页面状态是否符合预期。

坐标点击很脆弱：视口、缩放、DPI、滚动位置或响应式布局一变，旧坐标就会失效。

因此视觉模型应负责“理解页面”，浏览器驱动仍应尽量通过语义定位执行。

下面的示例让 Playwright 截取页面并提取主要 DOM 文本，再要求模型返回结构化 JSON。

它只允许 click、fill、assert_text 和 stop 四种动作，不会直接执行模型返回的任意 JavaScript。

from __future__ import annotations

import json
import os
from typing import Literal

from google import genai
from google.genai import types
from playwright.sync_api import Page, sync_playwright
from pydantic import BaseModel, Field


class UiDecision(BaseModel):
    action: Literal[
        "click",
        "fill",
        "assert_text",
        "stop",
    ]

    selector: str | None = None
    value: str | None = None
    expected_text: str | None = None

    reason: str = Field(
        description=(
            "Brief observable reason, "
            "not chain-of-thought"
        )
    )


def decide_next_action(
    page: Page,
    goal: str,
) -> UiDecision:
    screenshot = page.screenshot(
        full_page=False
    )

    dom_excerpt = page.locator(
        "body"
    ).inner_text(
        timeout=5_000
    )[:20_000]

    client = genai.Client(
        api_key=os.environ["GEMINI_API_KEY"]
    )

    response = client.models.generate_content(
        model=os.environ["GEMINI_MODEL"],
        contents=[
            f"""Goal: {goal}
Current URL: {page.url}
Visible DOM text:
{dom_excerpt}

Choose exactly one safe next action.
Prefer role-, label-, data-testid-, or id-based selectors.
Never approve purchases, delete data, change account security,
or leave the allowed origin.
Return stop when human approval is needed.
""",
            types.Part.from_bytes(
                data=screenshot,
                mime_type="image/png",
            ),
        ],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=UiDecision,
            temperature=0.1,
        ),
    )

    if getattr(
        response,
        "parsed",
        None,
    ):
        return response.parsed

    return UiDecision.model_validate_json(
        response.text
    )


def execute_decision(
    page: Page,
    decision: UiDecision,
    allowed_origin: str,
) -> None:
    if not page.url.startswith(
        allowed_origin
    ):
        raise RuntimeError(
            "browser left the allowed origin"
        )

    if decision.action == "stop":
        raise RuntimeError(
            f"agent stopped: {decision.reason}"
        )

    if not decision.selector:
        raise ValueError(
            "selector is required"
        )

    locator = page.locator(
        decision.selector
    ).first

    if (
        locator.count() != 1
        or not locator.is_visible()
    ):
        raise RuntimeError(
            "selector is not uniquely visible"
        )

    if decision.action == "click":
        locator.click(timeout=5_000)

    elif decision.action == "fill":
        locator.fill(
            decision.value or "",
            timeout=5_000,
        )

    elif decision.action == "assert_text":
        actual = locator.inner_text(
            timeout=5_000
        )

        if (
            decision.expected_text or ""
        ) not in actual:
            raise AssertionError(
                {
                    "expected": decision.expected_text,
                    "actual": actual,
                }
            )


def run_login_check() -> None:
    allowed_origin = (
        "http://127.0.0.1:3000"
    )

    goal = (
        "Log in with the test account "
        "and verify the dashboard heading "
        "is visible"
    )

    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(
            headless=True
        )

        page = browser.new_page(
            viewport={
                "width": 1440,
                "height": 900,
            }
        )

        page.goto(
            f"{allowed_origin}/login",
            wait_until="networkidle",
        )

        for _ in range(8):
            decision = decide_next_action(
                page,
                goal,
            )

            print(
                json.dumps(
                    decision.model_dump(),
                    ensure_ascii=False,
                )
            )

            execute_decision(
                page,
                decision,
                allowed_origin,
            )

        browser.close()


if __name__ == "__main__":
    run_login_check()

这段代码仍只是骨架。

真实 GUI Agent 还应补充测试账号注入、敏感字段脱敏、页面域名白名单、下载禁用、弹窗处理和动作录像。

对于“删除”“购买”“发送消息”“修改权限”等不可逆动作，模型只能提出建议，必须进入人工确认状态。

在视觉回归中，Gemini 不应取代像素级或结构化断言，而应处理传统断言难以表达的问题。

例如：

• 按钮虽然存在但被浮层遮挡；
• 错误提示与输入框距离过远；
• 移动端导航挤压正文；
• 浅色文字在背景上几乎不可见。

最佳组合是三层验证：DOM 断言负责确定性行为，截图差异负责像素变化，多模态模型负责语义解释与缺陷归类。

多模态 Agent 同样需要离线评测。

可以从真实缺陷中建立截图集，为每张图保存视口、浏览器版本、DOM 快照、预期缺陷类型和人工标注区域。

评测时分别统计元素定位成功率、危险动作误触率、缺陷召回率和误报率。

只展示几张“识别很准”的截图无法说明系统可靠，因为登录弹窗、骨架屏、动画、Canvas、跨域 iframe 和相似按钮都会显著增加难度。

对于坐标型结果，最好要求模型返回归一化边界框，再由应用换算为当前视口坐标，并检查中心点是否落在可见、可交互元素上。

若视觉位置与 DOM 命中不一致，应停止而不是强行点击。

对关键流程，还可让第二个只读审查步骤检查“当前截图、建议动作、预期结果”三者是否一致，再交给执行器。

六、生产落地：高频调用下的并发控制与高可用网关

Agent 和普通聊天接口的流量形态不同。

一次用户任务可能触发十几次模型请求、数十次工具调用以及多轮长文本传输。当多个任务同时运行时，RPM、TPM、并发连接数和下游工具容量会一起成为瓶颈。

常见故障包括：

• 429 Rate Limit
• 请求超时
• 流式连接中断
• TLS 握手失败
• 重试风暴

最危险的处理方式，是收到 429 后立即无上限重试。

若 100 个任务同时退避相同时间，它们会在下一秒再次同时冲击服务。

正确做法是把并发治理放在 Agent 之外：

控制层	解决的问题	实现要点
有界任务队列	瞬时洪峰	队列长度上限、优先级、过载拒绝
并发信号量	连接和执行槽耗尽	按模型、租户、工具分别限流
Token 预算	TPM 与成本失控	预估输入、保留输出、超限降级
指数退避 + Full Jitter	429 和短暂网络故障	尊重服务端等待提示，随机化重试时间
熔断器	持续故障扩大	达阈值后快速失败，半开探测恢复
检查点	长任务中断	持久化状态，恢复时避免重复副作用
幂等键	重试导致重复操作	由业务应用生成并在执行器校验

如果团队需要减少海外 Endpoint 连接抖动，也可以把 https://178.nz/dn 这类免配置测试 API 网关纳入候选方案，但不要仅凭宣传判断其高可用能力。

应使用脱敏的长 Payload、流式输出、并发阶梯和故障注入做独立压测，同时核查数据保留、日志脱敏、协议兼容、超时、重试、配额与故障转移策略。

涉及源代码、客户数据或密钥时，还要先通过安全与合规评审。

下面是一个简化的异步调度器。信号量控制并发，指数退避加入随机抖动，检查点回调在每轮后持久化状态：

import asyncio
import random
from collections.abc import Awaitable, Callable


class RetryableAgentError(Exception):
    pass


async def run_with_limits(
    task_id: str,
    step: Callable[
        [],
        Awaitable[dict],
    ],
    save_checkpoint: Callable[
        [str, dict],
        Awaitable[None],
    ],
    semaphore: asyncio.Semaphore,
    max_attempts: int = 5,
) -> dict:
    for attempt in range(max_attempts):
        try:
            async with semaphore:
                result = await asyncio.wait_for(
                    step(),
                    timeout=90,
                )

            await save_checkpoint(
                task_id,
                result,
            )

            return result

        except RetryableAgentError:
            if attempt == max_attempts - 1:
                raise

            cap = min(
                30.0,
                1.5 * (2**attempt),
            )

            await asyncio.sleep(
                random.uniform(0, cap)
            )

    raise RuntimeError("unreachable")

重试策略必须区分错误类型。

参数错误、鉴权失败、内容策略拒绝和工具输入不合法通常不应重试；429、网关 502/503、连接重置和部分超时才适合有限重试。

对于写数据库、发送邮件、创建订单这类副作用，重试前必须查询幂等记录，不能把“网络没有返回”误判成“操作没有执行”。

长上下文还会放大成本与尾延迟。可以采用四种手段控制：

1. 稳定材料使用缓存；
2. 工具日志只保留关键窗口并链接原始制品；
3. 在阶段边界生成可验证摘要；
4. 把轻量分类和格式转换交给更低延迟模型，把跨文件推理与最终审查留给能力更强的模型。

模型路由规则应由应用配置，而不是让 Agent 自己决定无限升级。

生产观测至少应覆盖：

• 每个任务的模型请求数；
• 输入与输出 Token；
• 工具调用数；
• P50、P95、P99 延迟；
• 429 比例；
• 重试次数；
• 上下文缓存命中率；
• 人工接管率；
• 测试通过率；
• 单位成功任务成本。

只有同时观察质量、延迟和成本，才能判断 Agent 是否真的提高了生产力。

高可用也不等于“请求一定成功”。

长流式响应在连接中断后通常不能从任意 Token 位置无损续传，因此恢复单位应是业务步骤，而不是网络字节。

每一步开始前记录输入哈希和幂等键，完成后记录输出制品与状态；连接断开时，调度器先查询该步骤是否已经产生结果，再决定读取、重试或交给人工处理。

网关评估不能只测一次短文本延迟。

应覆盖不同上下文长度、首 Token 时间、完整响应时间、流式中断率、并发 1/10/50/100 的错误曲线、429 头信息透传、工具调用 JSON 完整性以及上游错误码是否被原样保留。

若网关把所有错误都包装成 HTTP 200，Agent 将无法正确区分重试与参数修复；若自动重试写操作，又可能制造重复副作用。

还要为租户设置独立预算。

全局信号量能保护服务，却不能防止一个大型任务耗尽全部配额。按租户限制并发、RPM、TPM 和日成本，再设置系统级总上限，才能兼顾公平与稳定。

队列中的任务应带截止时间，过期任务直接取消，不能在恢复后执行已经失去业务意义的动作。

七、总结与展望：百万级工作记忆将如何改变软件架构

Gemini 1.5 Pro 的历史意义，在于让开发者直观地看到：当模型可以同时阅读大量代码、日志、文档和多模态输入时，Agent 不必在每一步都依赖碎片化摘要。

后续 Gemini 模型与新版 Google GenAI SDK 延续了这条路线，使 Function Calling、长上下文和多模态更容易组合成完整工作流。

但长上下文不会自动带来长期记忆，也不会让 RAG、数据库和权限系统消失。

更可能出现的架构是：

• 数据库保存事实与状态；
• 检索系统缩小候选范围；
• 长上下文保持任务内的完整因果链；
• 模型负责提出下一步动作；
• 确定性工具负责执行；
• 测试与策略门禁负责裁决是否完成。

真正可靠的 Agent，不是能够自主执行最多动作的系统，而是能够在正确边界内执行、在证据不足时停止、在故障后恢复，并让人清楚知道它改了什么、为什么改、验证结果是什么。

当 Agent 拥有百万级工作记忆后，软件开发会从“人逐条操作工具”逐渐转向“人定义目标、边界和验收标准，Agent 编排工具并提交证据”。

这不会消除工程师，反而会把架构设计、测试策略、权限治理和故障分析推到更重要的位置。

你在搭建 Agent 时遇到过哪些问题：上下文失控、工具死循环、RAG 召回漂移，还是高并发下的 429？

你认为向量数据库会被超长上下文取代，还是会成为 Agent 的长期记忆索引？欢迎在评论区交流。