🏛 给 AI 配一间办公室：Harness Engineering 六大模块与它的实现

> 写在前面：今天这节课的主题叫 **Harness Engineering（智能体工程）**，readme 开篇一句话给出了定义——**"用工程化手段，让 AI Agent 可靠、持续完成任务。"** 这句话里有两个关键词：**可靠**、**持续**。会调 API 的人很多，但能让 AI 稳定地干完一件长活的人不多。上几节课学的 Memory、LangGraph、结构化输出、SSE——这些其实都是 Harness 的零件。今天这节是总纲：把这些零件装进同一张图纸。readme 还给出了对应的岗位画像——"AI 应用开发工程师（Agentic RAG 应用）、AI Agent 开发工程师（Harness Agent FDE）"。以下所有代码和概念均来自课堂真实文件。

一、LLM 是个没入职的应届生

先想一个问题——LLM 那么聪明，为什么不能直接拿来干活？

因为它是个**"裸脑"**：会说话、会推理，但：

缺什么	后果
没有手	不能读写文件、不能跑命令
没有记忆	关掉即失忆（前面讲过的"金鱼"）
没有边界	要是真给了它命令执行权，它能 `rm -rf /`
没有履历	出了问题你不知道它刚才做了什么

Harness 就是给这颗"裸脑"配的一整套办公设施——工位、工具箱、资料柜、门禁、笔记本、监控。装完之后，它才从"聊天机器人"变成"Agent"。

readme 里那句被反复提到的公式，今天终于要完整展开了：

<span>Agent</span> = LLM（大脑）+ Harness（tool + mcp + rag + skill ... + ...）

二、六大基础模块：一间办公室的六件标配

readme 列了六个模块，说得很清楚：

"包含 6 个基础模块。"

模块	职责	办公室类比
Loop 主循环	核心控制层，调用大模型、分发工具调用、判断终止条件	工作节奏
Tool 工具集	LLM 能使用的工具集	工具箱
Context 上下文管理器	输入给模型的全部内容 + 压缩 + 窗口控制	桌上的资料
Environment 沙箱环境	工具运行时的隔离环境	工位与门禁
Memory 记忆层	跨轮次、跨会话状态持久化	笔记本
Observability 可观察性	日志、Trace、评估指标、调试、回放、评测	监控与复盘

逐个看。

1. Loop 主循环：心跳

readme 原文：

"Loop 主循环——自主长时间干活。ReAct。核心控制层，调用大模型，分发工具调用，判断任务终止条件。Agent 的主执行流。"

这是整个 Agent 的心脏。它干的事很简单，但必须永不停歇：

while (任务未完成) {
<span>    1.</span> 把当前上下文发给 LLM
<span>    2.</span> LLM 决定：直接回答？还是调用工具？
<span>    3.</span> 如果调工具 → 执行 → 把结果塞回上下文 → 回到第 1 步
<span>    4.</span> 如果直接回答 → 结束
}

这就是 ReAct（Reason + Act）——推理一步、行动一步、观察结果、再推理。前面讲 Memory 时提到过这个流程，今天它是六大模块里的第一号。

关键在"判断终止条件"——LLM 可能陷入死循环（反复调同一个工具），也可能话说到一半就停。主循环必须管住这两件事。

2. Tool 工具集：给它一双手

"Tool 工具集，llm 能使用的工具集。"

工具的本质是把 LLM 的"文字能力"变成"行动能力"。前面学的结构化输出、Tool Call、JSON Schema 全部用在这里——每个工具都需要：

名字（bash、read_file）
描述（LLM 靠它判断什么时候用）
参数 Schema（LLM 靠它生成合法参数）

agent.py 里有 4 个基础工具（下面细讲）——bash、read_file、write_file、edit_file。就这四个，已经能覆盖"写代码"这件事的绝大部分场景了。

3. Context 上下文管理器：资料怎么摆

"Context 上下文管理器。无状态。输入给模型的全部内容：系统提示词、历史对话、rag、检索、es 上下文。工具返回结果。上下文压缩、窗口控制。"

这里有个反直觉的点——readme 特意标了**"无状态"**。

LLM 本身是无状态的（每次调用都不记得上次），所以"上下文管理器"的职责就是：每次调用前，把该给的资料全都准备好，摆到它面前。

包括：

内容	来源
系统提示词	代码里写死的 SYSTEM
历史对话	messages 数组 / Memory
RAG 检索结果	向量数据库（前面学的 Milvus）
工具返回结果	上一轮工具的执行输出

