综合调研

Archipelago 单任务执行全流程深度解析：从 Docker 启动到 LLM Tool Calling 的每一行代码

基于 Mercor/Archipelago 源码，逐层拆解一个 Agent 任务从环境启动、MCP 工具注册、工具发现、LLM 调用循环到评分的完整执行链路。每个环节引用具体代码片段，让你既知道'做了什么'，也知道'怎么做的'。

2026年4月22日 Agent MCP Archipelago Tool Calling Benchmark 源码解析

一、问题：Agent 到底是怎么跑起来的

Archipelago 是 Mercor 开源的 Agent 评测框架（对应论文 APEX-Agents, arXiv:2601.14242），用于在 Docker 沙箱环境中运行 AI Agent 并评分。

我在读 examples/simple_task 时有一个核心困惑：给到模型的输入到底长什么样？模型怎么知道自己能调哪些工具？ 这篇文档就是围绕这个问题，从源码层面逐行拆解整个执行链路。

二、全局架构：三组件 × 五阶段

三个解耦组件

┌──────────────────┐    HTTP API     ┌──────────────────┐
│  Environment     │◄───────────────►│  Agent Runner    │
│  (Docker 容器)   │   /mcp/ /apps   │  (Python + uv)   │
│  /data/populate  │   /data/snapshot│                  │
└──────────────────┘                 └──────────────────┘
                                            │
                                     快照 Diff + 轨迹
                                            ▼
                                     ┌──────────────────┐
                                     │  Grading         │
                                     │  (LLM-as-Judge)  │
                                     └──────────────────┘

五个执行阶段

阶段	做什么	代码入口
① 环境启动	Docker compose up + 健康检查	`examples/simple_task/main.py` → `start_environment()`
② 快照加载	zip → tar.gz → POST /data/populate	`main.py` → `zip_to_tar_gz()` + `requests.post()`
③ MCP 注册	POST /apps 注入 MCP Server 配置	`main.py` → `requests.post(ENV_URL + "/apps")`
④ Agent 运行	LLM ↔ MCP 工具调用循环	`runner/main.py` → `loop_agent/main.py`
⑤ 评分	前后快照 Diff + LLM Judge	`grading/runner/main.py`

三、阶段 ①②③：环境准备（main.py 编排脚本）

examples/simple_task/main.py 是端到端编排入口，核心流程：

3.1 环境启动

def start_environment():
    # 强制清除旧容器，保证干净状态
    subprocess.run(["docker", "compose", "down", "-v"], cwd=ENVIRONMENT_DIR)
    # 重新构建并启动
    subprocess.run(["docker", "compose", "up", "-d", "--build"], cwd=ENVIRONMENT_DIR)
    # 等待健康检查通过（轮询 /health，最长 120 秒）
    wait_for_health(ENV_URL)

每次运行都 down -v 再 up --build，保证环境幂等。

3.2 快照加载

# 将 original_snapshot.zip 转换为 tar.gz（去除 "filesystem/" 前缀）
world_tar_gz = zip_to_tar_gz(EXAMPLE_DIR / "original_snapshot.zip")

# 上传到环境容器
requests.post(
    f"{ENV_URL}/data/populate",
    files={"archive": ("world.tar.gz", f, "application/gzip")},
    params={"subsystem": "filesystem"},
)

subsystem=filesystem 告诉环境容器把数据加载到 /filesystem 目录。对于 simple_task 示例，加载后容器内的文件系统如下：

/filesystem/animals/
  xk92m/qz7fw.png  (gorilla - 目标)
  ab47z/rt3ky.png  (cat)
  mn83p/jw9vb.png  (elephant)
  jh21q/pl4nc.png  (penguin)

3.3 MCP 服务器注册

with open(EXAMPLE_DIR / "mcp_config.json") as f:
    mcp_config = json.load(f)

resp = requests.post(f"{ENV_URL}/apps", json=mcp_config)

mcp_config.json 的内容：

{
  "mcpServers": {
    "filesystem-server": {
      "transport": "stdio",
      "command": "python",
      "args": ["main.py"],
      "cwd": "./mcp_servers/filesystem_server",
      "env": {"APP_FS_ROOT": "/filesystem"}
    }
  }
}

