从 System Prompt 到 Error Recovery 再到 Task System：Agent Harness 的工程化构造

发表于2026-06-11|更新于2026-07-04|Agent/Harness

|浏览量:

从 System Prompt 到 Error Recovery 再到 Task System：Agent Harness 的工程化构造

前面几个阶段里，Agent 已经具备了工具调用、Hook 扩展、Todo 规划、Subagent 拆任务、Skill 按需加载、Context 压缩和 Memory 长期记忆。

做到这里，Agent 已经不再是一个简单聊天程序，而是逐渐变成了一个可以处理复杂开发任务的系统。

但是如果继续往工程化方向走，还会遇到三个问题：

system prompt 越来越长，不能再硬编码；
API 调用可能失败，Agent 不能一报错就崩；
Todo 只能管当前会话，不能管理跨会话、带依赖的大任务。

所以这三个章节继续补齐 harness 的三层能力：

s10：System Prompt，运行时组装 prompt；
s11：Error Recovery，错误恢复和重试；
s12：Task System，持久化任务系统。

一、s10 System Prompt：Prompt 应该运行时组装，而不是硬编码

1. 问题：硬编码 SYSTEM 会越来越难维护

最开始的 Agent 只有几个工具，所以 system prompt 可以很简单：

1	SYSTEM = f"You are a coding agent at {WORKDIR}. Use tools to solve tasks."

但是随着功能越来越多，prompt 也越来越长。

后面要告诉模型：

你是 coding agent；
你应该使用工具解决问题；
多步骤任务前要先 todo_write；
可以使用 load_skill 加载技能；
有 memory 时要参考 memory；
当前 workspace 是哪里；
哪些工具可用；
上下文可能被 compact；
遇到任务可以拆给 subagent。

如果全部写成一个大字符串，就会变成：

SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Use tools to solve tasks. Act, don't explain. "
    "Before starting any multi-step task, use todo_write. "
    "Skills are available via list_skills and load_skill. "
    "Relevant memories are injected below when available. "
)

这样有三个问题：

第一，换项目时很难改。

第二，加一个功能可能影响别的说明。

第三，每次请求都带上所有内容，哪怕当前任务根本不需要。

所以 s10 的核心思想是：

system prompt 不应该写死，而应该根据当前运行状态动态组装。

2. PROMPT_SECTIONS：把 Prompt 拆成模块

s10 先把原来的大 prompt 拆成多个片段：

PROMPT_SECTIONS = {
    "identity": "You are a coding agent. Act, don't explain.",
    "tools": "Available tools: bash, read_file, write_file.",
    "workspace": f"Working directory: {WORKDIR}",
    "memory": "Relevant memories are injected below when available.",
}

每个 section 负责一个主题：

identity：Agent 身份和工作方式；
tools：当前可用工具；
workspace：当前工作目录；
memory：记忆相关说明。

这样做的好处是：

每个 prompt 片段可以单独维护，不会互相污染。

比如你要改工具说明，只需要改 tools。

你要加 memory，只需要新增 memory 相关部分。

不用在一个巨大的字符串里到处找位置。

3. assemble_system_prompt：按需组装

有了 section 后，就需要一个组装函数：

def assemble_system_prompt(context: dict) -> str:
    sections = []

    sections.append(PROMPT_SECTIONS["identity"])
    sections.append(PROMPT_SECTIONS["tools"])
    sections.append(PROMPT_SECTIONS["workspace"])

    memories = context.get("memories", "")

    if memories:
        sections.append(f"Relevant memories:\n{memories}")

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

这里分成两类：

第一类是永远加载：

1
2
3

identity
tools
workspace

第二类是按需加载：

memory

如果没有 memory，就不把 memory section 放进去。

这说明 prompt 的组装不是靠猜关键词，而是靠运行时真实状态。

4. context：用真实状态决定 Prompt 内容

s10 用 context 记录当前 harness 的真实状态：

def update_context(context: dict, messages: list) -> dict:
    memories = ""

    if MEMORY_INDEX.exists():
        content = MEMORY_INDEX.read_text().strip()

        if content:
            memories = content

    return {
        "enabled_tools": list(TOOL_HANDLERS.keys()),
        "workspace": str(WORKDIR),
        "memories": memories,
    }

这里面有三个关键字段：

1
2
3

