Tianyao' s BLOG

Tianyao

0%

Agent驱动的GIS客户端OpenGIS：harness的架构升级

发表于 更新于 分类于
本文字数： 22k 阅读时长 ≈ 39 分钟

OpenGIS 完成了一次重大的架构升级，从CodeAct全面转向了function-call，并且在harness方面重新设计，做了大量优化以对齐主流的agent，因此，通过本文对当前新架构做一个系统性的总结，主要聚焦于 harness 部分。
项目仓库：https://github.com/ATFfang/OpenGIS.git

图1：使用OpenGIS进行数据可视化与数据分析

图2：使用OpenGIS可视化OD数据与自主分析

图3：利用WorkFlow功能完成可控的长工作流

图4：内置与可新增、可编辑、可调用的原子化GIS操作

图5：使用OpenGIS展示数据大屏

OpenGIS Demo
Watch Demo

总体架构

进程模型

在前文中，我们介绍了 OpenGIS 的总体架构，简单来说，分为主进程 Electron Main，地图与聊天等前端渲染进程 Renderer，用以Agent与空间分析等的 python运行进程 Python Sidecar。Python Sidecar 与前端通过 WebSocket 通信，大体上如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
┌─────────────────────────────────────┐
│ Electron Main (Node.js)             │
│   窗口、菜单、文件系统、Sidecar 生命周期│
└──────────────┬──────────────────────┘
               │ spawn
┌──────────────▼──────────────────────┐
│ Renderer (React + MapLibre)         │
│   地图渲染、聊天 UI、图层状态         │
└──────────────┬──────────────────────┘
               │ WebSocket (JSON-RPC 2.0)
┌──────────────▼──────────────────────┐
│ Python Sidecar (FastAPI + LiteLLM)  │
│   Agent 循环、工具运行时、Worker 管理 │
└─────────────────────────────────────┘

三个进程的职责边界为：

进程 技术 职责
Electron Main Node.js 窗口、菜单、文件系统、Sidecar 进程管理
Renderer React + MapLibre 地图渲染、聊天 UI、图层与工作区状态
Python Sidecar FastAPI + LiteLLM Agent 循环、工具运行时、空间分析、Worker

进程间通信仅通过双向的 WebSocket（JSON-RPC）执行：

  • Renderer → Sidecar：chat.user_message、rpc.code.run_script、rpc.agent.* 等请求
  • Sidecar → Renderer：rpc.ui.map.* 反向通知，由前端处理后写 store、驱动地图

通信链路本身（Electron 启动 Sidecar、token 认证、OPENGIS_READY 就绪信号）属于部署细节，与 harness 结构无关，这里不展开介绍。进程模型里有一个对 Agent 架构影响深远的决策：地图状态属于前端。Sidecar 侧没有任何代码持有 MapLibre 实例，如果 Agent 想控制底图，只能发送 rpc.ui.map.* 通知（如 add_layer_from_geojson、dynamic_layer_update），前端 store 接收后统一同步到底图。

Agent 结构

Agent Loop 的本质（loop 的四个步骤）

Anthropic 在 Building Effective Agents 中把 Agent 定义为模型自主驱动的决策过程：执行过程中每一步都要从环境拿到 ground truth（工具调用结果或代码执行输出）来评估进展，并需要停止条件（最大迭代次数之类）维持控制。

Lilian Weng 总结的 ReAct 范式给出同样的循环形状：推理（thought）→ 行动（action）→ 观察（observation）。主流的 agent 框架，无论实现多复杂，循环都可以归约成这个形状。OpenGIS 的循环在代码里对应四个步骤：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# AgentLoop.run() 主循环（伪代码）
while True:
    # ① 构造：把系统提示词、历史、工具 Schema 拼成一条请求
    #    ContextManager.build_provider_request() → messages
    #    ToolMaterializer.materialize()          → tools
    outcome = kernel.run_turn(LoopTurnRequest(...))
    # ② 发送：ProviderTurnCaller.call(messages, tools) → LLMResponse

    # ③ 解析：把响应拆成两个东西
    response_text = outcome.provider_result.response_text   # 纯文本
    tool_calls    = outcome.provider_result.tool_calls      # 结构化工具调用

    if tool_calls:
        # ④ 处理：逐个执行工具，结果写回 context
        ToolCallSettler.settle_all(tool_calls)
        RuntimeControl.observe_settlements(...)
        continue    # 工具结果成为下一轮构造的输入
    else:
        return response_text    # 纯文本即最终答案

循环有构造和解析两个方向：

  • 构造方向（①②）把各层的产物文本化，塞进 messages 和 tools；
  • 解析方向（③④）把响应还原成 response_text 和 tool_calls，执行后把结果回流为下一轮的输入。

