理解 Agent 开发过程：harness + LLM 的任务循环

LLM API：一次调用 = 发 JSON 进、收 JSON 出

大模型本身是一个 HTTP 服务。你按它规定的 JSON（JavaScript Object Notation，一种人类可读的文本数据格式）格式发请求，它按规定的 JSON 格式回响应。这个 JSON 格式就是协议（llm-protocol-fundamentals.html 专门讲）。理解 LLM API 的关键：它就是"一次请求一次响应"，不"记住"任何前文，没有"自动做事"的能力——你看到的"Agent 自动完成多步任务"，全是外面包了一层循环代码（那就是 harness，下一节讲）。这里 LLM API 指"通过 HTTP 接口调用大模型"——API（Application Programming Interface，应用程序接口）即程序之间约定的调用方式。

curl https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "1+1=?"}]
  }'

# → 响应（简化）
{
  "choices": [{
    "message": {"role": "assistant", "content": "1+1 等于 2。"},
    "finish_reason": "stop"
  }]
}

from openai import OpenAI
client = OpenAI()                                    # 从环境变量读 API key
r = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "1+1=?"}],
)
print(r.choices[0].message.content)                  # → "1+1 等于 2。"

• ❌ 没有持久状态：每次调用都得把全部历史塞进 messages，否则 LLM 不知道之前说了啥
• ❌ 不能调外部工具：默认只会"说"，不能"做"（查数据库、改文件、跑命令）
• ❌ 不知道何时停：复杂任务需要"反复调 LLM → 看结果 → 决定下一步"，LLM 自己不知道什么时候该收手
• ❌ 不处理错误：API 超时、JSON 解析失败、工具执行报错——全得有人兜底

把这 4 件事补齐的"那一层"——就是 harness。下一节破题。

一个等式：Agent = harness（你写的代码）+ LLM（外部 API）

所有"Agent"——无论是 LangChain、AutoGPT、还是本仓 SWE-bench（Software Engineering benchmark，用真实 GitHub issue 评测代码 agent 的软件工程基准测试）的 S1-S7——拆到底都是同一个结构：一段你自己写的代码（harness），去调一个外部 LLM 服务（OpenAI / Anthropic / Ollama），中间穿插工具调用和状态管理。"Agent"不是一个新物种，是一种编程范式。下面这张组件图把 ownership 划清楚。

• LLM 不知道任务进度：它只看到当前这一轮的 messages，不知道"我正在做第 3 步 / 总共 10 步"
• LLM 不知道工具的结果格式：你把工具输出塞进 role:"tool" 消息，它才"看见"
• LLM 不知道历史轨迹的元信息：时间戳、token 用量、工具调用次数——这些都在 harness 的 logger 里
• 所以 "AI 自动完成多步任务"这个体验 = harness 的循环 + LLM 的每步推理，缺一不可

拆开 harness：6 大组件 + 1 张速查表

所有成熟的 Agent 框架（LangGraph、AutoGen、Claude Agent SDK、本仓 agent_runtime）拆到底都是这 6 个组件的排列组合。理解这 6 个，你就看穿了所有 Agent 框架的"魔法"。

职责：维护 messages[] / items[]，每次循环后追加新消息（assistant 回复、tool 结果）。这是 LLM 唯一能"看到历史"的窗口。

messages = [{"role": "user", "content": "帮我看看 README"}]
# 循环里：messages.append(assistant_msg); messages.append(tool_result_msg)

职责：声明 LLM 能调的工具（tools=[...]），并提供实际执行函数（name → callable 的映射）。声明与实现解耦。

TOOL_REGISTRY = {
    "read_file": lambda p: open(p).read(),
    "bash":      lambda c: subprocess.run(c, shell=True, capture_output=True).stdout,
}
tools = [{"type": "function", "function": {"name": n, "description": "...", "parameters": {...}}} for n in TOOL_REGISTRY]

职责：从 LLM 响应里抽出 content / tool_calls / finish_reason。手写 Agent 最常踩的坑都在这里：arguments 是 JSON 字符串需要 json.loads；流式 chunk 要按 index 聚合；finish_reason="tool_calls" 时 content 可能是 null。

