深入解析 screenshot-to-code 的 Agentic Runner 重构：统一流式循环、Provider 适配器与规范化工具定义

screenshot-to-code 是一个“丢一张截图，生成 HTML/Tailwind/React/Vue 代码”的开源项目，其后端生成链路已经全面 Agentic 化：模型通过 create_file、edit_file、generate_images、screenshot_preview 等工具迭代地产出、修改并验证页面。本文以仓库中的设计文档 Agentic Runner Refactor Spec 为主线，完整还原这次“Agentic Runner 重构”的设计目标、两项核心决策（统一流式循环 + Provider 适配器、规范化工具定义 + 序列化层）、计划移除项与模块拆分方案，并结合当前仓库中 backend/agent/ 目录下的真实实现逐层佐证，帮助读者掌握“多 LLM 供应商共享一条 Agent 流水线”的架构设计与落地细节。

一、重构背景与目标

在重构之前，screenshot-to-code 需要同时支持 OpenAI、Anthropic、Gemini 三家模型供应商。每家供应商的流式协议、消息格式、工具调用载荷都不同，导致流式处理逻辑（解析增量文本、聚合工具调用参数、向 WebSocket 推送事件）在三个 runner 中各自实现一遍。设计文档给出了三条清晰的 Goals：

1. 减少 OpenAI/Anthropic/Gemini runner 之间重复的流式逻辑（Reduce duplicated streaming logic）；
2. 集中管理工具 schema 与遥测（telemetry）格式化（Centralize tool schemas and telemetry formatting）；
3. 让 Agent 流水线更易测试、扩展和推理（Make the agent pipeline easier to test, extend, and reason about）。

围绕这三个目标，文档确立了“统一流式循环 + Provider 适配器”和“规范化工具定义 + 序列化层”两个核心决策，随后列出了若干计划移除项和模块拆分方案。下文逐节展开，并对照当前仓库实现说明每个决策如何落地。

二、核心决策一：统一流式循环 + Provider 适配器

文档 Decision 章节明确要求：

• 引入一个 provider 无关的流式循环（provider-agnostic stream loop），它只消费一组归一化事件（normalized events）；
• 为每个供应商编写小颗粒度适配器（per-provider adapters），把各自的 native stream 翻译成归一化事件，适配器“保持小巧、专注于解析供应商特定的载荷”；
• 工具执行、toolStart/toolResult 事件发射、setCode 预览流式输出统一集中在统一循环内；
• 各供应商适配器保持小而专注。

2.1 归一化事件协议：StreamEvent 与 ProviderSession

在仓库当前实现中，这套归一化协议定义在 provider 协议文件。其中 StreamEventType 是一个 Literal 类型，收敛为三种增量事件：

StreamEventType = Literal[
    "assistant_delta",      # 助手文本增量
    "thinking_delta",       # 思考/推理内容增量
    "tool_call_delta",      # 工具调用参数增量
]

@dataclass
class StreamEvent:
    type: StreamEventType
    text: str = ""
    tool_call_id: Optional[str] = None
    tool_name: Optional[str] = None
    tool_arguments: Any = None

与文档中列举的五个事件（assistant_delta、thinking_delta、tool_call_delta、tool_call_complete、done）相比，从源码结构看，当前实现把“工具调用完成”和“回合结束”两类语义上移到了协议返回值的层面：适配器不再发出 tool_call_complete/done 事件，而是由 stream_turn() 直接返回一个结构化的 ProviderTurn，其中携带本回合的完整文本与已聚合好的工具调用列表：

@dataclass
class ProviderTurn:
    assistant_text: str
    tool_calls: list[ToolCall]
    # Provider-native assistant turn object required to continue the conversation.
    assistant_turn: Any = None

EventSink = Callable[[StreamEvent], Awaitable[None]]

class ProviderSession(Protocol):
    async def stream_turn(self, on_event: EventSink) -> ProviderTurn: ...
    async def append_tool_results(
        self,
        turn: ProviderTurn,
        executed_tool_calls: list[ExecutedToolCall],
    ) -> None: ...
    def total_cost_usd(self) -> Optional[float]: ...
    async def close(self) -> None: ...