enabled_tools
workspace
memories

它们都不是模型猜出来的，而是 harness 实际检查出来的。

比如：

1	"enabled_tools": list(TOOL_HANDLERS.keys())

表示当前真正注册了哪些工具。

1	"memories": memories

表示 .memory/MEMORY.md 里是否真的有内容。

所以 s10 的 prompt 装配逻辑是：

根据系统真实状态决定加载哪些 prompt section。

这比“看用户有没有提到 memory”更可靠。

5. get_system_prompt：加一层缓存

如果 context 没变，每轮都重新拼 prompt 就没必要。

所以 s10 加了缓存：

def get_system_prompt(context: dict) -> str:
    global _last_context_key, _last_prompt

    key = json.dumps(
        context,
        sort_keys=True,
        ensure_ascii=False,
        default=str
    )

    if key == _last_context_key and _last_prompt:
        return _last_prompt

    _last_context_key = key
    _last_prompt = assemble_system_prompt(context)

    return _last_prompt

这里有一个小细节：为什么不用 hash(context)？

因为 Python 的 dict/list 不能直接 hash。

而且 Python 内置 hash() 有随机化，不适合做稳定缓存 key。

所以这里用：

1	json.dumps(context, sort_keys=True)

把 context 转成稳定字符串。

如果字符串没变，说明 context 没变，就直接返回上一次的 prompt。

6. 整合进 agent_loop

最终循环里这样用：

def agent_loop(messages: list, context: dict):
    system = get_system_prompt(context)

    while True:
        response = client.messages.create(
            model=MODEL,
            system=system,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000
        )

        # tool execution ...

        context = update_context(context, messages)
        system = get_system_prompt(context)

每一轮都会：

根据 context 获取 system prompt；
调用模型；
执行工具；
更新 context；
必要时重新组装 prompt。

7. s10 的核心价值

s10 解决的是 prompt 工程化问题。

以前是：

1	SYSTEM 是一个写死的大字符串

现在是：

1
2
3

PROMPT_SECTIONS 分段管理
context 决定是否加载
get_system_prompt 负责缓存

它让 prompt 从“写死的文本”变成了“运行时配置”。

一句话概括：

Prompt 不应该越写越长，而应该像系统配置一样按模块组装。

二、s11 Error Recovery：错误不是结束，而是重试的开始

1. 问题：Agent 不能一报错就崩

真实环境里，LLM API 调用失败很正常。

常见错误有：

输出太长，被 max_tokens 截断；
上下文太长，出现 prompt_too_long；
请求太频繁，出现 429；
服务过载，出现 529；
网络波动；
超时。

如果没有错误恢复，Agent 只要遇到一次异常就会退出。

这在玩具 demo 里还能接受，但在真实 Agent 里不行。

所以 s11 的核心思想是：

错误不是终点，而是进入恢复路径的起点。

2. 三类主要恢复路径

s11 主要处理三种常见问题：

错误类型	触发条件	恢复方式
输出被截断	`stop_reason == "max_tokens"`	提升 max_tokens 或继续生成
上下文太长	`prompt_too_long`	reactive compact 后重试
临时失败	429 / 529	指数退避重试，必要时切换模型

这三类错误分别放在不同位置处理。

3. RecoveryState：记录恢复状态

错误恢复需要状态。

比如：

是否已经提升过 max_tokens；
是否已经 reactive compact；
当前连续 529 几次；
当前使用哪个模型；
recovery 已经重试几次。

所以 s11 会有类似 RecoveryState 的结构。

伪代码可以理解成：

@dataclass
class RecoveryState:
    has_escalated: bool = False
    has_attempted_reactive_compact: bool = False
    recovery_count: int = 0
    consecutive_529: int = 0
    current_model: str = MODEL

它不是业务逻辑，而是 harness 的运行状态。

4. 路径一：输出被 max_tokens 截断

模型生成太长时，会出现：

1	response.stop_reason == "max_tokens"

这说明模型还没说完，但输出 token 用完了。

s11 的处理分两步。

第一步，先提高输出上限：

if response.stop_reason == "max_tokens":
    if not state.has_escalated:
        max_tokens = ESCALATED_MAX_TOKENS
        state.has_escalated = True
        continue

默认可能是：

1	max_tokens = 8000

第一次截断后提升到：