构造方向有一个统一的终点：ContextManager.build_provider_request() 把各层内容拼成一条请求，序列化后就是发送给 provider 的 token，这条请求的物理布局（稳定前缀与动态尾部的切分）在第三部分展开。

Agent的九层分层

图6：Agent的九层分层

Agent 本体全部在 python-backend/opengis_backend/agent/ 下，如上图所示，按职责分成九层：

层 代码路径 职责
AgentProfile governance/profile.py 声明 Agent 身份：模式、工具组、权限、预算
AgentFactory agent_factory.py + factory_common.py 每 run 装配一次完整运行时
AgentLoop loop/agent_loop.py 外层策略循环，决定继续或终止
LoopKernel loop/loop_kernel.py 一次 LLM 调用周期的纯执行机制
LoopPolicy loop/policy.py 预算硬约束，超限强制终结
RuntimeControl loop/runtime_control.py 行为软治理，事前拦截、事后纠偏
ToolRuntime execution/tool_runtime.py 工具执行流水线：校验、权限、执行、截断
ToolMaterializer execution/tool_materializer.py 决定本轮模型可见哪些工具 Schema
ContextManager context/context_manager.py 上下文组装、token 预算、协议消毒

按照职责的类别归纳，我们可以精简一下分类为四类：

  1. 身份与装配：AgentProfile、AgentFactory，用于服务 ① （消息构造）
  2. 循环与执行：AgentLoop、LoopKernel，用于服务 ②③④（发送、解析与处理），LoopKernel 内部还含解析与结算的子链
  3. 治理：LoopPolicy、RuntimeControl，在 ③④ （解析与处理）之后决定继续或终止
  4. 工具与上下文：ToolRuntime、ToolMaterializer、ContextManager，服务 ① 构造与 ④ 回流

在下文中，我们会按照这四个类别，详细的展开 OpenGIS harness 结构的介绍。

装配

上一章节中，我们阐述了 OpenGIS harness 结构两个维度的概念，在Loop层面，遵守构造-发送-解析-处理的四步循环，在代码层面，则是分为了九个模块，用于处理特定的工作。为了更好的阐述一个loop的流程，我们按照loop层面的四步循环来分段阐述。不过，本章节先介绍一下四步循环的前序步骤：装配。

装配是一次 run 的起点，是 run 级流程，发生在循环之前，而四步循环是 turn 级流程。装配的目的是为一次 run 构造完整的运行时，其会构造AgentLoop 和子进程 executor两项产物。其中，AgentLoop 是循环引擎，它需要一组具体的依赖才能工作：

  • 调用 LLM 的通道（llm_call）
  • 执行 Python 的通道（executor_call）
  • 系统提示词（system_prompt）
  • 工具运行时与工具 Schema（tool_runtime / tool_schemas）
  • 工具物化器（tool_materializer）
  • 上下文管理器（context）
  • 项目记忆（project_memory）
  • 策略来源（agent_profile）。

而 executor 是代码执行的沙箱子进程，它需要接收工具可调用对象才能在隔离环境里执行它们。这些依赖来自不同的来源，其中工具来自 ToolRegistry，模型通道来自 LLMConfig，策略来自 AgentProfile，系统提示词要按工作区和模型能力现场拼装。它们之间还存在关联，Schema 要进提示词和 provider 请求，可调用对象要捕获本次运行的 ToolContext，权限策略要从 Profile 推导。来源分散加上关联复杂，需要一个统一入口按固定顺序完成组装，这就是 build_loop_runtime_bundle。产物依赖本次运行的工作区、模型配置和用户消息，因此每次 run 都重新装配，上一轮的产物不复用。

装配的输入是 ToolRegistry、LLMConfig、ToolContext 和 AgentProfile，输出是 AgentLoop 与子进程 executor。本节涉及模块：AgentProfile（governance/profile.py）、AgentFactory（agent_factory.py + factory_common.py）、PermissionRuntime（governance/permission.py）。

AgentProfile：Agent 的身份

governance/profile.py 里 AgentProfile 是一个不可变 dataclass，用一组字段声明 Agent 的全部身份：

1
2
3
4
5
6
7
8
9
10
@dataclass(frozen=True)
class AgentProfile:
    name: str                    # 身份名
    mode: AgentMode              # build / plan / explore / workflow / subagent
    description: str
    tool_groups: list[str] | None    # 可见工具的注册分组
    permission_level: PermissionLevel  # read_only / safe_write / full_access
    max_steps: int | None
    prompt_suffix: str               # 追加到系统提示词的策略性说明
    metadata: dict                   # 预算与权限覆盖参数

身份决定行为，我们内置了 5 个 Profile，以决定agent不同行为放行的权限以及暴露的工具：

身份 mode tool_groups 权限 预算 metadata
gis-build build core, qgis, osm, datasource, worker safe_write 无限制
gis-plan plan core read_only 3 provider turns, 1 code step
gis-explore explore core, datasource, osm read_only 4 provider turns, 2 code steps
workflow-runner workflow 全部 safe_write 8 provider turns
gis-subagent subagent 调用方指定 safe_write 4 provider turns