而且还要管**"窗口控制"和"上下文压缩"**——前面学的截断、总结、检索，就是这一模块的具体手段。工具返回了 50000 字符的结果？得压缩。历史对话太长了？得截断。

4. Environment 沙箱环境：工位与门禁

"Environment 沙箱环境。工具运行时的隔离环境，文件系统、网络、权限、工作区隔离。"

这条是安全底线。你不能让 AI 在整台电脑上随便跑命令——它得被关在一个"工位"里：

文件系统：只能读写指定目录
网络：能访问哪些地址
权限：能执行哪些命令
工作区隔离：一个任务的改动不能污染另一个任务

agent.py 里的 safe_path() 就是这一模块的落地——所有文件操作都要先过这道安检。

5. Memory 记忆层：笔记本

"Memory 记忆层。跨轮次、跨会话状态持久化状态，短期记忆 + 长期记忆。"

这一模块前面上了整整两节课（金鱼记忆上/下篇）——短期用内存/文件，长期用向量数据库。今天不重复，只强调它的定位：Memory 是六大模块之一，不是可选项。

6. Observability 可观察性：监控与复盘

"Observability 可观察性。日志（分析）、Trace、评估指标、调试、回放、评测。"

这是最容易被忽视、但在生产环境最救命的一环。

想象一下：Agent 跑了 20 步，最后给了一个错误答案。你怎么办？

没有 Observability：一脸茫然，不知道哪一步错了
有 Observability：看 Trace，第 7 步读错了文件 → 第 12 步基于错误信息推理 → 定位到根因

readme 列了五个能力：日志、Trace、评估指标、调试、回放、评测。其中"回放"最有意思——把整个执行过程重放一遍，像看录像一样复盘。

三、高级模块：从单人到团队

readme 说得很清楚：

"6 大基础模块是最小内核；sub-Agents 属于高级编排的扩展模块，不是底层必选，但现代 Agent Harness 工程体系普遍把它作为标准模块。"

除了六大内核，还有几个可插拔的高级模块：

模块	作用
Sub Agents	子智能体，任务拆分与并行
hooks	生命周期钩子
Policy & Safety	权限、输出过滤、资源配额

Sub Agents：外包团队

"职责清晰，拆分。上下文互不干扰。主 Agent 分配任务给子 Agent，子 Agent 像子进程，上下文不被打扰。主 Agent 也不会因为子 Agent 的运行，上下文受拖累。"

"上下文互不干扰"——这七个字是 Sub Agent 存在的根本理由。

举个具体例子，readme 里那个"多 Agent 分工"：

Agent <span>A</span> 写前端代码
Agent <span>B</span> 写后端代码
Agent C 写测试
Agent D 部署

如果全让一个 Agent 干——它的上下文里会堆满前端代码、后端代码、测试代码、部署脚本，全都搅在一起。而拆成四个子 Agent：

A 的上下文里只有前端相关
B 的上下文里只有后端相关
主 Agent 只需要知道"A 完成了、B 完成了"

readme 还点出了"按需加载工具"这一层：

"派发任务给子 Agent，独立的运行上下文，按需加载 tools，不用一次性加载那么多。"

这也呼应了前面 LangGraph 那节课讲的——单 Agent 把所有工具描述都塞进 system prompt，token 贵还干扰思考。 拆成子 Agent 后，每个子 Agent 只带自己需要的工具。

四、为什么是 Agent，而不是 Workflow？

这是 readme 里最有含金量的一段，值得单独开一节。

"固定的 workflow 是预先写死的步骤，（coze/dify/n8n 23-25 年，较固定、简单的任务），只能按预设路径执行，无法应对不确定、动态变化的任务。多 Agent（含 subAgent）可自主拆解任务，按需调用能力，根据中间结果调整执行分支，适合需求模糊、存在未知问题的复杂场景，具备更强的自适应与容错能力。"

Workflow 是一台自动售货机——投币、选号、出货。路径写死了，遇到问题不会变通。

Agent 是一个员工——你交给他一个模糊的任务，他自己决定先做什么、用什么工具、遇到问题怎么绕。

readme 给了一个特别好的对比案例：做一份行业技术调研报告。

简单版本：Workflow 就够了

问题 → 行业关键字 → 技术关键字 → 上网搜 → 分析 → 生成报告

六个固定步骤，一气呵成。这种任务路径明确、没有意外，Workflow 又稳又便宜。

专业版本：必须上多 Agent

"任务是专业的。分析需要哪些 agent？"

readme 列出的团队配置：