1	ESCALATED_MAX_TOKENS = 64000

注意这里有一个非常重要的点：

第一次截断时，不把被截断的 assistant 内容加入 messages。

因为它要用更大的 max_tokens 重试同一个请求。

如果把截断内容加入 messages，就会污染上下文。

5. 如果 64K 还不够：续写

如果已经提升过 max_tokens，但还是截断，就保存当前输出，然后加一条续写提示：

messages.append({
    "role": "assistant",
    "content": response.content
})

messages.append({
    "role": "user",
    "content": (
        "Output token limit hit. Resume directly — "
        "no apology, no recap. Pick up mid-thought."
    )
})

然后继续循环。

这条 prompt 的意思是：

不要道歉，不要总结，直接从断掉的地方继续。

续写不是无限的，一般最多几次。

否则模型可能一直写不完，浪费 token。

6. 路径二：prompt_too_long 上下文溢出

如果请求模型时直接报：

1	prompt_too_long

说明上下文太长了。

s08 已经有自动 compact，但可能还是不够。

所以 s11 兜底处理：

except PromptTooLongError:
    if not state.has_attempted_reactive_compact:
        messages[:] = reactive_compact(messages)
        state.has_attempted_reactive_compact = True
        continue

    return

第一次遇到时，执行：

1	reactive_compact(messages)

然后重试。

如果已经 reactive compact 过一次，还是太长，就退出。

因为继续压缩通常也没意义了。

7. 路径三：429 / 529 临时失败

429 和 529 通常不是代码错，而是服务暂时不可用。

所以处理方式不是退出，而是等待后重试。

核心函数：

def retry_delay(attempt, retry_after=None):
    if retry_after:
        return retry_after

    base = min(500 * (2 ** attempt), 32000) / 1000

    return base + random.uniform(0, base * 0.25)

这叫指数退避。

大概是：

第 1 次：0.5 秒
第 2 次：1 秒
第 3 次：2 秒
第 4 次：4 秒
后面最多到 32 秒

再加一点随机 jitter，避免大量请求同时重试。

8. with_retry：包装 API 调用

def with_retry(fn, state, max_retries=10):
    for attempt in range(max_retries):
        try:
            return fn()

        except (RateLimitError, OverloadedError):
            delay = retry_delay(attempt)
            time.sleep(delay)

            if is_overloaded:
                state.consecutive_529 += 1

                if state.consecutive_529 >= 3 and FALLBACK_MODEL:
                    state.current_model = FALLBACK_MODEL

    raise MaxRetriesExceeded()

这里有两个点。

第一，最多重试 10 次。

第二，如果连续 529，说明当前模型过载严重，可以切换 fallback model。

比如从一个更强但拥挤的模型切到一个更稳定的模型。

9. 整合进 agent_loop

s11 的 agent_loop 大概是这样：

def agent_loop(messages, context):
    system = get_system_prompt(context)
    state = RecoveryState()
    max_tokens = 8000

    while True:
        try:
            response = with_retry(
                lambda: client.messages.create(
                    model=state.current_model,
                    system=system,
                    messages=messages,
                    tools=TOOLS,
                    max_tokens=max_tokens
                ),
                state
            )

        except Exception as e:
            if is_prompt_too_long_error(e):
                if not state.has_attempted_reactive_compact:
                    messages[:] = reactive_compact(messages)
                    state.has_attempted_reactive_compact = True
                    continue

                return

            log_error(e)
            return

        if response.stop_reason == "max_tokens":
            if not state.has_escalated:
                max_tokens = 64000
                state.has_escalated = True
                continue

            messages.append({
                "role": "assistant",
                "content": response.content
            })

            messages.append({
                "role": "user",
                "content": CONTINUATION_PROMPT
            })

            continue

        messages.append({
            "role": "assistant",
            "content": response.content
        })

        if response.stop_reason != "tool_use":
            return

        # tool execution ...

注意处理顺序：

API 异常用 try/except；
429/529 用 with_retry；
max_tokens 是 response 返回后的 stop_reason；
正常情况才 append assistant message；
出错恢复后用 continue 回到循环顶部重试。

10. s11 的核心价值

s11 解决的是 Agent 稳定性问题。

以前是：

1	API 报错 → Agent 崩溃

现在是：

1
2
3