这一设计有两个值得注意的点。其一，ProviderSession 是一个 typing.Protocol，三个供应商的 session 类都以结构化的方式满足它，统一循环无需 import 任何具体供应商实现；其二，assistant_turn 字段要求适配器保留“供应商原生的 assistant 回合对象”，用于在多轮工具调用后继续对话——这正是三家 API（OpenAI Responses 的 output items、Anthropic 的 content blocks、Gemini 的 parts）继续会话方式各不相同之处，协议把这一差异封装在了每个适配器内部。total_cost_usd() 则把成本核算也统一到了协议层，为后文的预算熔断提供了入口。

2.2 统一流式循环：AgentEngine._run_with_session

文档要求“把工具执行、toolStart/toolResult 发射、setCode 预览流式输出集中到统一循环里”，这部分的实现位于 Agent 引擎。核心方法 _run_with_session（L205-L314）实现了文档所描述的主循环：

1. 事件分发：每一回合构造一个 on_event 回调，按事件类型把 assistant_delta 经 WebSocket 发送为 assistant 消息、thinking_delta 发送为 thinking 消息、tool_call_delta 交给 _handle_streamed_tool_delta 处理 create_file 的实时预览。这正是文档“统一循环消费归一化事件”的直接体现，也是前端不需要感知供应商差异的前提。
2. 回合结束判定：turn = await session.stream_turn(on_event) 返回后，若无工具调用则进入 _finalize_response 收尾；否则进入工具执行阶段。
3. 预算熔断：工具调用前检查 session.total_cost_usd()，一旦超过 config 中的 GENERATION_MAX_COST_USD 就抛出 BudgetExceededError。该异常的用户提示语是固定的 "Generation stopped: this variant exceeded its resource limit."，不暴露具体金额，精确花费记录在运行记录（AgentRunRecorder）中——这是统一协议层带来的、单一供应商时代难以做到的横向治理能力。
4. 工具执行与遥测集中：对每个工具调用，统一循环负责发送 toolStart（附带规范化的输入摘要）、对 create_file 执行 _stream_code_preview（把完整 HTML 分片流式推给前端，最多 18 片、每片间隔 10ms 的 setCode 预览）、调用 self.tool_runtime.execute(tool_call) 执行工具、发送 toolResult（附带 ok 标志与摘要），最后把全部 ExecutedToolCall 通过 session.append_tool_results 回传给供应商适配器，进入下一轮。
5. 回合上限：max_steps = 30，超过即抛出 "Agent exceeded max tool turns"，防止工具调用无限循环。

整个循环中没有任何一处 if provider == "openai" 之类的分支——供应商差异全部被隔离在适配器内部，这就是文档第一条 Goal（消除重复流式逻辑）的落点。

三、Provider 适配器：把原生流翻译成归一化事件

3.1 工厂：按模型创建会话

会话工厂 的 create_provider_session 是统一循环与供应商世界的唯一连接点。它的逻辑分两步：

1. 先构建一次规范化工具定义（见第四节），并按能力开关裁剪：generate_images 受 should_generate_images 控制，edit_image 需要 Replicate API key，extract_assets 需要 Gemini key，screenshot_preview 依赖无头 Chromium 可用（is_screenshot_preview_available()）。
2. 按模型路由到具体 session：OPENAI_MODELS 走 OpenAIProviderSession（AsyncOpenAI 客户端），ANTHROPIC_MODELS 走 AnthropicProviderSession（AsyncAnthropic），GEMINI_MODELS 走 GeminiProviderSession（genai.Client），并对每家做 API key 校验。三家 session 在创建时各自接收一份“已按该供应商序列化”的工具列表。

3.2 OpenAI 适配器：只走 Responses API

文档“Planned Removals”一节要求移除遗留的 ChatCompletion 流式路径，所有 OpenAI 模型一律走 Responses API。当前仓库的实现印证了这一目标已落地：OpenAI 适配器 中发起请求的唯一方式是 self._client.responses.create(**params)（L465），请求参数包含 max_output_tokens: 50000、tool_choice: "auto"，并按模型追加 reasoning（effort + summary）与 prompt_cache_retention 等 Responses 特有参数。

适配器的核心是一个无状态的 parse_event(event, state, on_event) 函数（L173 起），把 Responses 流中的原生事件翻译成归一化事件：

原生事件	归一化事件
response.output_text.delta	assistant_delta
response.reasoning_text.delta / response.reasoning_summary_text.delta	thinking_delta
response.output_item.added（function_call/custom_tool_call）	tool_call_delta（带初始参数）
response.function_call_arguments.delta / response.mcp_call_arguments.delta / response.custom_tool_call_input.delta	tool_call_delta（累加参数 JSON）
response.function_call_arguments.done 等	tool_call_delta（参数定稿）
response.completed	提取 usage（并从中扣除 cached tokens 得到未缓存输入量）