POST /apps 做了什么：环境容器内部启动 filesystem-server 作为子进程（stdio 传输），并将其注册到 MCP Gateway。此后所有通过 /mcp/ 端点的请求都会被路由到这个 server。

关键约束：transport 必须是 "stdio"；cwd 必须是容器内有效路径。

四、阶段 ④：Agent 运行——核心链路

这是最关键的部分。从编排脚本发出的命令：

uv run python -m runner.main \
  --trajectory-id "task_001" \
  --initial-messages ./initial_messages.json \
  --mcp-gateway-url "http://localhost:8080/mcp/" \
  --agent-config ./agent_config.json \
  --orchestrator-model "google/gemini-2.5-flash" \
  --output ./trajectory.json

4.1 入口：`runner/main.py`

async def main(
    trajectory_id: str,
    initial_messages: list[dict[str, Any]],
    mcp_gateway_url: str | None,
    mcp_gateway_auth_token: str | None,
    agent_config: AgentConfig,
    orchestrator_model: str,
    orchestrator_extra_args: dict[str, Any] | None,
    ...
) -> AgentTrajectoryOutput:

    # 1. 从注册表获取 agent 实现
    agent_impl = get_agent_impl(agent_config.agent_config_id)

    # 2. 构造运行输入
    run_input = AgentRunInput(
        trajectory_id=trajectory_id,
        initial_messages=cast(list[LitellmInputMessage], initial_messages),
        mcp_gateway_url=mcp_gateway_url,
        mcp_gateway_auth_token=mcp_gateway_auth_token,
        orchestrator_model=orchestrator_model,
        orchestrator_extra_args=orchestrator_extra_args,
        agent_config_values=agent_config.agent_config_values,
    )

    # 3. 带超时保护地执行 agent
    async with asyncio.timeout(settings.AGENT_TIMEOUT_SECONDS):
        output = await agent_impl(run_input)

get_agent_impl 如何工作（registry.py）：

from runner.agents.loop_agent.main import run as loop_agent_run

AGENT_REGISTRY: dict[AgentConfigIds, AgentDefn] = {
    AgentConfigIds.LOOP_AGENT: AgentDefn(
        agent_config_id=AgentConfigIds.LOOP_AGENT,
        agent_impl=loop_agent_run,
        agent_config_fields=[
            TaskFieldSchema(field_id="timeout", default_value=10800),
            TaskFieldSchema(field_id="max_steps", default_value=100),
            TaskFieldSchema(field_id="tool_call_timeout", default_value=60),
            TaskFieldSchema(field_id="llm_response_timeout", default_value=600),
        ],
    ),
}

注册表模式——通过 agent_config_id 字符串查找对应的异步函数。loop_agent 的 run() 就是整个 Agent 循环的入口。

4.2 LoopAgent 初始化

class LoopAgent:
    def __init__(self, run_input: AgentRunInput):
        self.model = run_input.orchestrator_model
        self.messages = list(run_input.initial_messages)

        # 构建 MCP 客户端
        self.mcp_client = FastMCPClient(
            build_mcp_gateway_schema(
                run_input.mcp_gateway_url,
                run_input.mcp_gateway_auth_token,
            )
        )

build_mcp_gateway_schema（utils/mcp.py）做了什么：

def build_mcp_gateway_schema(
    mcp_gateway_url: str,
    mcp_gateway_auth_token: str | None,
) -> dict:
    gateway_config = {
        "transport": "streamable-http",
        "url": mcp_gateway_url,
    }
    if mcp_gateway_auth_token:
        gateway_config["headers"] = {
            "Authorization": f"Bearer {mcp_gateway_auth_token}"
        }
    return {"mcpServers": {"gateway": gateway_config}}

注意这里 transport 是 "streamable-http"——Agent Runner 通过 HTTP 连接 MCP Gateway，而 Gateway 内部通过 stdio 连接各个 MCP Server。两层传输协议不同。

4.3 工具发现：从 MCP 到 OpenAI Tool 格式