五个 Profile 的用途和选型规则：

  • gis-build：默认执行 Agent。没有特殊上下文时，用户每条消息都由它接手。不限步数，SAFE_WRITE，可以执行读写文件、运行代码、操作地图等一系列操作。
  • gis-plan：只读规划 Agent，用于任务分解和方案设计，只读不执行。
  • gis-explore：数据集探索 Agent，工具组限定 core/datasource/osm，只读，4 轮预算。用于快速摸清一个数据集的字段、范围以及结构。
  • workflow-runner：工作流节点执行 Agent。消息带 workflow 附件时由 OpenGISAgent.run() 自动选择，每个节点一个 session，8 轮预算覆盖多步 DAG。
  • gis-subagent：子任务 Agent。由 run_subagent / run_subagents 工具派生，工具组由调用方指定，4 轮预算，隔离上下文处理自包含的子任务。

工作区可以覆盖内置 Profile。AgentProfileStore 读取工作区下的 .opengis/agents.json，from_dict 反序列化后按 name 覆盖默认集。同一份 OpenGIS，不同 workspace 可以有不同的 Agent 行为。

从身份推导权限

PermissionRuntime.from_profile 把 Profile 翻译成 PermissionPolicy：一张默认动作 + 工具覆盖 + 通配规则的查表。三个权限级别各自成文：

权限级别 默认 典型覆盖
read_only allow execute_code/bash/add_layer 需审批；write/edit/delete/move/copy 文件拒绝；remove_layer 拒绝
safe_write allow delete_file 需审批；其余写操作放行
full_access allow 全部放行

evaluate(tool_name, arguments) 的判定顺序是：强制审批工具 → 持久化规则（工作区 permission store）→ Profile 规则 → 参数风险检查 → 审批回调。

工具可见性过滤

Agent 能看到哪些工具取决于工具可见性过滤，agent/execution/tool_packs.py 维护一张工具名到 pack 的映射，把 80 个内置工具归入十余个能力包：

Pack 代表工具
skill load_skill, update_user_instructions
file read_file, write_file, edit_file, glob, grep
code list_scripts, read_script
system bash
map add_layer, remove_layer, fly_to, zoom_to_layer
style set_categorized_style, set_graduated_style, set_layer_label
raster add_raster, get_raster_info, set_raster_style
map_3d enter_3d_view, set_map_camera, set_extrusion_style
operation run_operation, edit_operation, promote_script_to_operation
worker start_worker, start_dynamic_map_worker, restart_worker
workflow create_workflow
subagent run_subagent, run_subagents
osm / datasource / qgis / web osm_call, datasource_call, qgis_call, webfetch, websearch
debug debug_agent_context

Agent 默认只暴露四个 pack：skill, file, code, map。其余 pack 在运行时按用户消息推断追加（infer_tool_packs_for_text），这是 RuntimeControl 的职责，后面第四部分展开叙述。这里只需记住一个事实：Agent 的工具面是分层可见的。execute_code 和 run_script_file 永远可见，不参与 pack 过滤。

装配线

build_loop_runtime_bundle（factory_common.py）是装配的主体，按固定顺序组装：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# factory_common.py 装配顺序（伪代码）
registered = tools.list_registered()
registered = [s for s in registered if s.schema.group in effective_groups]  # 身份过滤
registered = filter_agent_tools(registered, allow_3d=flag)                 # 平台过滤

tool_callables = build_tool_callables(registered, ctx_provider=lambda: ctx)
tool_schemas   = build_tool_schemas(registered)
llm_call       = build_llm_caller(llm_config)

executor   = build_subprocess_executor(ctx)
permission = PermissionRuntime.from_profile(profile, workspace_path=...)

tool_runtime = ToolRuntime(
    tool_schemas=tool_schemas,
    tool_callables=tool_callables,
    executor_call=executor_call,
    permission_runtime=permission,
    ...
)

system_prompt  = compose_system_prompt(registered, ctx)   # 稳定前缀
project_memory = project_run_memory(ctx)                  # 动态尾部

四个组件的构造如下：

工具可调用对象。 build_tool_callables 把 RegisteredTool 转成 name → callable 的字典，声明 needs_context 的工具自动注入 ToolContext。

工具 Schema。 build_tool_schemas 把 ToolSchema（含 ToolParam 列表）转成 JSON Schema，最终以 function-calling 格式交给 provider。

LLM caller。 build_llm_caller（agent/llm.py）把 LLMConfig（protocol/model/api_key/base_url）变成可调用对象，返回统一的 LLMResponse。

系统提示词。 compose_system_prompt 组装稳定前缀（模板、工具目录、能力清单、workspace 路径），项目记忆单独生成，运行时注入动态尾部。