参数聚合状态放在 OpenAIResponsesParseState 中，回合结束后由 _build_provider_turn 组装 ProviderTurn：优先从 output_items_by_index 里的 function_call item 提取工具调用；若 JSON 解析失败，则把原始参数包进 {"INVALID_JSON": ...} 交给上层处理，而不是直接崩溃。append_tool_results 则把工具结果封装成 function_call_output item 追加到输入流，并支持把工具产出的图片以 input_image part 形式回传给模型（文本结果在前、图片在后，保证模型同时拿到结构化数据与图像）。

3.3 Anthropic 适配器：thinking 与 tool_use 块

Anthropic 适配器 通过 _parse_stream_event（L209 起）消费 Anthropic 的 content_block_start / content_block_delta：thinking_delta 块映射为归一化的 thinking_delta；tool_use 块在 content_block_start 时注册、其 input JSON 增量聚合为 tool_call_delta。此外该适配器还处理了 Claude 多图请求的尺寸限制（CLAUDE_MANY_IMAGE_THRESHOLD，超过 20 张图阈值时对本地 base64 图片做降采样），并有独立的模型-参数映射表（ANTHROPIC_MODEL_CONFIG）把内部的 effort 级别（low/medium/high/xhigh/max）翻译成 API 的模型名与 thinking 配置。

3.4 Gemini 适配器

Gemini 适配器 遵循同一 ProviderSession 协议。除工具序列化外，它还内嵌了视频输入的默认帧率参数（DEFAULT_VIDEO_FPS = 10），这与文档“Planned Removals”中视频生成统一走 agent runner 的决策相呼应——视频素材不需要绕过 Agent 流水线的特供分支，而是作为普通输入进入统一循环（详见第五节）。

四、核心决策二：规范化工具定义 + 序列化层

文档第二个 Decision 要求：工具 schema 只定义一次（canonical representation），再经由序列化辅助函数为 OpenAI Responses、Anthropic、Gemini 各生成一份；同时集中工具输入/输出摘要，让 UI 遥测在各供应商间保持一致、消除重复。

4.1 规范化表示：CanonicalToolDefinition

规范化类型定义在 工具类型文件：

@dataclass(frozen=True)
class CanonicalToolDefinition:
    name: str
    description: str
    parameters: Dict[str, Any]   # 标准 JSON Schema

配套的还有统一的 ToolCall（id/name/arguments）、ToolExecutionResult（ok/result/summary/updated_content/multimodal_parts）与 ToolMultimodalPart。后者用不变式（__post_init__ 强制 data 与 image_url 二者必居其一，且 image_url 不允许是 localhost）防止了“本地图片 URL 被发给无法回源的 OpenAI/Anthropic”这类隐性错误。

canonical_tool_definitions 是唯一构建工具清单的入口，接收四个能力开关并据此裁剪工具集：

工具	条件	参数要点
create_file	恒开	path（缺省 index.html）、content（必填），用于一次性写入完整 HTML
edit_file	恒开	old_text/new_text/count（-1 表示全部）或批量 edits 数组
generate_images	image_generation_enabled	prompts 数组，可一次生成多张
remove_background	恒开	image_urls 数组
edit_image	image_editing_enabled（需 Replicate key）	prompt、image_urls（主图在前）、aspect_ratio（枚举，默认 match_input_image）
extract_assets	asset_extraction_enabled（需 Gemini key + 请求中确含静图）	asset_descriptions 数组，每个元素精确描述截图中一处视觉资产
screenshot_preview	screenshot_enabled（无头 Chromium 可用）	无参数；渲染当前 HTML 返回桌面/移动端整页截图
save_assets	恒开（定义在 uploaded_assets 工具模块）	保存用户上传的图片资产
retrieve_option	恒开	option_number（1 起始），取回某变体的完整 HTML 供参考

值得注意的一个细节：extract_assets 的启用条件在 引擎入口 中被进一步收紧——只有当请求消息里确实能提取出 data:image/ 前缀的静图 data URL 时才向模型暴露该工具，因为视频片段虽然共享 image_url 消息结构却不是合法的裁剪输入。

4.2 三家序列化器：同一 schema 的三种方言