这就是”模型怎么知道有哪些工具”的答案。

async def _initialize_tools(self) -> None:
    async with self.mcp_client as client:
        tools: list[ChatCompletionToolParam] = await load_mcp_tools(
            client.session, format="openai"
        )
    logger.info(f"Loaded {len(tools)} MCP tools")
    self.tools = tools

load_mcp_tools 是 LiteLLM 的实验性 API，它：

通过 MCP 协议调用 tools/list 方法
获取 MCP Server 暴露的所有工具定义
自动转换为 OpenAI function calling 格式（format="openai"）

转换前（MCP 原生格式）：

{
  "name": "read_file",
  "description": "Read contents of a file",
  "inputSchema": {
    "type": "object",
    "properties": {"path": {"type": "string"}},
    "required": ["path"]
  }
}

转换后（OpenAI Tool 格式，即传给 LLM 的）：

{
  "type": "function",
  "function": {
    "name": "read_file",
    "description": "Read contents of a file",
    "parameters": {
      "type": "object",
      "properties": {"path": {"type": "string"}},
      "required": ["path"]
    }
  }
}

差异很小——inputSchema 变成了 parameters，外面包了一层 type: "function" + function: ...。但这个转换是关键的：MCP 协议和 LLM function calling 协议是两个不同的标准，LiteLLM 在中间做了桥接。

4.4 核心循环：LLM 调用 → 工具执行 → 结果追加

async def run(self) -> AgentTrajectoryOutput:
    async with asyncio.timeout(self.timeout):  # 总超时 3 小时
        await self._initialize_tools()          # 加载工具

        for i in range(self.max_steps):         # 最多 100 步
            if self._finalized:
                break
            await self.step()

        if not self._finalized:
            self.status = AgentStatus.FAILED     # 100 步没完成 = 失败
        else:
            self.status = AgentStatus.COMPLETED

每一步 step() 的核心逻辑：

Step 4.4.1：调用 LLM

async def step(self):
    response = await generate_response(
        self.model,       # "google/gemini-2.5-flash"
        self.messages,    # 完整对话历史
        self.tools,       # OpenAI 格式的工具列表
        self.llm_response_timeout,
        self.extra_args,
    )

generate_response（utils/llm.py）的实现：

@with_retry(max_retries=10, base_backoff=5, jitter=5, retry_on=(
    RateLimitError, Timeout, BadRequestError, ServiceUnavailableError, ...
))
async def generate_response(model, messages, tools, llm_response_timeout, extra_args, ...):
    kwargs = {
        "model": model,
        "messages": messages,
        "timeout": llm_response_timeout,
        **extra_args,
    }
    if tools:
        kwargs["tools"] = tools    # 有工具时才传

    response = await acompletion(**kwargs)  # LiteLLM 的异步调用
    return ModelResponse.model_validate(response)

这就是传给模型的实际输入——messages（对话历史）+ tools（工具列表），通过 LiteLLM 的 acompletion 发给任意 LLM Provider。LiteLLM 会自动把 OpenAI 格式的 tools 转成各 Provider 的原生格式（Anthropic 的 tool_use、Gemini 的 function_declarations 等）。

重试策略：最多 10 次，指数退避 5 秒 + 5 秒抖动。上下文窗口溢出（ContextWindowExceededError）不重试。

Step 4.4.2：解析 LLM 响应

    response_message = LitellmOutputMessage.model_validate(choices[0].message)
    tool_calls = getattr(response_message, "tool_calls", None)

    # 兼容多种模型的思考过程
    if getattr(response_message, "reasoning_content", None):  # o1/o3
        logger.bind(message_type="reasoning").info(...)
    if getattr(response_message, "thinking_blocks", None):    # Claude
        for block in response_message.thinking_blocks:
            logger.bind(message_type="thinking").debug(...)