build_agent_loop：实例化 AgentLoop

装配的最后一环在 agent_factory.build_agent_loop，其将 runtime bundle 的产物实例化成 AgentLoop，伪代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# agent_factory.py（伪代码）
runtime = build_loop_runtime_bundle(...)
system_prompt = runtime.system_prompt
if profile.prompt_suffix:
    system_prompt += "\n" + profile.prompt_suffix.strip() + "\n"

agent_loop = AgentLoop(
    llm_call=runtime.llm_call,
    executor_call=runtime.executor_call,
    system_prompt=system_prompt,
    tool_runtime=runtime.tool_runtime,
    tool_schemas=runtime.tool_schemas,
    tool_materializer=ToolMaterializer(runtime.tool_schemas),
    context=context or ContextManager(),
    project_memory=runtime.project_memory,
    agent_profile=profile,
)
return agent_loop, runtime.executor

函数返回 AgentLoop 和子进程 executor ，其中，executor 持有子进程句柄，而生命周期仍然属于调用方，run 结束后由 OpenGISAgent.run() 的 _cleanup 调用 executor.cleanup()，并完成 git 快照、归档和知识抽取。到这一步后，Agent 的身份、权限、工具面、LLM 通道、上下文容器即全部就绪。

Loop step1: 构造

构造是四步循环的第一步，发生在 LoopKernel.run_turn() 的开头。它把上下文、工具 Schema、系统提示词拼成一条 provider 请求。

本节涉及模块：LoopKernel（loop/loop_kernel.py）、ContextManager（context/context_manager.py）、ToolMaterializer（execution/tool_materializer.py）、RequestBudgetManager（context/request_budget.py）、ProviderRequestBuilder（context/provider_request.py）。

压缩检查

上下文压缩是每一个 Agent 都需要做的工作，因此，run_turn 的第一步必须是由 ContextManager 评估上下文是否还有空间：

1
2
3
4
# loop_kernel.py
should_compress, reason = self.context.should_compress()
if should_compress:
    self.context.compress(self.llm_call)

should_compress（context_manager.py 第 450 行）有两种触发条件：累计消息的 token 估算超过阈值（token_budget * compress_threshold），或者工具结果占 live context 的比例超过 60%。后者单独成条，因为 GIS 工具的输出经常是大段 JSON，从而导致长任务中工具结果淹没对话。

compress 把旧消息压成一段摘要，保留最近 keep_recent 条不动。摘要优先用 LLM 生成（llm_summarize），并且是 anchored merge，即上一次的摘要作为 <previous-summary> 传入，新摘要已包含旧摘要，防止多以次压缩后摘要无限膨胀。LLM 摘要失败时回退到简单拼接，但此时必须把旧摘要追加进去，否则会静默丢失前几轮压缩累积的历史。

除此之外还有一层面向旧工具结果的裁剪，我们使用_prune_outputs 把旧的工具结果替换成骨骼占位符（工具名、调用 id、时间戳保留，正文删除），以压缩上下文。

工具物化：Schema 的选择

第二步是要决定本轮模型看到哪些工具。ToolMaterializer.materialize 负责持有装配阶段生成的完整 schema 集，并按包过滤：

1
2
3
4
5
6
7
8
9
10
# tool_materializer.py
def materialize(self, force_all=False, selected_packs=None) -> ToolMaterialization:
    if force_all:
        return ToolMaterialization(schemas=self.schemas, reason="all")
    selected = self._dedupe(self.schemas)
    if selected_packs:
        selected = [s for s in selected
                    if tool_pack_for_name(name(s)) in selected_packs
                    or name(s) in ALWAYS_AVAILABLE_SCHEMA_NAMES]
    return ToolMaterialization(schemas=selected, reason="packs:...")

selected_packs 来自 RuntimeControl 的任务模式推断，但 execute_code 和 run_script_file 始终可见，不参与包过滤。额外的，使用force_all 作为兜底，即模型在回复里说找不到某个工具时，系统判断这可能是工具过滤造成的困惑，下一轮会把全部工具重新暴露一遍。

过滤后的 schema 列表进入 provider 请求的 tools 参数，按包归组的工具摘要（format_active_tool_prompt）写进系统提示词的稳定前缀部分，模型靠前者拿到精确的函数签名，靠后者知道平台具备哪些能力包。

组装：build_provider_request

每次构造请求，系统提示词、历史、工具摘要、动态尾部这些内容，都要按固定顺序转成 provider 能接受的消息数组。这个转换只有一个入口ContextManager.build_provider_request ，它用 ProviderRequestBuilder 按固定布局把内容文本化成消息：