OpenAI 序列化（serialize_openai_tools）最复杂。Responses API 的 strict: true 模式对 JSON Schema 有严格约束，因此 _make_responses_schema_strict（L90-L118）会递归改写 schema：给 object 节点补上 additionalProperties: false 和完整 required 列表；把 object 属性下的标量类型改写为 [type, "null"] 的可空列表形式（array 类型仅在作为 object 属性时才可空）。最终每个工具输出为 {"type": "function", "name", "description", "parameters", "strict": true}。

Anthropic 序列化（serialize_anthropic_tools）直接透传深拷贝的 parameters 为 input_schema，并额外开启 eager_input_streaming: True——这让 tool_use 块在开始时就尽早携带 input 增量，直接支撑了统一循环里“create_file 参数边生成边预览”的体验。

Gemini 序列化（serialize_gemini_tools）把每个工具包装成 types.FunctionDeclaration（parameters_json_schema 直接复用规范化 schema），再包进单个 types.Tool(function_declarations=...)。

对比三个实现可以看到：差异全部局限在“方言翻译”层，schema 语义本体只维护一份——新增一个工具或修改参数时只需改 definitions.py 一处。

4.3 集中的输入摘要：UI 遥测一致性

文档要求“集中工具输入/输出摘要以跨供应商保持 UI 遥测一致”。实现位于 summaries.py 的 summarize_tool_input：统一循环在发送 toolStart 事件前调用它，把每个工具的大参数压缩成前端可展示的小摘要——create_file 输出 {path, contentLength, preview(200 字符)}，edit_file 输出每条 edit 的 160 字符截断与 count，generate_images/remove_background/edit_image/extract_assets 输出 count 加参数列表。前端 Agent 活动面板 消费的正是这些跨供应商同构的 toolStart/toolResult 结构，这也呼应了文档 Non-Goals 一节“不重新设计前端活动 UI，前端继续消费相同的 tool/assistant/thinking 事件”的承诺。

五、文件状态与会话播种：agent/state.py

文档 File/Module Split 中的 agent/state.py（file state + seeding utilities）对应 状态模块。其中 AgentFileState 以 (path, content) 两个字段承载“当前主 HTML 文件”的内存状态，create_file/edit_file 工具都直接读写它；统一循环的 _finalize_response 也以它作为最终产物来源（若模型最终没有调用工具，则从助手文本里兜底提取 HTML）。

seed_file_state_from_messages（L32-L70）解决“更新已有代码”场景下的状态播种：引擎在 run() 开头调用它，优先从最近一条 assistant 消息中提取 HTML；若无，则回退到 system 消息中以 "Here is the code of the app:" 标记之后的内容。这使得迭代修改（update）请求无需工具调用就能从提示词历史中恢复文件状态，与 更新提示词构建 的链路配合。

六、计划移除项：现状与证据

文档“Planned Removals”一节列出三项移除/收敛，以下结合当前仓库逐项说明其动机与现状。

6.1 移除 Agent 工具层的图片缓存

• 移除对象：agent 工具层的“prompt 到 URL 的图片缓存”（prompt-to-URL image caching）。
• 理由：简化状态、减少隐藏的跨变体（cross-variant）耦合——变体并发生成时，按 prompt 缓存图片 URL 会让不同变体意外共享同一资源，状态难以推理。
• 后续约束：文档明确后续要求“图片生成在需要时保持按请求确定性（例如显式传 seed，或把缓存上移到更高层）”。换言之，缓存并未被无条件否定，而是被逐出了工具层这个不该承担它的层级。

6.2 移除 OpenAI ChatCompletion 路径

文档要求删除遗留的 ChatCompletion 流式路径，所有 OpenAI 模型改走 Responses API，并同步更新模型列表与运行时检查以消除 ChatCompletion 分支。当前 OpenAI 适配器 中不存在任何 client.chat.completions 调用，请求入口只有 client.responses.create，消息转换也是专门的 _convert_message_to_responses_input（把 text/image_url part 转为 Responses 的 input_text/input_image，并按模型选择 detail: high 或 original），印证该移除项已经完成。

6.3 视频等非 Agent 生成路径并入 Agent Runner