角色	职责
主 Agent	负责任务拆解与整体调度
子 Agent（检索）	检索资料、反爬、判断资料优劣、自动分析、切换数据源
摘要 Agent	压缩提取
校验 Agent	"不要信资料，网上要有些分辨的"
情报 Agent	找到行业内大佬的联系方式
报告 Agent	生成最终报告

注意那个校验 Agent 的说明——readme 原话：

"不要信资料，网上要有些分辨的。"

这句话道破了多 Agent 的真正价值——不是并行提速，而是互相制衡。 检索 Agent 找来的资料，要经过校验 Agent 质疑真伪。跟前面 LangGraph 那节课提到的"AutoGen 法庭"是同一个思路。

而 Workflow 做不到这一点——它的第六步永远是"生成报告"，不会因为第五步发现资料可疑就临时插入一个"再查一遍"的分支。

动态调整执行分支，这才是 Agent 相对 Workflow 的质变。

五、多 Agent 的业务隔离：三件套

readme 末尾列了多 Agent 隔离的三个手段：

"专属 SYSTEM prompt（多个 system，主 system 规划、分工，子 system 执行任务）；子进程；独立的上下文（子 Agent 返回结果，有全新的上下文，不会干扰主 Agent 上下文）。"

隔离手段	隔离了什么
专属 SYSTEM prompt	角色定位——主 Agent 规划、子 Agent 执行
子进程	进程空间
独立上下文	对话历史

这三条在 agent.py 里全部有对应实现，我们马上进代码。

六、338 行代码：把六大模块装进一间办公室

现在进入今天的重头戏——agent.py。这份代码不长（338 行），但六大模块一个不少。我们按模块拆。

配置层：办公室的门牌号

<span>import</span> os
<span>import</span> re
<span>import</span> subprocess
<span>from</span> pathlib <span>import</span> Path
<span>import</span> json
<span>from</span> openai <span>import</span> OpenAI
<span>from</span> dotenv <span>import</span> load_dotenv

load_dotenv(override=<span>True</span>)
<span># Agent 工作目录 安全的，被授权的</span>
WORKDIR = Path.cwd()
<span># 从 .env 读取模型名，全脚本统一用这一个常量</span>
MODEL = os.getenv(<span>"DEEPSEEK_MODEL"</span>)

client = OpenAI(
    base_url=os.getenv(<span>"DEEPSEEK_BASE_URL"</span>),
    api_key=os.getenv(<span>"DEEPSEEK_API_KEY"</span>),
)

代码注释里有句话很关键：

"Agent 工作目录，安全的，被授权的。"

WORKDIR = Path.cwd() —— 这不是随便拿个路径，这是划定"办公室"的范围。后面所有文件操作都不能迈出这个门。

还有个小知识点，代码注释特意提了：

"python 没有常量变量之分，都是变量，用约定大写来表达"

Python 没有 const，全靠全大写命名约定表示"这是常量，别改"。

Environment 模块：门禁系统

<span>def</span> <span>safe_path</span>(<span>p: <span>str</span></span>) -> Path:
    <span># pathlib.Path 特有的 / 运算符，不是除法，运算符的重载</span>
    <span># 相当于路径的拼接</span>
    path = (WORKDIR / p).resolve()
    <span># 逻辑判断语法</span>
    <span>if</span> <span>not</span> path.is_relative_to(WORKDIR):
        <span># 抛出异常</span>
        <span>raise</span> ValueError(<span>f"Path escapes workspace:<span>{p}</span>"</span>)
    <span>return</span> path

十行代码，干掉了整个"逃出工作目录"的攻击面。两个细节值得说：

第一，WORKDIR / p 是运算符重载。 注释专门解释了：

"pathlib.Path 特有的 / 运算符，不是除法，运算符的重载。"

Python 里 / 本来是除法，但 Path 对象重载了它——变成路径拼接。所以 WORKDIR / "src/app.py" 得到 /Users/xx/project/src/app.py。

第二，.resolve() 之后再校验。 为什么要 resolve？因为 ../../etc/passwd 这种路径，不 resolve 你根本看不出它跑哪去了。resolve 把 .. 全部展开成绝对路径，然后才判断 is_relative_to(WORKDIR)。

如果 LLM 试图访问 ../../../etc/passwd（经典路径穿越攻击），第一次校验就会拦住——抛 ValueError 而不是"默默允许"。

这就是 readme 说的"工作区隔离"。

Policy & Safety 模块：危险命令黑名单