1
2
3
4
5
6
7
8
9
10
11
── STABLE PREFIX（cache_policy=cacheable）──
[S0] system core prompt        ← 装配阶段的 compose_system_prompt 产物，含 Profile.prompt_suffix
[S1] stable system sections    ← active tools 包摘要、能力清单
[S2] user preferences          ← 用户偏好指令（截断到 2000 字符）
[S3] conversation summary      ← 历史压缩的摘要
── conversation history（append-only）──
[H]  投影后的消息历史          ← 裁剪、占位符替换后的历史
── DYNAMIC TAIL（cache_policy=none）──
[D*] turn objective / 纠偏消息  ← RuntimeControl 每轮生成
[D1] runtime anchor
[D2] working state

每个 section 最终变成一条 {"role": "system"|"user"|"assistant"|"tool", "content": ...} 消息，ProviderRequestBuilder.build 把它们顺序拼接成 ProviderRequest.messages。同时，history 段经过 ProviderContextProjector 投影，按 BudgetLimits 裁剪内存记录、工具结果长度、代码字符数，超出部分替换为占位符。每个每轮变化的内容必须位于历史之后，否则从请求头部开始的缓存前缀就被打断。

预算反馈环

未来规避上下文过大的问题，RequestBudgetManager 会对拼好的请求做 token 估算和压力分级，压力大时裁剪后重装：

1
2
3
4
5
6
7
8
9
# loop_kernel.py（简化）
provider_request, messages = _assemble(budget_limits)      # 第一次组装
budget_report = analyze(messages, tools=active_tool_schemas)

if budget_report.pressure in {"hot", "overflow"}:
    saved = self.context.prune_tool_results()              # 裁旧工具结果
    budget_limits = budget_manager.suggest_limits(pressure=...)
    provider_request, messages = _assemble(budget_limits)  # 重新组装
    budget_report = analyze(...)                            # 再估算

预算模型（request_budget.py）：input_token_budget 默认 100k，减去 4096 的输出保留，得到可用输入预算，从而将 token 压力按已用比例分四档：ok / warm（0.55）/ hot（0.75）/ overflow（0.92），每档对应一组 BudgetLimits，控制下一轮组装时能放多少内存记录、多少字符的工具结果、多少字符的执行代码。

协议消毒与缓存计划

请求发出前还有两道处理。

_ensure_provider_tool_protocol（loop_kernel.py 第 389 行）清洗历史消息，保证满足 provider 的 tool-call 协议。同时，在开发过程中，观测到长对话里可能出现两类脏数据：孤立的 tool result（对应的 assistant tool_calls 已被压缩掉）和不完整的 assistant tool_calls（缺 call id）。直接把这类消息发给 OpenAI 兼容接口会报协议错误。对此，清洗策略是把它们替换成 system 消息，说明"历史工具调用事务已摘要"，保证消息对完整。

为了观测发送内容的稳定性，ProviderRequestAdapter.prompt_cache_plan 计算三个哈希：cacheable_prefix_hash（顶部连续 cacheable 段）、system_prefix_hash（历史之前的全部稳定段）、dynamic_suffix_hash（历史之后的尾部）。这些哈希连同缓存元数据传给 ProviderTurnCaller，最终出现在 provider 请求的缓存参数里，从而展示在设置页面。

到这一步即构造完成了，产物是 ProviderRequest（messages + sections 元数据）加 active_tool_schemas，两者就是这一轮发送的 token 的完整构成。

Loop step2: 发送

四步循环的第二步是将内容发送给供应商。构造产出的 ProviderRequest 在这里交给 ProviderTurnCaller 发出。

本节涉及模块：ProviderTurnCaller（loop/turn_runner.py）、build_llm_caller（agent/llm.py）。

ProviderTurnCaller 的调用

ProviderTurnCaller.call 是构造与 provider 之间的边界。它拿到 messages 和 tools，调用装配阶段生成的 llm_call：

1
2
3
4
5
6
7
8
9
# turn_runner.py 第 475 行
response = self.llm_call(
    messages,
    on_delta=_on_llm_delta,
    on_tool_delta=_on_tool_delta,
    tools=active_tool_schemas,
    prompt_cache_key=prompt_cache_key,
    prompt_cache_metadata=prompt_cache_metadata,
)

llm_call 由 build_llm_caller（llm.py 第 468 行）生成，内部是 litellm.completion。provider 前缀（openai/、anthropic/、deepseek/）决定走哪个协议的线格式。这一部分非常简单，并没有什么技术细节可以深究。

Loop step3: 解析

四步循环的第三步就是解析 LLM 供应商的返回内容。响应在这里被拆成 response_text 和 tool_calls 两部分，循环据此决定继续还是终止。

本节涉及模块：llm.py 的响应归一化层、ProviderTurnCaller（loop/turn_runner.py）、AgentLoop（loop/agent_loop.py）。

响应归一化