输出截断 → 提升 max_tokens / 续写
上下文太长 → reactive compact
429/529 → 指数退避重试 / fallback model

它让 Agent 从“能跑”变成“出错后还能继续跑”。

一句话概括：

工程化 Agent 必须默认错误会发生，并把错误当成流程的一部分处理。

三、s12 Task System：Todo 不是任务系统

1. 问题：TodoWrite 只能管理当前执行步骤

s05 里已经有了 todo_write。

它适合做当前任务的执行清单，比如：

1. 读取文件
2. 修改代码
3. 运行测试
4. 修复失败

但是它有明显限制：

只存在当前进程内；
会话结束就没了；
没有任务依赖；
没有 owner；
不能跨 session 恢复；
不能支持多 Agent 协作。

如果任务变成：

搭数据库
写 API
写测试
写文档

这就不是普通 todo 了。

因为它们有依赖关系：

1
2
3

写 API 依赖数据库
写测试依赖 API
写文档可能依赖数据库和 API

所以 s12 引入 Task System。

核心思想是：

大目标要拆成持久化任务，任务之间可以有依赖关系。

2. TodoWrite 和 Task System 的区别

对比项	TodoWrite	Task System
用途	当前任务执行清单	长期任务管理
存储	内存	`.tasks/{id}.json`
是否跨会话	否	是
依赖关系	没有	blockedBy
owner	没有	有
适合场景	当前步骤跟踪	项目级任务拆解
粒度	Agent 自己的操作步骤	可领取、可阻塞、可恢复的任务

所以 Task System 不是 TodoWrite 的简单升级。

它们是两个层级：

TodoWrite 管“我现在怎么干”；
Task System 管“整个项目有哪些任务，先后顺序是什么”。

3. Task 数据结构

s12 使用 dataclass 表示任务：

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]

每个字段含义：

id

任务唯一 ID。

subject

任务标题。

1	description

任务详细描述。

status

任务状态：

pending
in_progress
completed
owner

当前由谁负责。

blockedBy

当前任务被哪些任务阻塞。

例如：

{
    "subject": "create API endpoints",
    "blockedBy": ["task_schema"]
}

表示 API 任务必须等数据库 schema 完成后才能开始。

4. 文件持久化：.tasks/{id}.json

每个 task 都会保存成一个 JSON 文件：

.tasks/
  task_1710000000_abcd.json
  task_1710000001_ef12.json
  task_1710000002_991a.json

这样即使程序退出，下次启动也可以重新读取任务状态。

这就是 Task System 和 TodoWrite 最大的区别：

Task System 是持久化的。

5. create_task：创建任务

def create_task(
    subject: str,
    description: str = "",
    blockedBy: list[str] | None = None
) -> Task:
    task = Task(
        id=f"task_{int(time.time())}_{random_hex(4)}",
        subject=subject,
        description=description,
        status="pending",
        owner=None,
        blockedBy=blockedBy or [],
    )

    save_task(task)

    return task

创建任务时会：

生成 task id；
设置标题；
设置描述；
默认状态为 pending；
owner 为空；
设置依赖；
写入 .tasks/{id}.json。

6. can_start：判断任务能不能开始

不是所有 pending 任务都能开始。

如果它依赖的任务还没完成，它就不能 claim。

def can_start(task_id: str) -> bool:
    task = load_task(task_id)

    for dep_id in task.blockedBy:
        if not _task_path(dep_id).exists():
            return False

        dep = load_task(dep_id)

        if dep.status != "completed":
            return False

    return True

逻辑很简单：

如果依赖任务不存在，不能开始；
如果依赖任务没完成，不能开始；
所有依赖都 completed，才能开始。

这就是 DAG 任务依赖的基础。

7. claim_task：领取任务

当 Agent 准备处理一个任务时，需要 claim：

def claim_task(task_id: str, owner: str = "agent") -> str:
    task = load_task(task_id)

    if task.status != "pending":
        return f"Task {task_id} is {task.status}, cannot claim"

    if not can_start(task_id):
        deps = [
            d for d in task.blockedBy
            if load_task(d).status != "completed"
        ]

        return f"Blocked by: {deps}"

    task.owner = owner
    task.status = "in_progress"
    save_task(task)

    return f"Claimed {task_id} ({task.subject})"

它会做三步检查：