<span>def</span> <span>run_bash</span>(<span>command: <span>str</span></span>) -> <span>str</span>:
    dangerous = [<span>"rm -rf /"</span>, <span>"sudo"</span>, <span>"shutdown"</span>, <span>"reboot"</span>, <span>">/dev/"</span>]
    <span># any() 就是只有一个满足就返回真</span>
    <span>if</span> <span>any</span>(d <span>in</span> command <span>for</span> d <span>in</span> dangerous):
        <span>return</span> <span>"Error: Dangerous command blocked."</span>

    <span>try</span>:
        r = subprocess.run(command, shell=<span>True</span>, cwd=WORKDIR,
            capture_output=<span>True</span>, text=<span>True</span>, errors=<span>"replace"</span>, timeout=<span>120</span>,
        )
        out = (r.stdout + r.stderr).strip()
        <span>return</span> out[:<span>50000</span>] <span>if</span> out <span>else</span> <span>"(no output)"</span>
    <span>except</span> subprocess.TimeoutExpired:
        <span>return</span> <span>"Error: Timeout(120s)"</span>
    <span>except</span> (FileNotFoundError, OSError) <span>as</span> e:
        <span>return</span> <span>f"Error: <span>{e}</span>"</span>

这是"Policy & Safety"模块的极简落地版——一份危险命令黑名单：

危险命令	拦它的原因
`rm -rf /`	删库跑路一条龙
`sudo`	提权，超出授权范围
`shutdown` / `reboot`	把服务器关了
`>/dev/`	直接写设备文件

注意这里的黑名单策略是"字符串包含判断"（d in command）——非常朴素，甚至可以说很脆弱（稍微变形就能绕过）。但它体现了这个模块的定位：Policy & Safety 不是随便写写，是需要认真设计的一层。

三个工程细节也在这里：

细节	代码	作用
超时保护	`timeout=120`	命令卡死不至于拖死 Agent
输出合并	`r.stdout + r.stderr`	错误信息也是信息，LLM 需要看到
长度截断	`out[:50000]`	防止一次输出撑爆上下文窗口

那个 out[:50000] 尤其重要——它正是 Context 模块"窗口控制"的具体实现。 如果 ls -R 输出了 20 万字符全塞给 LLM，下一次调用可能直接超限。

Tool 模块：四个工具，四双手

<span>def</span> <span>run_read</span>(<span>path: <span>str</span>, limit: <span>int</span> = <span>None</span></span>) -> <span>str</span>:
    <span>try</span>:
        lines = safe_path(path).read_text(encoding=<span>"utf-8"</span>).splitlines()
        <span>if</span> limit <span>and</span> limit < <span>len</span>(lines):
            lines = lines[:limit] + [<span>f"... <span>{<span>len</span>(lines) - limit}</span> more"</span>]
        <span>return</span> <span>"\n"</span>.join(lines)[:<span>50000</span>]
    <span>except</span> Exception <span>as</span> e:
        <span>return</span> <span>f"Error: <span>{e}</span>"</span>

<span>def</span> <span>run_write</span>(<span>path: <span>str</span>, content: <span>str</span></span>) -> <span>str</span>:
    <span>try</span>:
        fp = safe_path(path)
        fp.parent.mkdir(parents=<span>True</span>, exist_ok=<span>True</span>)
        fp.write_text(content, encoding=<span>"utf-8"</span>)
        <span>return</span> <span>f"Wrote <span>{<span>len</span>(content)}</span> bytes"</span>
    <span>except</span> Exception <span>as</span> e:
        <span>return</span> <span>f"Error: <span>{e}</span>"</span>

<span>def</span> <span>run_edit</span>(<span>path: <span>str</span>, old_text: <span>str</span>, new_text: <span>str</span></span>) -> <span>str</span>:
    <span>try</span>:
        fp = safe_path(path)
        content = fp.read_text(encoding=<span>"utf-8"</span>)
        <span>if</span> old_text <span>not</span> <span>in</span> content:
            <span>return</span> <span>f"Error: Text not found in <span>{path}</span>"</span>
        fp.write_text(content.replace(old_text, new_text, <span>1</span>), encoding=<span>"utf-8"</span>)
        <span>return</span> <span>f"Edited <span>{path}</span>"</span>
    <span>except</span> Exception <span>as</span> e:
        <span>return</span> <span>f"Error: <span>{e}</span>"</span>

三个文件工具，各有讲究：

run_read 的"省略提示"：

lines = lines[:limit] + [<span>f"... <span>{<span>len</span>(lines) - limit}</span> more"</span>]

读文件时如果超了 limit，不是简单截断，而是在末尾加一行提示"还有 N 行没显示"——这样 LLM 知道"内容被截了"，可以选择再读一次，而不是误以为文件就这么长。