OpenGIS 使用 litellm 统一调用各种 provider，但仍有可能遇到不同 provider 返回的响应格式不一致的问题，为了规范化输出，我们规定llm_call 返回 LLMResponse 必须包含下面三个字段：content、tool_calls、finish_reason（“stop” | “tool_calls” | “length”）。

如果是规范化的情况，即类似于 OpenAI 风格的接口把工具调用放在标准字段里的模式，_extract_tool_calls（第 178 行）从 litellm 的 message 对象提取标准工具调用，遍历 message.tool_calls，取 id 和 function 的 name、arguments，组装成 OpenAI 格式的 dict 列表。在非标准情况下，_extract_xmlish_tool_calls 用正则从响应文本里识别工具调用标签，这里不多做赘述。此外，还有一种可能，即模型回复纯文本，这里不需要特殊处理。

ProviderTurnCaller 拿到 LLMResponse 后做最后一步拆解（第 517-518 行）：

1
2
response_text = response.content or ""
tool_calls = response.tool_calls

至此，一条响应被拆成两部分：给用户看的文本，和给系统执行的工具调用列表。

分流：AgentLoop 的决策

拆出两部分之后，LoopKernel.run_turn 返回 LoopTurnOutcome，AgentLoop.run() 据此决定下一步：

1
2
3
4
5
6
7
8
# agent_loop.py（简化）
if tool_calls:
    settle_decision = policy.after_settlements(...)
    control_decision = runtime_control.observe_settlements(...)   # 第七部分
    continue    # 工具结果已回流，回到构造，开始下一轮

# 无工具调用：纯文本即最终答案
return response_text

还有一条自愈路径。如果模型回复了纯文本，但内容疑似"工具不可见"的困惑（is_tool_visibility_miss 正则匹配"没有/不存在/无法…工具"这类表述），循环会把它当作一次物化失误，从而把这条回复记入历史，注入一条系统消息说明"前一条回复可能混淆了动态工具可见性与平台能力，用完整工具集重试一次"，然后 force_all_tools_once = True，下一轮构造时全量物化。平台有而模型看不见的能力，不会以"这个工具不存在"告终。

Loop step4: 处理

四步循环的第四步为处理，工具调用在这里被执行并结算，结果回流为下一轮构造的输入。

本节涉及模块：ToolCallSettler（loop/turn_runner.py）、ToolRuntime（execution/tool_runtime.py）。

结算：ToolCallSettler

ToolCallSettler.settle_all（第 573 行）逐个处理工具调用。两个来源会分流：

1
2
3
4
5
6
7
# turn_runner.py 第 583 行起
for tool_index, tc in enumerate(tool_calls):
    tc_id = str(tc.get("id", ""))
    if tc_id in blocked_call_ids:          # 被 RuntimeControl guard 拦截
        settlements.append(self._settle_blocked(tool_call=tc, reason=...))
        continue
    settlements.append(self._settle_one(tool_index=tool_index, tool_call=tc, ...))

_settle_one（第 645 行）处理真正要执行的工具调用（被 guard 拦下的走 _settle_blocked，不执行工具），其内部是固定步骤的流水线，无论哪个工具都按同一顺序运行，其步骤包括：

  1. 解析参数：parse_tool_arguments 把 function.arguments 的 JSON 字符串变成 dict
  2. 进度提示：tool_intent_progress 把工具名映射成用户可见的进度文案（“加载要素图层”、“设置分类符号”），不暴露链式推理
  3. on_tool_start 回调：通知前端工具开始执行
  4. tool_runtime.execute(tool_name, arguments)：进入 ToolRuntime 执行流水线，见本部分第 2 节
  5. 结果处理：execute_code 成功后，如果脚本已持久化，在结果末尾追加一行提示，告诉模型"这个脚本已存盘，后续修复请用 edit_file 改原文件再 run_script_file，不要另写近似副本"
  6. context.add_tool_result：把工具结果写回历史，成为下一轮构造的输入
  7. context.prune_tool_results：结算完立即裁剪旧结果，避免上下文被旧输出撑大
  8. 产出 ToolSettlement：call_id、name、arguments、content、error、duration_ms、metadata、counts_as_code_step

_settle_blocked（第 602 行）处理被 guard 拦下的调用。它不执行工具，而是构造一个 {"success": false, "error": "runner_guard_blocked", "reason": ...} 的结果，同样写回 context。模型能够看到"这个调用被系统拦了，原因是 X"，从而修正自己的行为。

ToolRuntime 执行流水线