第一，任务必须是 pending。

如果已经 in_progress 或 completed，就不能 claim。

第二，依赖必须完成。

如果 blockedBy 里有任务没完成，就返回 blocked。

第三，通过后设置：

1 2	task.owner = owner task.status = "in_progress"

然后保存。

8. complete_task：完成任务并解锁下游任务

任务完成后：

def complete_task(task_id: str) -> str:
    task = load_task(task_id)
    task.status = "completed"
    save_task(task)

    unblocked = [
        t.subject
        for t in list_tasks()
        if t.status == "pending"
        and t.blockedBy
        and can_start(t.id)
    ]

    msg = f"Completed {task_id} ({task.subject})"

    if unblocked:
        msg += f"\nUnblocked: {', '.join(unblocked)}"

    return msg

它除了把当前任务设为 completed，还会扫描所有任务，找出刚刚被解锁的任务。

比如原来任务关系是：

1 2	schema → endpoints → tests schema → docs

完成 schema 后：

1 2	endpoints unlocked docs unlocked

完成 endpoints 后：

1	tests unlocked

这就是任务图的推进。

9. get_task：查看完整任务

1
2
3

def get_task(task_id: str) -> str:
    task = load_task(task_id)
    return json.dumps(asdict(task), indent=2)

list_tasks 一般只看摘要。

get_task 用来读取完整任务信息，尤其适合跨会话恢复时使用。

10. 状态机

s12 的任务状态机很简单：

1	pending ──claim──→ in_progress ──complete──→ completed

也就是：

pending：还没人做；
in_progress：已经被领取，正在做；
completed：已经完成。

对应动作：

claim_task：pending → in_progress；
complete_task：in_progress → completed。

教学版没有实现：

1	in_progress → pending

也就是没有 release / rollback。

真实系统里，如果某个 agent 崩了，可能需要把它未完成的任务重新释放给别人。

但教学版为了简单，没有加这个恢复路径。

11. 五个 Task 工具

s12 把任务系统暴露为工具：

create_task
list_tasks
get_task
claim_task
complete_task

它们进入工具表：

TOOL_HANDLERS["create_task"] = create_task
TOOL_HANDLERS["list_tasks"] = list_tasks
TOOL_HANDLERS["get_task"] = get_task
TOOL_HANDLERS["claim_task"] = claim_task
TOOL_HANDLERS["complete_task"] = complete_task

这样模型就可以自己：

创建任务；
列出任务；
查看任务详情；
领取可开始的任务；
完成任务并解锁下游。

12. 一个完整例子

假设用户说：

1 2	创建任务：搭数据库、写 API、写测试、写文档。 API 依赖数据库，测试依赖 API，文档依赖数据库。

任务关系是：

setup database schema
    ├── create API endpoints
    │       └── write tests
    └── write docs

创建任务：

schema = create_task("setup database schema")

endpoints = create_task(
    "create API endpoints",
    blockedBy=[schema.id]
)

tests = create_task(
    "write tests",
    blockedBy=[endpoints.id]
)

docs = create_task(
    "write docs",
    blockedBy=[schema.id]
)

开始执行：

1 2	claim_task(schema.id) complete_task(schema.id)

此时解锁：

1 2	create API endpoints write docs

继续：

1 2	claim_task(endpoints.id) complete_task(endpoints.id)

此时解锁：

1	write tests

最后：

claim_task(docs.id)
complete_task(docs.id)

claim_task(tests.id)
complete_task(tests.id)

这个流程就保证了：

不会在数据库没建好之前先写 API，也不会在 API 没完成前先写测试。

13. s12 的核心价值

s12 解决的是项目级任务管理问题。

以前 TodoWrite 是：

1	当前会话里的临时 checklist

现在 Task System 是：

1	跨会话、可恢复、带依赖、可领取的任务图

它让 Agent 可以处理更长周期的项目任务。

一句话概括：

TodoWrite 管执行步骤，Task System 管项目进度。

四、三章合起来的 Harness 构造顺序

把 s10、s11、s12 串起来，可以得到一个更工程化的 harness。

第一步：把硬编码 SYSTEM 拆成 PROMPT_SECTIONS

PROMPT_SECTIONS = {
    "identity": "...",
    "tools": "...",
    "workspace": "...",
    "memory": "...",
}