msg = response.choices[0].message
if msg.tool_calls:
    for tc in msg.tool_calls:
        args = json.loads(tc.function.arguments)   # ← 字符串，不是对象！
        result = TOOL_REGISTRY[tc.function.name](**args)
elif msg.finish_reason == "stop":
    final_answer = msg.content

职责：while not done，附带三层守门：max_iterations（防死循环）/ wall_time（防超时）/ max_tokens（防上下文爆）。

for turn in range(MAX_ITERATIONS):                  # 守门 1：轮次
    if time.time() - start > MAX_WALL_TIME: break   # 守门 2：墙钟
    r = client.chat.completions.create(model=..., messages=messages, max_tokens=...)  # 守门 3：单次上限
    ...

职责：网络超时 → 重试；tool_call_id 配错 → 校验抛错；上下文超限 → 摘要压缩；LLM 返回非法 JSON → 兜底默认值。没有错误处理，Agent 跑 10 次崩 8 次。

try:
    r = client.chat.completions.create(..., timeout=60)
except openai.APITimeoutError:
    time.sleep(2 ** retry_count)                    # 指数退避
    continue
except json.JSONDecodeError:                        # arguments 不是合法 JSON
    return fallback_default

职责：全程旁路记录每轮的 messages、tool_calls、tool 结果、token 用量、墙钟。没有 logger，Agent 跑完你不知道它怎么做的——所有"为什么这次没成功"的事后分析都靠这个。本仓的 experiments/<id>/s4_worker.log + result.json 就是这一层。

log.write({
    "turn": turn, "ts": now(),
    "messages_snapshot": messages[-3:],            # 最近 3 条，避免日志爆炸
    "tool_call": msg.tool_calls, "tool_result": result,
    "usage": r.usage.total_tokens, "wall_time": time.time() - start,
})

LLM 知道的 vs 不知道的：每次调用是个"无状态函数"

LLM 每次被调用都是一个无状态函数：给它 messages + tools，它返回 content + tool_calls。它不会"记住"上一轮，不会"知道"你之前跑过什么工具、用了多少 token、剩多少预算。所有这些信息都得由 harness 显式塞进 messages 里，LLM 才能"看见"。

主循环伪代码：每行注释 ownership（harness / LLM / tool）

把 6 大组件、3 大守门、所有"知道/不知道"的边界，收成一段可读的伪代码。读完这段你就拿到了 Agent 开发的"主模板"——LangGraph、AutoGen、各 brand adapter 的源码都是这个模板的变体。

def run_agent(user_query: str, max_iter: int = 20, wall_time: int = 600):
    # ====== ① 初始化：State Manager ======
    messages = [{"role": "user", "content": user_query}]              # [harness]
    tools = declare_tools()                                          # [harness] 工具定义
    start = time.time()                                              # [harness] Loop 守门 2 的起点

    # ====== ② 主循环：Loop Controller ======
    for turn in range(max_iter):                                     # [harness] 守门 1：轮次
        if time.time() - start > wall_time:                          # [harness] 守门 2：墙钟
            return TimeoutError(f"超出 {wall_time}s")

        # ====== ③ 调 LLM ======
        response = client.chat.completions.create(                   # [LLM] ←—— 唯一的"AI 时刻"
            model=MODEL, messages=messages, tools=tools,
            max_tokens=4096,                                         # [harness] 守门 3：单次 token 上限
        )
        log(response.usage)                                          # [harness] Logger

        # ====== ④ Output Parser ======
        msg = response.choices[0].message                            # [harness] 拆出 assistant 消息
        messages.append(msg)                                         # [harness] State: 写回
        log({"turn": turn, "msg": msg})                              # [harness] Logger

        # ====== ⑤ 分支 ======
        if msg.finish_reason == "stop":                              # [harness] 自然结束
            return msg.content                                       # → 返回最终答案

        if msg.finish_reason == "length":                            # [harness] 截断
            raise LengthError("撞 max_tokens，需扩预算或续写")

        if msg.tool_calls:                                           # [harness] 模型要调工具
            for tc in msg.tool_calls:                                # [harness] 多个并行调用
                args = json.loads(tc.function.arguments)             # [harness] 字符串 → 对象
                try:
                    result = TOOL_REGISTRY[tc.function.name](**args) # [tool]  ←—— 副作用在这里
                except Exception as e:
                    result = f"工具执行失败：{e}"                     # [harness] Error Handler 兜底
                messages.append({                                    # [harness] State: 写回 tool 结果
                    "role": "tool", "tool_call_id": tc.id,
                    "content": str(result),
                })
            continue                                                 # [harness] 回到循环顶，再调一次 LLM

    raise MaxIterError(f"超出 {max_iter} 轮，可能死循环")            # [harness] 守门 1 触发