Step 4.4.3：执行工具调用

    if tool_calls:
        deferred_image_messages = []

        async with self.mcp_client as client:  # 每轮新建 MCP 会话
            for tool_call in tool_calls:
                try:
                    call_result = await asyncio.wait_for(
                        call_openai_tool(client.session, tool_call),
                        timeout=self.tool_call_timeout,  # 60 秒
                    )
                except TimeoutError:
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": "Tool call timed out",
                    })
                    continue

                # MCP 结果 → LiteLLM 消息格式
                messages = content_blocks_to_messages(
                    call_result.content,
                    tool_call.id,
                    tool_call.function.name,
                    self.model,
                    deferred_image_messages=deferred_image_messages,
                )
                self.messages.extend(messages)

        self.messages.extend(deferred_image_messages)  # 图片延迟追加

call_openai_tool 做了什么：把 OpenAI 格式的 tool_call（name + arguments）转成 MCP 的 tools/call 请求，发给 MCP Gateway，Gateway 路由到对应的 MCP Server 执行。

content_blocks_to_messages（utils/mcp.py）处理了一个重要的兼容性问题：

def content_blocks_to_messages(content_blocks, tool_call_id, name, model, ...):
    supports_image_tool_results = model.startswith("anthropic/")

    for content_block in content_blocks:
        match content_block:
            case TextContent():
                text_contents.append(block.text)
            case ImageContent():
                data_uri = f"data:{block.mimeType};base64,{block.data}"
                image_data_uris.append(data_uri)

    if supports_image_tool_results:
        # Anthropic: 图片直接嵌入 tool result
        content.append({"type": "image_url", "image_url": {"url": data_uri}})
    else:
        # 其他模型: 图片不能放在 tool result 里
        # 延迟追加为 user 角色的消息
        deferred_image_messages.append({
            "role": "user",
            "content": [{"type": "image_url", ...}],
        })

Anthropic 支持在 tool result 中嵌入图片，OpenAI 和 Gemini 不支持。所以非 Anthropic 模型的图片被单独提出来作为 user 消息追加。

Step 4.4.4：任务完成判定

    else:  # 没有 tool_calls
        self._finalized = True
        finalize_answer(response_message.content or "No content")

极简判定：LLM 不再请求任何工具 → 认为任务完成。

4.5 一个完整的 messages 演进示例

以 simple_task（找大猩猩图片）为例，messages 数组在循环过程中大致如此演进：

第 0 步（初始）：

[
  {"role": "user", "content": "Find the path to the gorilla image in the filesystem."}
]

第 1 步（LLM 决定列目录）：

[
  {"role": "user", "content": "Find the path to the gorilla image..."},
  {"role": "assistant", "tool_calls": [
    {"id": "tc_1", "function": {"name": "list_directory", "arguments": "{\"path\": \"/animals\"}"}}
  ]},
  {"role": "tool", "tool_call_id": "tc_1", "content": "[\"xk92m\", \"ab47z\", \"mn83p\", \"jh21q\"]"}
]

第 2-5 步（LLM 逐个子目录探索）：

[
  ... (前面的消息),
  {"role": "assistant", "tool_calls": [
    {"id": "tc_2", "function": {"name": "read_file", "arguments": "{\"path\": \"/animals/xk92m/qz7fw.png\"}"}}
  ]},
  {"role": "tool", "tool_call_id": "tc_2", "content": "(图片内容/描述)"}
]

最终步（LLM 给出答案，不再调用工具）：

[
  ... (所有历史消息),
  {"role": "assistant", "content": "The gorilla image is at /animals/xk92m/qz7fw.png"}
]

此时 tool_calls 为 None → self._finalized = True → 循环结束。

五、阶段 ⑤：评分

Agent 完成后，编排脚本：

从环境容器获取最终快照 POST /data/snapshot
调用 grading runner，输入：
- original_snapshot.zip（初始状态）
- final_snapshot.zip（最终状态）
- trajectory.json（Agent 行为轨迹）
- verifiers.json（评分标准）

评分系统使用 LLM-as-Judge 模式，根据 verifiers 定义的 criteria 评估 Agent 是否达成目标。

verifiers.json 示例：

[
  {
    "verifier_id": "ver_001",
    "verifier_values": {
      "criteria": "The agent found the gorilla image path",
      "is_primary_objective": true
    }
  }
]