第二步：用 context 表示真实运行状态

context = {
    "enabled_tools": list(TOOL_HANDLERS.keys()),
    "workspace": str(WORKDIR),
    "memories": memory_index,
}

第三步：运行时组装 system prompt

1	system = get_system_prompt(context)

如果 context 没变，就走缓存。

如果 context 变了，就重新：

1	assemble_system_prompt(context)

第四步：用 try/except 包住 LLM 调用

try:
    response = with_retry(
        lambda: client.messages.create(...)
    )
except PromptTooLongError:
    messages[:] = reactive_compact(messages)
    continue
except Exception:
    return

这样 Agent 不会因为普通 API 错误直接崩。

第五步：处理 max_tokens 截断

if response.stop_reason == "max_tokens":
    if not state.has_escalated:
        max_tokens = 64000
        state.has_escalated = True
        continue

    messages.append({"role": "assistant", "content": response.content})
    messages.append({"role": "user", "content": CONTINUATION_PROMPT})
    continue

这一步解决输出被截断的问题。

第六步：为 429 / 529 做指数退避

1
2
3

def retry_delay(attempt):
    base = min(500 * (2 ** attempt), 32000) / 1000
    return base + random.uniform(0, base * 0.25)

这一步解决临时失败问题。

第七步：创建 `.tasks/` 目录

.tasks/

每个任务一个 JSON 文件。

第八步：定义 Task 数据结构

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]

第九步：实现任务工具

create_task
list_tasks
get_task
claim_task
complete_task

然后注册到：

1	TOOL_HANDLERS

第十步：Agent 使用任务系统推进项目

完整流程是：

用户提出大目标
↓
Agent 创建多个 task
↓
用 blockedBy 建立依赖
↓
list_tasks 找可执行任务
↓
claim_task 领取任务
↓
执行真实工作
↓
complete_task 完成任务
↓
解锁下游任务
↓
继续推进

五、最终 Agent Harness 流程

整合后，一轮完整流程可以理解为：

更新 context
↓
组装 system prompt
↓
调用 LLM
    ↓
    如果 429/529：
        指数退避重试
    如果 prompt_too_long：
        reactive compact 后重试
    如果 max_tokens：
        提升 max_tokens 或续写
↓
正常得到 response
↓
如果 response 是普通回答：
    Stop Hook
    结束
↓
如果 response 是 tool_use：
    PreToolUse Hook
    ↓
    TOOL_HANDLERS 分发
        - 基础工具
        - load_skill
        - compact
        - task/subagent
        - create_task
        - claim_task
        - complete_task
    ↓
    PostToolUse Hook
    ↓
    tool_result 写回 messages
↓
继续下一轮

六、总结

这三个章节把 Agent harness 从“功能可用”进一步推向“工程可用”。

s10 解决 prompt 维护问题：

system prompt 不再硬编码，而是根据运行状态动态组装。

s11 解决运行稳定性问题：

API 报错、上下文过长、输出截断，都进入对应恢复路径。

s12 解决项目任务管理问题：

大目标被拆成持久化任务，任务之间有依赖，可以跨会话恢复。

所以这三个阶段的核心关系是：

1
2
3

System Prompt 让 Agent 知道自己是谁、有什么能力；
Error Recovery 让 Agent 出错后还能继续；
Task System 让 Agent 能管理长期项目目标。

如果说前面的章节是在教 Agent “怎么行动”，那么这三章是在教 harness “怎么管理一个长期运行的 Agent”。

一句话概括：

真正工程化的 Agent，不只是会调用工具，还要能动态组装上下文、从错误中恢复，并用任务系统管理长期目标。

文章作者: wuyukeng1

文章链接: https://wuyukeng1.github.io/2026/06/11/%E4%BB%8E%20System%20Prompt%20%E5%88%B0%20Error%20Recovery%20%E5%86%8D%E5%88%B0%20Task%20System%EF%BC%9AAgent%20Harness%20%E7%9A%84%E5%B7%A5%E7%A8%8B%E5%8C%96%E6%9E%84%E9%80%A0/

Claude Code Agent Harness 工程化

相关推荐

2026-06-07

从 Hooks 到 TodoWrite 再到 Subagent：一个 Claude Code 风格 Harness 的构造过程