本仓 S1-S7：6 个 stage 怎么对应到 harness loop

本仓 experiment_modules/orchestration/ 有一个完整的工业级 Agent pipeline：出题 → 准备环境 → 作答 → 抽 diff → 评分 → 记录。看着是 6 个 stage，核心循环只发生在 S4_solve 一个 stage 里——其他 5 个都是它的"周边"。下表把映射关系摊开。

{
  "instance_id": "django__django-10914",
  "model": "qwen-code",
  "resolved": true,                            // ← S6 评分结果：F2P + P2P 全过
  "stages": {
    "S1_build": "done", "S2_prepare": "done",
    "S4_solve": "done", "S5_patch": "done",
    "S6_score": "done", "S7_record": "done"
  },
  "stage_timings": {                            // ← Loop Controller 的 3 大守门日志
    "S1_build": 0.1, "S2_prepare": 16.8,
    "S4_solve": 420.4,                          // ← 420s = 7 分钟 agentic 循环
    "S5_patch": 0.0, "S6_score": 162.4
  },
  "adapter_attempts": [{
    "adapter": "qwen-agent", "subprocess": true,
    "wall_time_seconds": 420.3, "returncode": -15,
    "killed": true, "rescued": true,            // ← 被 SIGTERM（墙钟超时），但已被 S5 抽到 diff
    "outcome": "resolved"
  }]
}

30 行最小可运行 harness（拷走就能跑）

用 OpenAI 官方 SDK 写一个最小 harness：模型可以调 get_weather 工具查天气，多轮循环直到 finish_reason="stop"。整个文件 30 行，对应卡片 5 的伪代码每一行。

import json, openai
client = openai.OpenAI()                                          # 从 OPENAI_API_KEY 读

def get_weather(city: str) -> str:                                 # 工具实现
    return f"{city} 25°C 晴"

messages = [{"role": "user", "content": "北京今天几度？"}]
tools = [{"type": "function", "function": {
    "name": "get_weather", "description": "查天气",
    "parameters": {"type": "object",
                   "properties": {"city": {"type": "string"}},
                   "required": ["city"]}}}]

for turn in range(10):                                            # 守门 1
    r = client.chat.completions.create(
        model="gpt-4o-mini", messages=messages, tools=tools)
    msg = r.choices[0].message
    messages.append(msg)                                           # State
    if msg.tool_calls:                                             # 工具分支
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            messages.append({"role": "tool",
                             "tool_call_id": tc.id,
                             "content": get_weather(**args)})
    else:                                                          # 自然结束
        print(msg.content)
        break

主循环的形状决定了 Agent 的"性格"。同样 6 大组件，循环形状不同就变成不同的范式。

1. 任务能"一眼看穿"步骤吗？ 能 → Plan-and-Execute；不能 → ReAct
2. 结果对质量敏感吗（代码 / 文档 / 数据）？ 是 → Reflection；否 → ReAct 即可
3. 需要多个"角色"协作吗（计划 / 执行 / 评审）？ 是 → Multi-Agent；否 → 单 loop 起步
4. 成本敏感吗？ 敏感 → 优先 ReAct（每轮 token 最少）；不敏感 → Reflection / Multi-Agent 都可以试