run_write 的自动建目录：

fp.parent.mkdir(parents=<span>True</span>, exist_ok=<span>True</span>)

parents=True 自动创建多级父目录，exist_ok=True 目录已存在也不报错。一行代码解决"目录不存在"这个高频报错。

run_edit 的"只替换第一处"：

content.replace(old_text, new_text, <span>1</span>)

第三个参数 1 是关键——str.replace 默认替换全部匹配，传 1 就只替换第一个。为什么？因为 LLM 想改的是"某一段"，如果这段文本在文件里出现多次，全替换可能误伤。"找不到就报错，不猜" 也是这个思路：

<span>if</span> old_text <span>not</span> <span>in</span> content:
    <span>return</span> <span>f"Error: Text not found in <span>{path}</span>"</span>

注意这三个函数全部用 try/except 兜底，出错时返回字符串而不是抛异常。这是 Harness 的常见做法——错误信息本身就是要喂给 LLM 的上下文。告诉它"文件不存在"，它就知道该先创建文件。

工具声明：给 LLM 看的说明书

CHILD_TOOLS = [
    {
        <span>"type"</span>: <span>"function"</span>,
        <span>"function"</span>: {
            <span>"name"</span>: <span>"bash"</span>,
            <span>"description"</span>: <span>"Run a shell command."</span>,
            <span>"parameters"</span>: {
                <span>"type"</span>: <span>"object"</span>,
                <span>"properties"</span>: {
                    <span>"command"</span>: {<span>"type"</span>: <span>"string"</span>}
                },
                <span>"required"</span>: [<span>"command"</span>]
            }
        }
    },
    <span># read_file / write_file / edit_file 省略</span>
]

PARENT_TOOLS = CHILD_TOOLS + [
    {
        <span>"type"</span>: <span>"function"</span>,
        <span>"function"</span>: {
            <span>"name"</span>: <span>"task"</span>,
            <span># 给子Agent 分配完全独立的上下文</span>
            <span>"description"</span>: <span>"Spawn a subagent with fresh context. It shares the filesystem but not conversation history."</span>,
            <span>"parameters"</span>: {
                <span>"type"</span>: <span>"object"</span>,
                <span>"properties"</span>: {
                    <span>"description"</span>: {<span>"type"</span>: <span>"string"</span>, <span>"description"</span>: <span>"Short description of the task"</span>},
                    <span>"prompt"</span>: {<span>"type"</span>: <span>"string"</span>, <span>"description"</span>: <span>"The task for the subagent to complete"</span>}
                },
                <span>"required"</span>: [<span>"prompt"</span>]
            }
        }
    }
]

这是今天最精妙的一处设计。

<span>CHILD_TOOLS</span>  = [bash, read_file, write_file, edit_file]        ← 子 Agent 的工具
<span>PARENT_TOOLS</span> = CHILD_TOOLS + [task]                            ← 主 Agent 多一个 task

子 Agent 有四个文件操作工具，但没有 task——它不能再往下派发子 Agent，否则会无限套娃。

主 Agent 则多一个 task 工具，description 写得极具信息量：

"Spawn a subagent with fresh context. It shares the filesystem but not conversation history."

"shares the filesystem but not conversation history"——共享文件系统，但不共享对话历史。一句话讲清了 Sub Agent 的本质：

共享什么	不共享什么
文件系统（都在同一个 WORKDIR）	对话历史（独立上下文）

这正是 readme 说的"上下文互不干扰"，也是"主 Agent 上下文不受拖累"的实现方式。

工具分发：一本花名册

TOOL_HANDLERS = {
    <span>"bash"</span>: <span>lambda</span> **kw: run_bash(kw[<span>"command"</span>]),
    <span>"read_file"</span>: <span>lambda</span> **kw: run_read(kw[<span>"path"</span>], kw.get(<span>"limit"</span>)),
    <span>"write_file"</span>: <span>lambda</span> **kw: run_write(kw[<span>"path"</span>], kw[<span>"content"</span>]),
    <span>"edit_file"</span>: <span>lambda</span> **kw: run_edit(kw[<span>"path"</span>], kw[<span>"old_text"</span>], kw[<span>"new_text"</span>])
}

字典映射，工具名 → 执行函数。注释还贴心地把 Python 的 **kw 跟 JS 对比了一下：

"**kw js ...kw rest 运算符"

**kw 把 LLM 返回的参数字典展开成关键字参数——{"command": "ls"} 变成 run_bash(command="ls")。跟 JS 的展开运算符 ...kw 是一个思路。