从 Hooks 到 TodoWrite 再到 Subagent：一个 Claude Code 风格 Harness 的构造过程一、什么是 Harness？在 Agent 系统里，LLM 本身只负责“思考”和“决定下一步”。但真正让它能稳定工作的，不只是模型，而是模型外面那层控制结构，也就是 harness。可以把 harness 理解成：包住 Agent 的运行外壳，负责接收用户输入、调用模型、分发工具、拦截危险操作、记录状态、控制循环退出，以及在复杂任务中帮 Agent 维持方向。这三个文件刚好按顺序展示了一个 harness 的逐步增强过程： s04：加入 Hooks，让循环可扩展； s05：加入 TodoWrite，让 Agent 先规划再执行； s06：加入 Subagent，让复杂任务拆出去，用干净上下文处理。第一部分：s04 Hooks —— 不要把扩展逻辑写死在循环里1. 问题：agent_loop 会越来越臃肿在 s03 里，Agent 已经能调用工具，也能做权限检查。但是问题来了：如果后面还想加这些功能：每次 bash 调用都写日志；写...

2026-06-13

从 Team Protocol 到 MCP：Claude Code Harness 后半程架构演进分析

从 Team Protocol 到 MCP：Claude Code Harness 后半程架构演进分析前面十几个章节里，我们已经完成了一个 Agent 的基本能力： Tool Calling Hook Todo Memory Task System Background Task Scheduler Team 看到这里会发现一个变化。最开始的 Agent 其实只有一个循环： 123while True: response = llm(...) execute_tool(...) 但到了 Team 阶段之后，系统已经不再是一个 Agent。而变成了： 1234Lead├── Alice├── Bob└── Charlie 这时候很多以前不存在的问题开始出现： Agent 之间如何协作？ Agent 如何自己找任务？ Agent 之间如何避免修改同一个文件？如何接入外部系统？ s16～s19 实际上都在解决这些问题。一、s16：为什么 Team 需要协议s15 里面已经有 Team 了。 Lead 可以给 Alice 发消息： 12345BUS.se...

2026-06-07

从 Skill Loading 到 Context Compact 再到 Memory：Agent Harness 的持续运行能力构造

从 Skill Loading 到 Context Compact 再到 Memory：Agent Harness 的持续运行能力构造前面几个阶段里，Agent 已经具备了基础工具调用、Hook 扩展、Todo 规划、Subagent 拆任务等能力。但做到这里还不够。一个真正能长期工作的 Agent，还会遇到三个更现实的问题：项目知识太多，不能全部塞进 system prompt；上下文会越来越长，最终超过模型窗口；压缩会丢失细节，新会话也无法记住用户偏好。所以这三个章节继续增强 harness： s07：Skill Loading，按需加载知识； s08：Context Compact，压缩上下文； s09：Memory，长期保存重要信息。一、s07 Skill Loading：知识不要全塞进 Prompt，要按需加载1. 问题：system prompt 不是垃圾桶假设项目里有很多规范文档： React 组件规范； SQL 编写规范； API 设计规范；代码审查规范。最直接的做法是把它们全部塞进 system prompt： 123456SYS...

2026-06-11

从 Background Tasks 到 Cron Scheduler 再到 Agent Teams：Agent Harness 的异步化与多 Agent 协作演进

从 Background Tasks 到 Cron Scheduler 再到 Agent Teams：Agent Harness 的异步化与多 Agent 协作演进在前面的章节中，Agent 已经具备： Tool Calling Hooks Todo Planning Subagent Skill Loading Context Compact Memory Error Recovery Task System 此时的 Agent 已经能够独立完成复杂任务。但还有三个明显瓶颈：问题一：等待Agent 执行： 1npm install 可能需要 5 分钟。执行： 1pip install torch 可能需要十几分钟。这期间 Agent 什么都做不了。问题二：只能手动触发Agent 只能这样工作： 123用户说一句↓Agent 做一次无法做到： 12每天 9 点跑测试每隔 5 分钟检查服务问题三：单 Agent 注意力有限当任务变成： 123456重构整个后端重构数据库修改鉴权修改 API补测试写文档一个 Agent 的上下文根本装不下。因此：...

2026-05-20

初识智能体：看完这篇文章，我终于把 Agent 给串起来了