文档要求保留视频生成能力，但移除“绕过 agent 路径”的视频专用流式分支，把视频的提示词与媒体处理整合进统一的工具/流式流水线。当前实现中可以看到这一收敛的痕迹：引擎 在提取输入图片时显式排除非图片媒体（注释说明 Gemini 视频共享 image_url 结构但不是合法的 extract_assets 输入，纯视频请求不应暴露必然失败的工具）；Gemini 适配器 内嵌视频默认帧率参数，说明视频素材是在 provider 输入层被规范化后进入统一循环的，而不是另起一条生成链路。文档同时要求相应更新测试与文档以体现“支持视频输入的单一条 agent 生成路径”。

七、模块拆分：设计稿与最终落地

文档给出的 File/Module Split 如下，对照当前仓库结构可以看到落地时的演进：

设计稿	职责	当前仓库对应实现
agent/runner.py	编排 + 共享流式循环	主循环实现在 engine.py 的 AgentEngine；runner.py 保留为 class Agent(AgentEngine): pass 的兼容别名
agent/providers/	provider 适配器（openai, responses, anthropic, gemini）	providers 目录：base.py 协议、factory.py 工厂、openai.py、anthropic/、gemini.py；OpenAI 因“Responses API 唯一路径”决策不再区分 responses/chatcompletion 两套文件
agent/tools.py	工具定义、序列化、执行	演进为 tools 包：definitions.py（定义）、runtime.py（执行）、summaries.py（摘要）、parsing.py（参数解析）、screenshot_preview.py/extract_assets.py/local_assets.py（具体工具）
agent/state.py	文件状态 + 播种工具	state.py，与设计稿一致

可以看出最终落地相比设计稿做了两处“长大”：编排逻辑独立到 engine.py（runner.py 退化为兼容层），工具模块因工具数量增多而从单文件扩展成包。两者都未偏离文档的拆分意图——“循环一处、适配器一层、工具一处、状态一处”。

Non-Goals 一节同样得到遵守：除移除项外无功能性 UX 变更，前端 Agent 活动组件 依旧消费同一套 assistant/thinking/toolStart/toolResult/setCode WebSocket 事件，供应商切换对 UI 完全透明。

八、测试布局：重构“可测试性”目标的验证

文档第一条 Goal 把“可测试”列为重构目标，仓库的测试文件分布与之高度对应，每个供应商适配器与统一循环都有独立可测的单元：

• test_agent_engine.py：统一循环（事件分发、工具执行、回合上限、收尾）；
• test_agent_tool_runtime.py 与 test_agent_tools.py：规范化工具定义与执行运行时；
• test_openai_provider_session.py：OpenAI Responses 适配器的事件解析与 ProviderTurn 构建；
• test_anthropic_provider_config.py、test_anthropic_many_image_limit.py：Anthropic 适配器的配置映射与多图限制；
• test_gemini_provider_config.py、test_gemini_provider_session.py：Gemini 适配器；
• test_openai_input_compare.py 与 test_evals_openai_input_compare.py：对 OpenAI 请求输入做跨实现对比，正是“ChatCompletion 路径移除后统一走 Responses”这一决策的回归保障；
• test_openai_reasoning_parser.py：reasoning/thinking 增量解析。

九、总结：这套架构给多供应商 Agent 系统的启示

回顾 Agentic Runner Refactor Spec 的全脉络，screenshot-to-code 用一次结构化重构把“三家供应商各写一套流式逻辑”收敛为四层清晰的分层：

1. 归一化事件协议层（StreamEvent 三事件 + ProviderTurn 回合值 + ProviderSession 协议，base.py）：把“流结束”“工具完成”等终态语义放入返回值而非事件流，简化了事件枚举；
2. 统一循环层（engine.py 的 _run_with_session）：事件分发、预算熔断、工具执行、toolStart/toolResult/setCode 遥测全部集中于此，30 回合上限兜底；
3. 适配器层（providers/）：每个供应商一个解析器，把 native stream 翻译成归一化事件，并负责各自 API 的消息续写（append_tool_results）；
4. 规范化工具层（tools/）：schema 单点定义、三种方言序列化、输入摘要统一格式化。

这套设计的直接收益是：新增一家供应商只需实现 ProviderSession 四个方法；新增一个工具只需在 canonical_tool_definitions 加一条定义并在 summarize_tool_input 补一个摘要分支；而预算控制、工具遥测、回合上限等横切能力则自动作用于所有供应商。对需要同时对接多家 LLM 的 Agent 类项目（截图生码、代码补全、多模型路由网关），这都是一套可以直接借鉴的参考实现。