注意 read_file 用的是 kw.get("limit")——因为 limit 是可选参数（Schema 里 required 只有 path）。必填用 kw["x"]，选填用 kw.get("x")，这个区分很重要。

子 Agent：独立的小办公室

<span>def</span> <span>run_subagent</span>(<span>prompt: <span>str</span></span>) -> <span>str</span>:
    sub_messages = [{<span>"role"</span>: <span>"user"</span>, <span>"content"</span>: prompt}]
    <span># 最多尝试30次</span>
    <span># 独立的Agentic Loop</span>
    <span>for</span> _ <span>in</span> <span>range</span>(<span>30</span>):
        response = client.chat.completions.create(
            model=MODEL,
            <span># 子Agent 自己的历史必须带上，否则 prompt 根本没发给模型</span>
            messages=[{<span>"role"</span>: <span>"system"</span>, <span>"content"</span>: SUB_SYSTEM}] + sub_messages,
            tools=CHILD_TOOLS,
            max_tokens=<span>8000</span>
        )
        msg = response.choices[<span>0</span>].message
        sub_messages.append(msg.model_dump())

        <span># OpenAI 协议里这个值是复数 tool_calls</span>
        <span>if</span> response.choices[<span>0</span>].finish_reason != <span>"tool_calls"</span>:
            <span>break</span>

        results = []
        <span>for</span> tool_call <span>in</span> msg.tool_calls:
            func_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            handler = TOOL_HANDLERS.get(func_name)
            output = handler(**args) <span>if</span> handler <span>else</span> <span>f"Unknown tool <span>{func_name}</span>"</span>
            results.append({
                <span>"role"</span>: <span>"tool"</span>,
                <span>"tool_call_id"</span>: tool_call.<span>id</span>,
                <span>"content"</span>: <span>str</span>(output)[:<span>50000</span>]
            })
        sub_messages.extend(results)
    <span>return</span> msg.content <span>or</span> <span>"(no summary)"</span>

这段是 Sub Agent 的核心——注意它跟主循环长得几乎一样，因为它本身就是一个完整的、独立的 Agentic Loop。

几个关键设计：

1. 独立的上下文（fresh context）

sub_messages = [{<span>"role"</span>: <span>"user"</span>, <span>"content"</span>: prompt}]

子 Agent 的对话历史从零开始——只有 main agent 传进来的那个 prompt。它看不到主 Agent 之前聊了什么，也不知道自己是被谁派来的。

这就是"业务隔离"——子 Agent 的上下文里没有主 Agent 的历史包袱。

代码注释还强调了一句：

"子 Agent 自己的历史必须带上，否则 prompt 根本没发给模型"

因为 LLM 是无状态的，sub_messages 必须在每轮请求里完整带上，否则子 Agent 会"失忆"。

2. 30 次上限

<span>for</span> _ <span>in</span> <span>range</span>(<span>30</span>):

注释说：

"最多尝试 30 次 / 下标我不用，占位置"

for _ in range(30) 里的下划线是 Python 惯例——"这个变量我不关心"。这 30 次是循环安全阀：防止子 Agent 陷入无限工具调用。

3. 终止条件

<span>if</span> response.choices[<span>0</span>].finish_reason != <span>"tool_calls"</span>:
    <span>break</span>

这就是 readme 说的"判断任务终止条件"——只要模型不再要求调工具，就说明它给出最终答案了，退出循环。

注释还提了个细节：

"OpenAI 协议里这个值是复数 tool_calls"

tool_calls 是复数——因为一次响应里 LLM 可能同时要求调多个工具。

4. 只返回结论

<span>return</span> msg.content <span>or</span> <span>"(no summary)"</span>

子 Agent 干了 30 轮活儿，最后只返回一段文本给主 Agent。中间那些工具调用细节、文件内容、报错信息——全部留在子 Agent 自己的上下文里，不给主 Agent 添乱。

or "(no summary)" 是兜底——万一模型返回空内容，也得给主 Agent 一个可读的结果，不然主 Agent 会困惑。

readme 对 Sub Agent 的描述在这里得到了完整验证：

"子 Agent 返回结果，有全新的上下文，不会干扰主 Agent 上下文。"

主循环：Agent 的心脏