📝 学习笔记：初识智能体——看完这篇文章，我终于把”Agent”给串起来了大家好，我是小元。之前我在博客里梳理了 JavaScript 的全地图，那篇总结帮我打通了前端基础里的很多关节。最近我把目光转向了 AI，尤其是大语言模型带火的“智能体（Agent）”。但网上资料太多太杂，一上来就是各种框架、多智能体协作，看得我一头雾水。直到我读到了这篇《第一章初识智能体》，它像一张精心绘制的导览图，帮我从零开始理解了智能体的全貌。今天这篇博客，就是我的学习笔记。我想用自己的话，把这一章的内容重新说一遍。如果你也刚入门，希望它能帮到你。 🤖 什么是智能体？——一个会看、会想、会动的闭环文章一开头就给了一个非常清晰的定义：智能体 = 通过传感器感知环境 + 通过执行器采取行动 + 达成目标的自主实体。我把它理解成：一个能“看到”周围、“思考”下一步、“动手”改变世界的家伙。对于自动驾驶汽车，环境是马路；对于交易算法，环境是金融市场。而传感器可能是摄像头、API 接口，执行器可能是方向盘或者一段代码调用。最关键的是“自主性”——它不是机械地执行死命令，而是能自己判断、自...

2026-05-24

Claude Code 下载与安装指南

Claude Code 下载与安装指南前提条件 1. 安装 Node.js（版本 ≥ 18）Windows / macOS 直接前往( Node.js 官网) ，点击获取node.js 选择x64，点击windows安装程序，下载好后一路点击安装程序后，点击next直至安装完毕接着按win键搜索cmd，打开命令提示符，输入： 1node -v 出现版本号代表安装完毕（如下）顺便设置一下国内镜像源，不然接下来可能安装失败，在命令提示符中输入： 1npm config set registry https://registry.npmmirror.com/ 回车后安装 2. 安装 Git（推荐）Claude Code 的文件操作和提交功能依赖 Git。 Git 官网下载并安装或者在浏览器中搜索git，点击进入官网，移动到底端找到在windows上安装，打开(https://git-scm.com/install/windows) 打开后选择x64 setup 之后安装步骤按照node.js来完成后打开终端/命令提示符验证： 12Bashgit -...

从 System Prompt 到 Error Recovery 再到 Task System：Agent Harness 的工程化构造

一、s10 System Prompt：Prompt 应该运行时组装，而不是硬编码

1. 问题：硬编码 SYSTEM 会越来越难维护

2. PROMPT_SECTIONS：把 Prompt 拆成模块

3. assemble_system_prompt：按需组装

4. context：用真实状态决定 Prompt 内容

5. get_system_prompt：加一层缓存

6. 整合进 agent_loop

7. s10 的核心价值

二、s11 Error Recovery：错误不是结束，而是重试的开始

1. 问题：Agent 不能一报错就崩

2. 三类主要恢复路径

3. RecoveryState：记录恢复状态

4. 路径一：输出被 max_tokens 截断

5. 如果 64K 还不够：续写

6. 路径二：prompt_too_long 上下文溢出

7. 路径三：429 / 529 临时失败

8. with_retry：包装 API 调用

9. 整合进 agent_loop

10. s11 的核心价值

三、s12 Task System：Todo 不是任务系统

1. 问题：TodoWrite 只能管理当前执行步骤

2. TodoWrite 和 Task System 的区别

3. Task 数据结构

4. 文件持久化：.tasks/{id}.json

5. create_task：创建任务

6. can_start：判断任务能不能开始

7. claim_task：领取任务

8. complete_task：完成任务并解锁下游任务

9. get_task：查看完整任务

10. 状态机

11. 五个 Task 工具

12. 一个完整例子

13. s12 的核心价值

四、三章合起来的 Harness 构造顺序

第一步：把硬编码 SYSTEM 拆成 PROMPT_SECTIONS

第二步：用 context 表示真实运行状态

第三步：运行时组装 system prompt

第四步：用 try/except 包住 LLM 调用

第五步：处理 max_tokens 截断

第六步：为 429 / 529 做指数退避

第七步：创建 .tasks/ 目录

第八步：定义 Task 数据结构

第九步：实现任务工具

第十步：Agent 使用任务系统推进项目

五、最终 Agent Harness 流程

六、总结

第七步：创建 `.tasks/` 目录