结算的第 4 步把工具调用交给 ToolRuntime.execute（execution/tool_runtime.py），这是工具执行的总入口。它内部是一条四段流水线：校验参数、评估权限、分派执行、处理输出。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# tool_runtime.py（简化）
def execute(self, tool_name, arguments) -> ToolExecutionResult:
    # ① 特判与参数校验
    if tool_name == "execute_code":
        error = validate_execute_code_payload(arguments.get("code", ""))
        if error: return error_result("invalid_execute_code_payload", ...)
    else:
        prepared = self.argument_contract.prepare(tool_name, arguments)
        if not prepared.ok: return error_result("invalid_tool_arguments", ...)

    # ② 权限评估
    decision = self.permission_runtime.evaluate(tool_name, args)
    if decision.action != ALLOW and policy.enforce:
        return error_result("permission_required"/"permission_denied", ...)

    # ③ 分派执行
    if tool_name == "execute_code":
        content = self.python_runtime.execute_code(args)       # 子进程
    elif tool_name == "run_script_file":
        return self._execute_script_file(args, ...)            # 加载脚本再执行
    elif tool_name in CODE_ONLY_TOOLS:
        content = self.python_runtime.execute_code_only_tool(...)
    else:
        content = self._execute_tool(tool_name, args)          # 普通函数调用

    # ④ 输出处理
    bounded = self._bound_output(tool_name, content)           # 截断 + artifact 提取
    return ToolExecutionResult(...)

① 校验。 execute_code 的代码先过 validate_execute_code_payload，必须是不带 markdown 围栏、不带 <think> 标签、没有长篇叙述性注释的纯 Python。校验失败则直接返回 invalid_execute_code_payload。其余工具走 ToolArgumentContract.prepare，按 Schema 检查参数类型、必填项和枚举值，失败时返回 invalid_tool_arguments，并附上该工具实际接受的参数清单，可以用于模型下一轮次的参考

② 权限。 PermissionRuntime.evaluate 返回 ALLOW / ASK / DENY。ASK 且启用了强制审批时，将在前端 Chat 部分弹出对话框让用户进行审批；DENY 直接返回 permission_denied。run_script_file 在加载脚本后还要对脚本代码再评估一次 execute_code 权限。

③ 执行分派 ，包含四条路径：

  • execute_code 将调用 PythonExecutionRuntime.execute_code，这是最复杂的一条路径，它先做三道预检：代码为空直接报错；检测到动态地图类的死循环代码时提示改用 Worker（resident_worker_required，提示 start_dynamic_map_worker）；检测到 import opengis 这类保留导入时提示"直接调注册工具，别自己 import SDK"（虽然在设计上不合理，但确实可以规避一些 LLM 误判的问题）。预检通过后，扫描代码里的 import，缺的包自动装（auto_install_for_code），然后交给子进程执行。执行结果若报 ModuleNotFoundError，装包后自动重试一次。返回内容把安装记录、日志、输出、错误拼在一起。
  • run_script_file 先按路径加载脚本，过权限，再走同一套子进程执行。
  • CODE_ONLY_TOOLS 目前只有 save_plot，它是子进程里的辅助函数，用于制图与展示，运行时代理成一次 execute_code("save_plot(...)")。
  • 其余工具在 _execute_tool 里查 callables 表调用。工具函数若返回协程（部分工具是 async 的），用独立的 event loop 驱动，不阻塞 AgentLoop 的线程。

④ 输出处理。 ToolOutputRuntime.bound 对结果做两件事：超过长度上限的内容截断，附 truncated 标记；从输出里提取 artifact 提示（生成的文件路径、图表路径），写进 metadata，前端据此展示产物。最终返回 ToolExecutionResult：content、error、duration_ms、metadata、truncated。

到这里，一次工具调用从参数到结果的处理闭环完成。结果回到 _settle_one 写进 context，成为下一轮构造的输入。但循环是否继续，还取决于结算后的治理层，包括预算够不够、行为是否偏离、是否该收束等问题，在下一章我们将详细阐述设计。

Loop的边界：治理与收敛

治理层不在四步循环内部，它在循环边界上做决策，AgentLoop.run() 的主循环有两个决策点：

  • 本轮开始前（预算检查）
  • 结算后（观察纠偏）。
    本节涉及模块：LoopPolicy（loop/policy.py）、RuntimeControl（loop/runtime_control.py）、auto_final（loop/auto_final.py）、RunnerState（loop/runner_state.py）。

LoopPolicy：预算硬约束

LoopPolicy 从 AgentProfile.metadata 读取四个预算参数，在每轮开始时检查：

参数 含义
max_provider_turns 最多多少次 provider 调用
max_code_steps 最多多少次代码执行
max_tool_steps 最多多少次非代码工具调用
max_work_steps 代码与非代码合计上限

before_provider_turn 逐项比对当前计数，任一超限就返回 force_final,触发后，下一轮 provider turn 的工具被禁用，并注入一条 final_turn_instruction：“工具已禁用，基于已有结算结果给出最终回答，不要开始新方案”。

RuntimeControl：三阶段软治理

每一次 run 都将调用一次 RuntimeControl ，状态在轮与轮之间累积：工具历史（tool_history）、近期失败（recent_failures）、偏离计数（deviation_count）。它在一个 turn 的前、中、后三个时点介入，下面是详细的介绍：