<span>def</span> <span>agent_loop</span>(<span>messages: <span>list</span></span>):
    <span>while</span> <span>True</span>:
        response = client.chat.completions.create(
            model=MODEL,
            messages=[{<span>"role"</span>: <span>"system"</span>, <span>"content"</span>: SYSTEM}] + messages,
            tools=PARENT_TOOLS,
            max_tokens=<span>8000</span>
        )
        msg = response.choices[<span>0</span>].message
        <span>print</span>(msg.content, <span>"??"</span>)
        messages.append(msg.model_dump())
        <span>if</span> response.choices[<span>0</span>].finish_reason != <span>"tool_calls"</span>:
            <span>return</span>
        results = []
        msg = response.choices[<span>0</span>].message
        <span>if</span> msg.tool_calls:
            results = []
            <span>for</span> tool_call <span>in</span> msg.tool_calls:
                func = tool_call.function
                args = json.loads(func.arguments)
                <span># 主Agent分任务</span>
                <span>if</span> func.name == <span>"task"</span>:
                    desc = args.get(<span>"description"</span>, <span>"subtask"</span>)
                    prompt = args.get(<span>"prompt"</span>, <span>""</span>)
                    <span>print</span>(<span>f"> task(<span>{desc}</span>): <span>{prompt[:<span>80</span>]}</span>"</span>)
                    <span># 启动子Agent</span>
                    output = run_subagent(prompt)
                <span>else</span>:
                    <span># 主Agent 也可以自己干活</span>
                    handler = TOOL_HANDLERS.get(func.name)
                    output = handler(**args) <span>if</span> handler <span>else</span> <span>f"Unknown tool <span>{func.name}</span>"</span>

                results.append({
                    <span>"role"</span>: <span>"tool"</span>,
                    <span>"tool_call_id"</span>: tool_call.<span>id</span>,
                    <span>"content"</span>: <span>str</span>(output)
                })
        messages.extend(results)

这就是 Loop 模块——readme 说的"核心控制层，调用大模型，分发工具调用，判断任务终止条件"，逐条对上：

readme 描述	代码实现
调用大模型	`client.chat.completions.create(...)`
分发工具调用	`for tool_call in msg.tool_calls:`
判断任务终止条件	`if finish_reason != "tool_calls": return`
自主长时间干活	`while True:`

核心分支在 if func.name == "task"：

<span>if</span> func.name == <span>"task"</span>:
    ...
    output = run_subagent(prompt)      <span># 派活给子 Agent</span>
<span>else</span>:
    handler = TOOL_HANDLERS.get(func.name)
    output = handler(**args)           <span># 自己干</span>

注释写得明白：

"主 Agent 分任务" / "启动子 Agent" / "只关注结果" / "主 Agent 也可以自己干活"

主 Agent 有两副面孔——简单活儿自己干（读个文件、跑个命令），复杂活儿派给子 Agent。这个"分派 or 自办"的决策完全由 LLM 自己根据任务性质决定，代码不干预。

这就是 Agent 和 Workflow 的分野——Workflow 里"谁来干"是写死的，Agent 里是 LLM 现场决定的。

两个 System Prompt：角色隔离

<span># 主Agent 系统提示</span>
SYSTEM = (
    <span>f"You are a coding agent at <span>{WORKDIR}</span>."</span>
    <span>"Use task for focused exploration or a self-contained subtask."</span>
)

SUB_SYSTEM = (
    <span>f"You are a coding agent at <span>{WORKDIR}</span>."</span>
    <span>"Complete the given task, then return a concise final answer."</span>
)

readme 说的"专属 SYSTEM prompt"，就是这两行：

Prompt	角色定位	关键指令
SYSTEM	主 Agent（规划、分工）	"Use task for focused exploration..."
SUB\_SYSTEM	子 Agent（执行任务）	"Complete the given task, then return a concise final answer."

三处细节：

1. 用 f-string 把 WORKDIR 嵌进去

两个 prompt 都以 f"You are a coding agent at {WORKDIR}." 开头——告诉 LLM 自己在哪个目录。 这样它调 bash 和读写文件时不会迷路。

2. 主 Agent 被明确引导使用 task

主 SYSTEM 里那句 "Use task for focused exploration or a self-contained subtask." 是在主动引导 LLM 使用子 Agent——不写这句，模型可能什么都自己干，Sub Agent 就白设计了。

3. 子 Agent 被要求"简洁收尾"

"return a concise final answer"——子 Agent 干了多少活都行，但给主 Agent 的回复必须简洁。这是上下文管理的最后一道防线：子 Agent 自己的上下文可以很脏，但传给主 Agent 的必须是干净结论。

还有个 Python 语法彩蛋，注释特意标了：

"python 隐式字符串拼接，括号里连续放多个字符串变量"