最终输出 grades.json，包含分数和评判理由。

六、关键设计决策总结

6.1 三层超时保护

层级	位置	默认值	作用
L1 工具调用级	`asyncio.wait_for(call_openai_tool(...))`	60s	单个 MCP 工具卡死不影响整体
L2 LLM 响应级	`generate_response` 的 `timeout` 参数	600s	单次 LLM 调用超时
L3 总运行级	`asyncio.timeout(self.timeout)`	10800s (3h)	整个 Agent 生命周期

6.2 错误处理策略

错误类型	处理方式
LLM 空响应	自动追加 `"continue"` 消息让 LLM 继续
工具调用超时	记录 “Tool call timed out”，跳过继续
非致命 MCP 错误	记录错误信息，跳过该工具继续
致命 MCP 错误	直接 raise，终止 Agent
LLM API 限流/5xx	指数退避重试，最多 10 次
上下文窗口溢出	不重试，直接抛出

6.3 多模型兼容

差异点	处理方式
Tool 格式	LiteLLM 统一转换（OpenAI ↔ Anthropic ↔ Gemini）
图片 in tool result	Anthropic 直接嵌入；其他模型延迟追加为 user 消息
推理过程	`reasoning_content`（o1/o3）和 `thinking_blocks`（Claude）都记录

七、一张图看完全链路

examples/simple_task/main.py
│
├─① docker compose up → Environment 容器 (8080)
│
├─② POST /data/populate ← original_snapshot.zip → tar.gz
│
├─③ POST /apps ← mcp_config.json
│     └→ 容器内启动 filesystem-server (stdio)
│     └→ 注册到 MCP Gateway (/mcp/)
│
├─④ uv run python -m runner.main
│     │
│     ├─ runner/main.py
│     │   └─ get_agent_impl("loop_agent") → LoopAgent
│     │
│     ├─ LoopAgent.__init__()
│     │   └─ FastMCPClient(build_mcp_gateway_schema(url))
│     │       └→ {"mcpServers": {"gateway": {"transport": "streamable-http", "url": ".../mcp/"}}}
│     │
│     ├─ _initialize_tools()
│     │   └─ load_mcp_tools(session, format="openai")
│     │       └→ MCP tools/list → OpenAI ChatCompletionToolParam[]
│     │
│     ├─ for step in range(100):
│     │   │
│     │   ├─ generate_response(model, messages, tools, ...)
│     │   │   └─ litellm.acompletion(model=..., messages=..., tools=...)
│     │   │       └→ LLM 看到 messages + tools，决定调用哪个工具
│     │   │
│     │   ├─ if tool_calls:
│     │   │   └─ call_openai_tool(session, tool_call)
│     │   │       └→ MCP tools/call → filesystem-server 执行
│     │   │       └→ content_blocks_to_messages() → 追加到 messages
│     │   │
│     │   └─ else: _finalized = True (任务完成)
│     │
│     └─ 输出 trajectory.json
│
└─⑤ POST /data/snapshot → final_snapshot.zip
     └─ uv run python -m grading/runner/main.py
         └→ 对比快照 + LLM Judge → grades.json

附录：核心源码文件索引

文件	职责	关键函数/类
`examples/simple_task/main.py`	端到端编排	`start_environment()`, `main()`
`agents/runner/main.py`	Agent 运行入口	`main()`
`agents/runner/agents/registry.py`	Agent 注册表	`get_agent_impl()`, `AGENT_REGISTRY`
`agents/runner/agents/loop_agent/main.py`	循环 Agent 实现	`LoopAgent`, `step()`, `_initialize_tools()`
`agents/runner/agents/models.py`	数据模型	`AgentRunInput`, `AgentTrajectoryOutput`
`agents/runner/utils/llm.py`	LLM 调用封装	`generate_response()`
`agents/runner/utils/mcp.py`	MCP 工具函数	`build_mcp_gateway_schema()`, `content_blocks_to_messages()`

仓库地址：github.com/Mercor-Intelligence/archipelago 关联论文：APEX-Agents (arXiv:2601.14242)