事前：任务模式推断。 infer_task_mode 从用户消息的关键词推断模式，共包含7 种：

模式 触发关键词示例 约束要点
OPERATION_REPAIR 修、改、失败、fix、repair 先修 Operation 再重跑，禁止绕过
OPERATION_RUN operation、操作 用 Operation 生命周期工具
MAP_RENDERING 图层、地图、样式、颜色 用当前地图状态，避免无关工作
DATA_ANALYSIS 分析、统计、评价、analy 证据够了就答，不额外造文件
WORKER worker、后台、驻守 用 Worker 生命周期工具，不用一次性脚本
WORKFLOW workflow、工作流 先 workflow 工具再考虑副作用
GENERAL 默认 无额外约束

模式决定两件事情，一是注入 TurnObjective 的约束文本（模型每轮都能看到"当前任务模式是 X，规则是 Y"），二是补充工具 pack（比如 OPERATION_REPAIR 强制带上 operation pack）。

事中：调用拦截。 模型发出工具调用后、执行之前，guard_tool_calls 逐个检查。典型拦截规则如下：

  • OPERATION_REPAIR 模式且 run_operation 刚失败、还没编辑：拦截 execute_code、run_script_file、bash 以及所有地图副作用工具，理由写清"先修现有 Operation，别用一次性脚本绕过"
  • WORKER 模式：拦截一次性执行工具（execute_code / run_script_file / bash），要求用 start_worker 这类生命周期工具
  • DATA_ANALYSIS 模式且用户没提可视化：拦截地图工具，理由"当前请求是分析，不是渲染"

拦截后通过调用 _settle_blocked，构造一个带 runner_guard_blocked 错误和拦截原因的结果，从而写回 context，在下一轮次中展示给 LLM 。

事后：观察纠偏。 observe_settlements 会在每批结算后扫描，产出一个 ControlDecision，从设计层面，其可能是一条纠偏消息，也可能直接要求收束。核心逻辑如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
# runtime_control.py（简化）
for settlement in settlements:
    if run_operation 失败 and 模式是 OPERATION_REPAIR:
        correction = "run_operation 失败。先 get_operation(include_code=true) 检查契约，
                      再用 edit_operation/edit_file 修复，最后 rerun。禁止用 execute_code 绕过。"
    if 编辑类工具成功:          # 修复动作发生
        recent_failures.clear()  # 旧失败不再是"原样重试"的证据
    if 重复失败签名:            # 同工具同错误形状连续两次
        correction = "不要原样重试，改输入、改契约或说明阻塞原因。"
    if Worker 启动且健康:
        return ControlDecision(force_final_reason="worker_running_verified")
    if DATA_ANALYSIS 且连续 2+ 次成功代码且输出充分:
        return ControlDecision(force_final_reason="analysis_answer_ready")

"输出充分"的判断本身是个关键词启发式：结论标记（综合评价、评级、结论、score、rating）至少命中 2 个，且证据标记（道路、公交、距离、覆盖、count、mean）至少命中 2 个，且文本超过 160 字。这防止模型分析到一半就开始自主探索。

收束机制

循环有五种方式结束，除了自然收敛（模型不调工具直接回文本），还有四条主动收束路径：

auto_final：跳过一轮 LLM 往返。 17 个白名单工具（全部是地图和样式类副作用工具：fly_to、zoom_to_layer、set_graduated_style、set_layer_label 等）在一批结算里全部成功时，本地直接生成确认文本返回，这样的设计模式可以降低 LLM 调用次数。

LoopAnomalyDetector：绕圈检测。 两种模式会被识别：map_churn（最近 8 步里 remove_layer 和 add_layer 各出现至少 2 次，典型的"改了又撤"）和 repeated_failure（同工具同错误形状连续两次）。命中后注入纠偏消息，提示"检查这还在不在服务当前目标"。

force_final：强制收束。 包含三个来源：预算超限（LoopPolicy）、Worker 就绪（worker_running_verified）、分析就绪（analysis_answer_ready）。触发后下一轮禁工具，要求基于已有结果给最终答案。

中断：协作式停止。 即用户强制中断，则将在外部调 agent_loop.interrupt() 置标志，循环顶部检查后返回 "(Task interrupted by user.)"的信息提示。

RunnerState：终止状态的记录

runner_state.py 把循环的可变状态（迭代计数、步数、force_final 原因、终止类型）收进一个对象，AgentLoop 和 WorkflowLoop 共用，避免两套状态机漂移。终止类型 RunnerTerminationKind 有七种：running、completed、interrupted、forced_final、failed、budget_exceeded、halted。run 结束时终止类型写进 RunArchive，前端据此展示"这次 run 是怎么结束的"。

示例：一次 run 的完整时序

图7：一次 run 的完整时序