f"You are..." "Use task..." 两个字符串写在一起，Python 会自动拼接成一个——不需要 + 号。这就是为什么 f-string 只加在第一个上，第二个却能共享 WORKDIR 插值的上下文。

交互界面：一间能进的办公室

<span>if</span> __name__ == <span>"__main__"</span>:
    <span>print</span>(<span>"Subagent - fresh messages, final text returns"</span>)
    <span>print</span>(<span>"Enter a question, press Enter to sent. Type q to quit.\n"</span>)
    history = []
    <span>while</span> <span>True</span>:
        <span>try</span>:
            query = <span>input</span>(<span>"\001\033[36m\002s06 >> \001\033[0m\002"</span>)
        <span>except</span> (EOFError, KeyboardInterrupt):
            <span>break</span>
        <span>print</span>(query)
        <span>if</span> query.strip().lower() <span>in</span> (<span>"q"</span>, <span>"exit"</span>, <span>""</span>):
            <span>break</span>
        history.append({<span>"role"</span>: <span>"user"</span>, <span>"content"</span>: query})
        agent_loop(history)

一个极简的交互式 CLI：

history 累积对话（这就是最基础的 Memory 模块——跨轮次保持在内存里）
input() 读用户输入
q / exit / 空行退出
EOFError / KeyboardInterrupt 优雅退出

那串 "\001\033[36m\002s06 >> \001\033[0m\002" 是 ANSI 转义序列——把提示符染成青色（\033[36m），\001 / \002 是 readline 库需要的边界标记（不加会导致光标位置计算错乱）。一个彩色提示符，藏着两个知识点。

注意 agent_loop(history) 是在主循环内部调用的——每轮用户输入都调一次，而 history 一直在累积。这就是"多轮对话"的实现方式：Agent 的 Loop 是一层，用户交互的 Loop 是外面那层。

七、一张图收尾：六大模块在代码里的落点

agent<span>.py</span>（<span>338</span>行）
│
├── Environment 沙箱 ────────── <span>safe_path</span>() + WORKDIR
│                              路径校验、工作区隔离
│
├── Policy & Safety ─────────── dangerous 黑名单
│                              rm -rf / sudo shutdown 拦截
│
├── Tool 工具集 ─────────────── run_bash / run_read / run_write / run_edit
│                              + CHILD_TOOLS / PARENT_TOOLS 声明
│
├── Loop 主循环 ─────────────── <span>agent_loop</span>()  while True
│                              ReAct：调用→分发→判断终止
│                              <span>run_subagent</span>()  子 Agent 独立循环（<span>30</span>次上限）
│
├── Context 上下文管理 ───────── messages + SYSTEM + SUB_SYSTEM
│                              out<span>[:50000]</span> 窗口控制
│                              子 Agent 只 return msg<span>.content</span>（压缩）
│
├── Memory 记忆层 ───────────── history 列表（跨轮次）
│                              子 Agent 的 sub_messages（独立上下文）
│
└── Observability ───────────── <span>print</span>(msg.content) / <span>print</span>(f"> task(...)")
                                （最简版：日志输出）

六个模块，一个不少，全在这 338 行里。 当然这是教学版本的极简实现——真正的生产级 Harness（比如你用的 Claude Code、Codex）在这六个方向上都有几十倍的工程量，但骨架是同一个。

八、给未来的自己：这份代码教会了什么

最后总结三层收获：

第一层：架构视角

Agent 不是"调用 LLM 的代码"，而是六个模块协同的系统。Loop 是心脏、Tool 是手、Context 是记忆的调度台、Environment 是围栏、Memory 是笔记本、Observability 是监控。缺一个，"可靠、持续"就打折扣。

第二层：工程取舍

代码里处处是取舍——黑名单拦危险命令（简单但脆弱）、输出截断 5 万字符（防爆窗口但可能丢信息）、子 Agent 上限 30 轮（防死循环但可能不够用）。没有一个方案是完美的，都是权衡的结果。 这才是"Engineering"这个词的分量。

第三层：职业方向

readme 开头的两个岗位——"AI 应用开发工程师（Agentic RAG 应用）"、"AI Agent 开发工程师（Harness Agent FDE）"。学完这节课你会发现，这两个岗位的核心能力，不是"会调 API"，而是能设计出上面这套系统。

PS：这节课学完，再看任何 Agent 产品，你都能拆成六个问题问它——你的主循环怎么设计的？工具集有哪些？上下文怎么压缩？沙箱怎么隔离？记忆存在哪？出问题怎么查？能答上来的，是认真做的产品；答不上来的，大概就是"套了个壳的聊天框"。