背景概念

今天的 Agent，本质上正在收敛为"LLM + Harness"的闭环结构。

过去我们搭 Agent，习惯用编排框架把模型能力嵌进一条固定流程的某个节点——说白了，是把模型当成一个"输入指令、输出给定答案"的函数来调用。这套做法有它的历史必然性：彼时的模型，还撑不起自主循环、自我纠错的长链路工作。

但模型能力的演进速度，已经跑在了工程范式的前面。当模型自己就能走完感知、规划、调工具、反思修正这一整套闭环时，那些重框架、硬编排的脚手架，反而可能变成枷锁——它会把一个本可泛化的通用智能，重新框进我们为旧时代画好的流程里。

真正的问题，正在从"如何编排模型"，转向"如何最小化地托举模型"。

第一个把这条路走通、并真正定义了新范式的产品，是 Claude Code。

在它之前，AI 编程工具的默认形态是"寄生在 IDE 里的插件"——模型被框在编辑器侧边栏里，只能补全、只能问答、只能在人划好的格子里行动。Claude Code 反其道而行：它直接活在终端里，自己读项目、自己改文件、自己跑命令、自己看报错——你给它一个目标，它在循环里自己把活干完。

它证明了一件事：Agent 不需要被塞进 IDE，也不需要被编排框架拎着走。它只需要一个最小的 Harness、一组能碰真实世界的工具，然后在"思考 → 行动 → 观察"的循环里自己跑。

参考项目：PI-Agent** 整个的项目还是蛮有结构的，然后分模块去写的。（里面还有一个Agent-coding专门为coding准备的，学习ing）**

参考项目：Nami-Agent** 从0手写Agent**

Agent中Harness基础必备模块

1. 真实必备基础工具（读文件、写文件、执行命令行）

大模型只能输出文字，但通过 Harness 的工具系统，它能间接地操作真实世界：

工具类型	举例	作用
文件操作	read、write、edit	读写代码文件
命令执行	shell/bash	运行终端命令，如编译代码、安装依赖
搜索工具	grep、find、web_search	在代码中搜索、在网上查资料
浏览器	截图、点击、填表	测试网页应用
外部 API	数据库查询、发送邮件	与外部服务交互

工具调用的本质是什么？

其实就是模型在回复中输出一段特殊格式的文本，Harness 识别后去执行对应的函数：

▼
text
复制代码
模型输出：shell(command=["cat", "README.md"])
    ↓
Harness 识别到这是工具调用
    ↓
Harness 在本地执行 cat README.md
    ↓
把文件内容作为"工具结果"追加到对话历史中
    ↓
再次调用模型，让它基于文件内容继续思考

Anthropic 在《Building Effective Agents》中强调了一个重要原则：设计工具接口要像设计人机交互（HCI）一样用心。他们甚至造了一个新词叫 ACI（Agent-Computer Interface，智能体-计算机接口）：工具的名字、参数描述、示例用法，都要让模型"一看就懂"。

2. 最小循环

OpenAI 在关于 Codex 的技术博客中详细描述了这个循环。用更直白的话说：

1. 收集信息：用户说了什么？之前做了什么？现在项目是什么状态？把这些信息打包

2. 模型思考：把打包好的信息发给大模型，让它决定下一步该干什么

3. 执行动作：如果模型说"运行这段代码"或"读取这个文件"，Harness 就去执行

4. 观察结果：执行完了把结果反馈给模型，让它继续思考

5. 循环往复：直到模型觉得任务完成了，输出最终结果

这就是所谓的"思考 → 行动 → 观察"循环（Think → Act → Observe Loop），是所有 Agent 系统的核心模式。

3. 权限边界

当你给 Agent 执行 Shell 命令的能力时，一个绕不开的问题就出现了：如果它执行了 rm -rf / 怎么办？

这就是为什么安全系统是 Harness 中不可或缺的一部分。业界通常采用**纵深防御（Defense in Depth）**策略——多层防护、每层独立：

防护层	做什么	比喻
提示词层	在系统指令里明确告诉模型哪些事不能做	给员工发员工手册
工具层	限制不同场景下可用的工具（如"规划模式"只能读、不能写）	不同岗位给不同权限的钥匙
审批层	对危险操作弹窗确认，用户点同意才执行	银行转账要二次确认
校验层	自动检测危险命令模式（如 rm -rf、sudo），即使模型请求也拒绝执行	海关安检
钩子层	用户自定义脚本，在工具执行前后插入检查	安全审计

以 OpenAI 的 Codex 为例，它采用了沙盒机制：对内置 shell 等受控工具有沙盒和审批，Agent 只能在指定目录内读写、只能访问指定网络资源，超出范围的操作必须用户明确授权。需要注意的是，其他工具（尤其 MCP 提供的工具）不受这层沙盒直接约束，需要各自实现自己的防护。

4. 上下文管理

上下文管理是 Harness 最复杂也最关键的部分。它要解决的核心问题是：如何在有限的上下文窗口里，装下最有价值的信息？

提示词的构成

每次调用大模型时，Harness 需要组装一个完整的提示词。以 Codex 为例，提示词的结构大致是这样的：

层级	内容	来源
系统指令	"你是一个编程助手，请遵守以下规则……"	Harness 内置
工具描述	可用工具的名称、参数格式、使用说明	工具注册表
沙盒权限	"你可以读写 /project 目录下的文件"	安全配置
用户指令	用户自定义的规则（如 AGENTS.md 文件）	项目配置
环境信息	当前工作目录、操作系统等	自动检测
对话历史	之前所有的对话、工具调用和结果	会话记录
当前输入	用户最新的请求	用户输入

对话历史会随着每一轮交互不断增长，这就引出了两个关键机制：上下文压缩和提示词缓存。

上下文压缩（Compaction）

当对话太长、快要撑爆上下文窗口时，Harness 会对历史信息进行压缩：

• 软裁剪：把较早的工具返回结果截断，只保留前面和后面一小段

• 硬清除：直接把很久以前的工具结果替换成 [结果已清除]

• 摘要压缩：用模型把之前的对话总结成一段精炼的摘要，替代原始内容

Anthropic 把这个过程叫做 Compaction，OpenAI 的 Codex 也实现了类似的自动压缩机制。

一个形象的比喻：上下文窗口就像你的书桌。桌面空间有限，你不可能把所有文件都摊在上面。聪明的做法是：把最近在用的放在桌上，不太常用的收进抽屉（压缩），很久不用的归档到柜子里（清除）。

提示词缓存（Prompt Caching）

每次调用模型都要发送完整的对话历史，而随着对话增长，这个成本会越来越高。但如果前面的内容没变，只是末尾追加了新的内容，模型服务商可以缓存之前已处理的部分，只计算新增的内容。

这就是为什么 Codex 的 Harness 在设计上刻意保证：新的提示词总是旧提示词的精确前缀。就像写作文时，每次都在上一次的基础上往后面加内容，而不是把前面的内容改来改去——这样服务端就能复用之前的计算。

简单的Agent循环-Python

第一步：创建一个大模型的Client

万事开头先建 Client。这一步做的事很简单：引入 OpenAI SDK，填好 api_key 和 base_url，拿到一个能发请求的客户端对象。

为什么先做这个？因为后面不管是聊天、调工具、还是压缩历史，所有对模型的请求都要从这个客户端发出去——它是整个 Harness 对外的唯一通道。代码里后面那几个被注释掉的 model、messages、tools、reasoning_effort，就是接下来要逐个填进来的参数。

▼
Python
复制代码
from openai import OpenAI

def make_client():
"""创建一个API客户端"""
    return OpenAI(
        *api_key*="magpie",
        *base_url*="http://127.0.0.1:3425"
    )
    
# 调用构建Client函数
client = make_client()

# model = ....

# messages = .....

# 调用AI服务得到答案
resp = *client*.chat.completions.create
(
    *model*=*model*,                 # 模型名称
    *messages*=messages,           # 消息上下文
    *tools*=get_schemas()          # 调用的工具集合
    reasoning_effort="high",     # 推理强度：服务支持时使用
    # temperature=0.3,           # 模型温度、服务和模型支持时再启用
    # stream = True              # 默认是False，不开启流式输出
)

第二步：构建上下文管理-messages

看上面的调用，一次完整请求需要三样东西：model（用哪个模型）、tools（能干什么）、messages（聊到哪了）。model 是个字符串，tools 是工具注册表自动生成的 schema，最复杂的是 messages——它不是一次性写死的，而是随着每一轮对话不断增长的会话状态。

这一步我们把 messages 从一个裸数组，封装成一个 Session 对象：它负责组装系统提示词、持久化历史、跟踪 token 用量，后面还要负责在上下文快撑爆时做压缩。

▼
Python
复制代码
class Session:
    """一个会话持有自己的消息、权限管理器和上一次的 token 数。"""

    def __init__(*self*, *name*, *ask*=None, *persist*=True):
        *self*.name = *name*
        *self*.persist = *persist*
        *self*.messages = (load_session(*name*) if *persist* else None) or []
        # 系统消息
        system_message = {"role": "system", "content": build_system()}
        # 第一条已经是 system 消息：直接用新的 system_message 替换旧的，避免出现两条 system 消息。
        if *self*.messages and *self*.messages[0].get("role") == "system":
            *self*.messages[0] = system_message
        # 消息为空，或者第一条不是 system 消息：在列表开头插入新的 system 消息，原有消息向后移动，不会丢失。
        else:
            *self*.messages.insert(0, system_message)
        *self*.perms = PermissionManager(*ask*=*ask*)
        # 它是一个计数状态，记录**上一次模型请求使用了多少输入 token**。初始化为 0 
        # 表示当前还没有记录到上一次请求的统计数据，**不代表会话里的实际内容是空的或只有 0 个 token**。
        *self*.prompt_tokens = 0

    def save(*self*):
        if *self*.persist:
            save_session(*self*.name, *self*.messages)

BASE_SYSTEM = ("你是一个能使用工具的助手，在用户当前目录下完成任务。"
               "必要时调用工具，完成后用简洁的语言汇报结果。"
               f"当前操作系统：{platform.system() or '未知'}；run_shell 默认使用的 shell：{shell}。"
               "调用 run_shell 前，根据操作系统和 shell 选择兼容的命令语法。"
               "直接执行命令时，不要混用 CMD、PowerShell 和 Bash 的语法。")    
               
def build_system():
    return BASE_SYSTEM + skills.project_notes() + skills.skills_section()
    

        

▼
Python
复制代码
# context.py
import json
import os
from pathlib import Path

MAX_TOOL_CHARS = 4000          # 单个工具结果最多保留的字符数
KEEP_RECENT_TOOL_RESULTS = 3   # 清理时，保留最近几条工具结果
KEEP_LAST_TURNS = 2            # 压缩时，原样保留最近几轮用户对话

# 程序自己的状态（会话记录、后面的定时任务）放在工作区之外，Agent 的文件工具碰不到
STATE_DIR = Path(os.environ.get("NAMI_STATE_DIR", Path.home() / ".nami-agent"))
SESSION_DIR = STATE_DIR / "sessions"

PLACEHOLDER = "[旧工具结果已清理，如需要请重新调用工具]"
KEEP_PREFIXES = ("[skill]",)

SUMMARY_PROMPT = """你是对话摘要器。把下面这段 Agent 的历史对话压缩成摘要，要求：
1. 保留：用户的目标、已完成的步骤、关键结论、重要的文件路径和参数、尚未解决的问题
2. 丢弃：寒暄、重复内容、工具返回的大段原文
3. 用简洁的条目写，不要超过 500 字
只输出摘要本身。"""

def truncate_tool_result(*text*, *limit*=MAX_TOOL_CHARS):
    """截断工具结果，保留前后各一半，中间省略"""
    if len(*text*) <= *limit*:
        return *text*
    half = *limit* // 2 # 除于2且得到整数
    return f"{*text*[:half]}\n...[中间省略 {len(*text*) - *limit*} 字符]...\n{*text*[-half:]}"

def clear_old_tool_results(*messages*, *keep*=KEEP_RECENT_TOOL_RESULTS):
    """清理旧的工具结果，保留最近几条"""
    tool_idx = [
        i for i, m in enumerate(*messages*)
        if m["role"] == "tool" and not (m.get("content") or "").startswith(KEEP_PREFIXES)
    ]
    old = tool_idx[:-*keep*] if *keep* > 0 else tool_idx
    for i in old:
        *messages*[i]["content"] = PLACEHOLDER

def split_for_compact(*messages*, *keep_turns*=KEEP_LAST_TURNS):
    """返回 (要被压缩的旧消息, 原样保留的新消息)；没有可压缩的内容返回 None。
    约定 messages[0] 是 system。"""
    user_idx = [i for i, m in enumerate(*messages*) if m["role"] == "user"]
    if len(user_idx) <= *keep_turns*:
        return None
    cut = user_idx[-*keep_turns*] # 找到保留的最早用户消息的索引
    return *messages*[1:cut], *messages*[cut:] # 返回旧消息和新消息

def render_for_summary(*msgs*):
    """把消息渲染成文本，方便摘要器理解"""
    lines = []
    for m in *msgs*:
        if m["role"] == "assistant" and m.get("tool_calls"):
            calls = ", ".join(
                f'{c["function"]["name"]}({c["function"]["arguments"]})'
                for c in m["tool_calls"]
            )
            if m.get("content"):
                lines.append(f'[assistant] {m["content"]}')
            lines.append(f"[assistant 调用工具] {calls}")
        else:
            lines.append(f'[{m["role"]}] {m.get("content") or ""}')
    return "\n".join(lines)

def compact(*client*, *model*, *messages*):
    """压缩对话历史，保留最近几轮用户对话，旧的对话压缩成摘要"""
    parts = split_for_compact(*messages*)
    if parts is None:
        return *messages*
    old, recent = parts
    resp = *client*.chat.completions.create(
        *model*=*model*,
        *messages*=[
            {"role": "system", "content": SUMMARY_PROMPT},
            {"role": "user", "content": render_for_summary(old)},
        ],
    )
    summary = resp.choices[0].message.content
    if not isinstance(summary, str) or not summary.strip():
        raise ValueError("摘要模型没有返回有效内容")
    return [
        *messages*[0],
        {"role": "user", "content": f"[以下是此前对话的摘要]\n{summary}"},
        {"role": "assistant", "content": "好的，我已了解之前的进展，继续。"},
        *recent,
    ]

def save_session(*name*, *messages*):
    """保存会话到本地"""
    SESSION_DIR.mkdir(*parents*=True, *exist_ok*=True)
    (SESSION_DIR / f"{*name*}.json").write_text(
        json.dumps(*messages*, *ensure_ascii*=False, *indent*=2), *encoding*="utf-8")

def load_session(*name*):
    """从本地加载会话"""
    path = SESSION_DIR / f"{*name*}.json"
    if not path.exists():
        return None
    return json.loads(path.read_text(*encoding*="utf-8"))

看 Session 的构造函数，初始化一个会话其实只做了三件事：

1. 加载历史消息。 传入 name 作为会话的唯一标识。如果 persist=True（默认），就从本地 ~/.nami-agent/sessions/ 目录加载对应的历史记录；文件不存在就返回空列表，等于开新会话。ask 是权限确认时的回调，这里先按下不表。

2. 组装系统提示词。 调用 build_system() 拼出 system 消息，由三部分叠加：基础指令（"你是一个能使用工具的助手……"）+ 项目级说明（project_notes()）+ 已注册 Skill 的说明（skills_section()）。

3. 把 system 消息放到列表首位。 如果历史里已经有 system 消息就直接替换，没有就插到最前面——保证 messages[0] 永远是最新的系统提示词。之后每轮用户消息追加到列表末尾，初始化就算完成。

还有一个容易忽略的设计点：工具（tools）并不存进 messages 里。它是每次发请求时，从工具注册表实时取一份 schema 列表，通过 tools 参数单独传给模型。好处是工具集可以动态变化，而不用动历史消息；更重要的是，这让"新提示词是旧提示词的精确前缀"成为可能——前面讲过的 Prompt Caching 就靠这个。

4. 每轮请求前，检查上下文是否需要压缩。 模型的上下文窗口不是无限的，对话越长，喂进去的 token 越多。所以每轮对话开始前，都要拿"上一次请求用了多少 token"做判断：要不要清理旧工具结果？要不要把更早的对话压缩成摘要？这就是为什么 Session 里要记一个 prompt_tokens——它是下一轮是否触发压缩的依据。

▼
Python
复制代码
def prepare_context(*session*, *client*, *model*):
    """按上一次请求的 token 用量清理或压缩对话。"""
    messages = *session*.messages
    if *session*.prompt_tokens > HARD_LIMIT:
        try:
            compacted = compact(*client*, *model*, messages)
        except Exception as error:
            print(f"[context] 摘要失败（{type(error).__name__}），保留原对话。")
        else:
            messages[:] = compacted
            clear_old_tool_results(messages)
    elif *session*.prompt_tokens > SOFT_LIMIT:
        clear_old_tool_results(messages)

prepare_context() 里设了两条警戒线，从轻到重分别处理：超过软上限，只清理旧工具结果；超过硬上限，先把早期对话压缩成摘要，再清理旧工具结果。

两条警戒线都根据上一次模型请求的输入 token 数判断：超过 16000 时，将较早的普通工具结果替换为占位文字，保留最近 3 条；超过 24000 时，把更早的对话交给模型压缩成摘要（保留最近两轮），再清理旧工具结果。单次工具返回内容的字符裁切是另一套独立机制。需要注意：这里没有在发送前重新计算当前上下文，所以不能保证下一次请求一定不超模型限制。

▼
Python
复制代码
def compact(*client*, *model*, *messages*):
    """压缩对话历史，保留最近几轮用户对话，旧的对话压缩成摘要"""
    parts = split_for_compact(*messages*)
    if parts is None:
        return *messages*
    old, recent = parts
    resp = *client*.chat.completions.create(
        *model*=*model*,
        *messages*=[
            {"role": "system", "content": SUMMARY_PROMPT},
            {"role": "user", "content": render_for_summary(old)},
        ],
    )
    summary = resp.choices[0].message.content
    if not isinstance(summary, str) or not summary.strip():
        raise ValueError("摘要模型没有返回有效内容")
    return [
        *messages*[0],
        {"role": "user", "content": f"[以下是此前对话的摘要]\n{summary}"},
        {"role": "assistant", "content": "好的，我已了解之前的进展，继续。"},
        *recent,
    ]

▼
Python
复制代码
def clear_old_tool_results(*messages*, *keep*=KEEP_RECENT_TOOL_RESULTS):
    """清理旧的工具结果，保留最近几条"""
    tool_idx = [
        i for i, m in enumerate(*messages*)
        if m["role"] == "tool" and not (m.get("content") or "").startswith(KEEP_PREFIXES)
    ]
    old = tool_idx[:-*keep*] if *keep* > 0 else tool_idx
    for i in old:
        *messages*[i]["content"] = PLACEHOLDER

第三步：构建工具体系管理

之前提过，每次发请求时要把工具列表通过 tools 参数传给模型。那这个列表从哪来？

本质上，每个工具要告诉模型的就三件事：工具叫什么、能干什么、参数怎么填。如果手动维护这张表，每加一个工具就要改三处，很容易漏。所以用 Python 装饰器来做自动注册——写函数的时候顺手标一下，工具就自动进表了：

▼
Python
复制代码
def tool(*name*, *description*, *parameters*):
    """装饰器：把一个函数注册为工具"""
    def decorator(*func*):
        """注册工具到 TOOLS 中，并返回原函数本身"""
        TOOLS[*name*] = {
            "schema": {
                "type": "function", 
                "function": {
                    "name": *name*, 
                    "description": *description*, 
                    "parameters": *parameters*
                    }
                },
            "func": *func*,
        }
        return *func*
    return decorator
    
def get_schemas():
    """返回给模型看的全部工具说明书"""
    return [t["schema"] for t in TOOLS.values()]

def _params(***props*):
    """简写：所有参数都是必填的字符串"""
    return {"type": "object",
            "properties": {k: {"type": "string", "description": v} for k, v in *props*.items()},
            "required": list(*props*)}
    
    
@tool("list_dir", "列出某个目录下的文件和子目录。", _params(*path*="目录路径，当前目录写 ."))
def list_dir(*path*="."):
    p = Path(*path*)
    return "\n".join(sorted(x.name + ("/" if x.is_dir() else "") for x in p.iterdir())) or "(空目录)"

这段代码做了两件事：@tool 装饰器把函数注册进全局 TOOLS 字典，get_schemas() 把所有工具的 schema 统一吐出来给模型。到这里，"工具对外暴露"这半边就齐了。

但只有 schema 还不够——模型决定调工具后，会返回一个带工具名和参数的请求，我们还需要一个执行器真正把函数跑起来。这里有两个设计要点：

失败不能让后端崩。 工具调用失败是常态（文件不存在、命令报错、权限不够），不应该抛异常中断整个流程，而要把错误信息转成字符串回给模型——让它自己看到错误、自己决定下一步怎么办。

返回结果要裁切。 一个工具可能返回几 MB 的日志，直接塞进上下文会瞬间撑爆窗口。超过长度上限就截断（load_skill 例外，因为 Skill 说明不能切）。

▼
Python
复制代码
# 工具执行器：根据工具名找到对应的 Python 函数，传入参数执行，再把结果或错误转换成字符串返回。
def execute(*name*, *args*):
    """执行工具，永远返回字符串，永远不抛异常"""

    # 检查工具是否存在
    if *name* not in TOOLS:
        return f"Error: 不存在工具 '{*name*}'，可用工具：{', '.join(TOOLS)}"
    try:
    # 找到函数并执行
        result = str(TOOLS[*name*]["func"](***args*))
    except Exception as e:
    # 返回执行工具的报错文段
        return f"Error: {type(e).__name__}: {e}"
    # 限制过长的结果 load_skill 有例外：如果结果以 [skill] 开头，就不在这里截断，避免把 Skill 的完整说明切掉。
    if len(result) > MAX_OUTPUT and not (*name* == "load_skill" and result.startswith(KEEP_PREFIXES)):
        result = result[:MAX_OUTPUT] + f"\n...（已截断，共 {len(result)} 字符）"
    return result

到这里，一次"发请求"的闭环就通了。接下来是怎么处理模型返回：

1. 把模型的回复追加进 messages。 不管是文字回复还是工具调用请求，都要存进会话历史。

2. 记录这轮用了多少 token。 就是前面讲压缩时提到的 prompt_tokens——从响应的 usage 里取出来，供下一轮判断要不要压缩。

▼
Python
复制代码
**response_usage = getattr(resp, "usage", None)**
**prompt_tokens = getattr(response_usage, "prompt_tokens", None)**
# 拿到的 prompt_tokens 是否为非负整数。符合条件就保存原值，否则保存 0
*session*.prompt_tokens = prompt_tokens if isinstance(prompt_tokens, int) and prompt_tokens >= 0 else 0

3. 如果模型一轮里调了多个工具，每个都要执行并回填结果。 模型返回的 tool_calls 是一个数组，遍历执行后，把每个工具结果以 role: "tool" 的消息追加回 messages，模型下一轮就能看到了。

▼
Python
复制代码
msg = resp.choices[0].message
tool_calls = msg.tool_calls

# 模型可能一轮里请求多个工具，每个都必须回应
for call in msg.tool_calls:
    name = call.function.name
    *log*(f"[tool] {name}")
    try:
        result = handle_call(call, *session*.perms)
    except Exception as error:
        result = f"Error: 处理工具调用失败（{type(error).__name__}）"
    if name != "load_skill" or not result.startswith(KEEP_PREFIXES):
        result = truncate_tool_result(result)
    messages.append({
        "role": "tool",
        "tool_call_id": call.id,
        "content": result,
    })

第四步：着手增添工具权限控制

第三步已经让模型能够调用工具，但这也带来了一个问题：模型发起了工具调用，程序就应该无条件执行吗？

例如，读取文件和执行 shell 命令的影响并不一样。文件操作可能访问工作区之外的路径，shell 命令也可能执行危险操作。因此，在真正调用工具函数前，需要先做一次权限检查。

本项目把这部分逻辑放在 PermissionManager 中。工具调用的处理流程变成：

▼
Plain
复制代码
模型返回工具调用
    ↓
解析工具名和参数
    ↓
检查硬规则
    ↓
根据风险等级决定放行、拒绝或询问
    ↓
通过后执行工具，未通过则返回拒绝原因

给每个会话配一个权限管理器

在 Session 初始化时，会创建一个 PermissionManager：

▼
Plain
复制代码
self.perms = PermissionManager(ask=ask)

这里的 perms 是 permissions 的缩写，表示当前会话使用的权限管理器。`ask` 是可选的确认函数：终端可以通过终端询问，Telegram 入口也可以注入自己的询问方式。

当模型发起工具调用时，第三步中的代码会把它传给 handle_call：

▼
Plain
复制代码
result = handle_call(call, session.perms)

这样做还有一个作用：每个 Session 都有独立的权限状态。如果某个工具被设置为“本会话一直允许”，这个选择不会自动影响其他 Session。

需要注意，目前 Session.save() 保存的是对话消息，不保存权限选择。重新创建 Session 后，权限管理器会重新初始化。

第一关：检查硬规则

硬规则是无论风险等级如何都要先检查的限制。例如：

• 文件工具访问的路径必须在工作区内。

• write_file 和 edit_file 不能修改技能目录。

• run_shell 的命令不能命中项目中登记的危险命令模式。

工作区由下面这行决定：

▼
Plain
复制代码
WORKSPACE = Path(os.environ.get("NAMI_WORKSPACE", os.getcwd())).resolve()

启动程序前如果设置了 NAMI_WORKSPACE，就使用指定目录；否则使用启动时的当前工作目录。

检查文件路径时，项目会先将相对路径、.. 和符号链接解析成绝对路径，再判断它是否属于工作区：

▼
Plain
复制代码
def resolve_in_workspace(path_str: str) -> Path | None:
    p = Path(path_str).expanduser()
    if not p.is_absolute():
        p = WORKSPACE / p

    p = p.resolve()
    return p if p.is_relative_to(WORKSPACE) else None

例如，当 read_file、`write_file`、`list_dir` 或 edit_file 请求访问工作区之外的路径时，权限检查会拒绝这次调用。`write_file` 和 edit_file 还会额外检查是否试图修改技能目录。

硬规则先于其他判断执行：即使某个工具被标记为 SAFE，或者用户之前选择过“本会话一直允许”，命中硬规则的调用仍然会被拒绝。

第二关：根据风险等级处理

项目定义了三种风险等级：

▼
Plain
复制代码
class Risk(Enum):
    SAFE = "safe"        # 通过硬规则后直接放行
    CONFIRM = "confirm"  # 需要人工确认
    DENY = "deny"        # 拒绝执行

工具与风险等级的对应关系记录在 TOOL_RISK 中：

▼
Plain
复制代码
TOOL_RISK = {
    "list_dir": Risk.SAFE,
    "read_file": Risk.SAFE,
    "write_file": Risk.SAFE,
    "run_shell": Risk.SAFE,
    "fetch_url": Risk.SAFE,
    "web_search": Risk.SAFE,
    "edit_file": Risk.SAFE,
    "load_skill": Risk.SAFE,
}

没登记的工具默认按照 Risk.CONFIRM 处理。`PermissionManager.check()` 的核心逻辑如下：

▼
Plain
复制代码
def check(self, name: str, args: dict) -> Decision:
    # 第 1 关：先检查硬规则
    reason = self._hard_limits(name, args)
    if reason:
        return Decision(False, reason)

    # 第 2 关：检查风险等级
    risk = TOOL_RISK.get(name, Risk.CONFIRM)
    if risk is Risk.DENY:
        return Decision(False, f"工具 {name} 被策略禁用。")

    if risk is Risk.SAFE or name in self.always_allowed:
        return Decision(True)

    # 第 3 关：需要时询问用户
    return self._confirm(name, args)

Decision 用来表示检查结果：`allowed` 表示是否允许执行，`reason` 用于记录拒绝原因。

这里有一个容易误解的地方：当前代码把 write_file、`edit_file` 和 run_shell 也标记成了 Risk.SAFE。 因此，只要它们通过硬规则，就会直接执行，不会默认弹出人工确认。`SAFE` 在这里表示“按当前配置直接放行”，不代表这些操作没有风险。

第三关：需要时询问用户

当工具的风险等级为 CONFIRM，且当前会话没有记录过允许时，权限管理器会调用 ask(name, args) 询问用户。

确认结果有三种：

• y：允许本次操作。

• n：拒绝本次操作，也可以附带原因。

• a：如果该工具允许记住选择，则在当前会话中一直允许。

项目通过 REMEMBERABLE 指定哪些工具可以记住选择：

▼
Plain
复制代码
REMEMBERABLE = {"write_file"}

因此，当 write_file 被配置为需要确认时，用户可以选择让它在当前会话中一直允许。这个记录按工具名保存，并不是只针对某一个文件；硬规则仍然有效。

默认的 ask 会在终端展示工具调用并等待输入。如果当前环境不是交互式终端，默认会拒绝需要确认的操作。Telegram 入口则可以注入自己的 ask 函数，通过聊天消息询问，并在超时后拒绝。

如果希望写文件、编辑文件和执行命令前都询问用户，可以考虑将 TOOL_RISK 中对应的值改成 Risk.CONFIRM。这属于配置调整，不是项目当前的默认行为。

把权限检查接入工具执行

最后，回到第三步里调用的 handle_call。它负责解析参数、检查权限，然后决定是否调用 execute：

▼
Plain
复制代码
def handle_call(call, perms):
    """解析参数 → 权限检查 → 执行。"""
    name = call.function.name

    try:
        args = json.loads(call.function.arguments or "{}")
    except json.JSONDecodeError:
        return "Error: 参数不是合法的 JSON"

    if not isinstance(args, dict):
        return "Error: 工具参数必须是 JSON 对象"

    decision = perms.check(name, args)
    if not decision.allowed:
        return f"[权限拒绝] {decision.reason}"

    return execute(name, args)

如果检查通过，就交给第三步的 execute 执行工具。如果检查未通过，就把拒绝原因作为工具结果返回；外层 run 会把结果追加到 messages，让模型在下一轮看到。

这套权限检查有边界。 run_shell 的工作目录虽然设置为 WORKSPACE，但这不是操作系统沙箱；命令仍可能访问工作区之外，也可能绕过文件工具对技能目录的限制。现有危险命令匹配只能拦截部分已登记的模式，不能保证拦住所有危险命令。直接调用 execute 也不会经过 handle_call 中的权限检查。

另外，当前 list_dir 的权限检查按工作区解析相对路径，但函数执行时使用进程当前目录；如果两者不同，检查的位置和实际访问的位置可能不一致。

第五步：从单次对话改造成循环自检

到这里我们已经能完成"发一次请求、执行一次工具"的单次调用了。但 Agent 的核心是自己把活干完——你给它一个目标，它要能在循环里不断"思考 → 行动 → 观察"，直到自己觉得任务完成。

做法就是把前面的单次调用包进一个循环：每轮把 messages 发给模型，如果模型返回的是工具调用请求，就执行工具、把结果回填，然后进入下一轮；如果模型不再要求调工具、直接给出文字回复，说明它认为任务做完了，退出循环，把最终结果交给用户。

实际工程里还要加一个最大轮数上限，防止模型陷入"调工具 → 看错结果 → 再调"的死循环——超过上限就强制停下，把当前进度交给用户。

▼
Python
复制代码
def run(*session*, *user_input*, *client*, *model*=MODEL, *log*=print):
    messages = *session*.messages # 提取会话上下文
    messages.append({"role": "user", "content": *user_input*}) # 添加最新用户Query
    for _ in range(MAX_STEPS): # 开始循环
        # 校验是否需要压缩上下文
        prepare_context(*session*, *client*, *model*) 
        # 获取AI回复
        resp = *client*.chat.completions.create(*model*=*model*, *messages*=messages, *tools*=get_schemas())
        # 获取这轮TOKEN消耗以便去统计是否去压缩上下文
        response_usage = getattr(resp, "usage", None)
        prompt_tokens = getattr(response_usage, "prompt_tokens", None)
        # 拿到的 prompt_tokens 是否为非负整数。符合条件就保存原值，否则保存 0
        *session*.prompt_tokens = prompt_tokens if isinstance(prompt_tokens, int) and prompt_tokens >= 0 else 0

        msg = resp.choices[0].message
        # model_dump() 把它转换成 Python 字典，方便放入 messages。 exclude_none=True 忽略 None 值，避免占用空间。
        messages.append(msg.model_dump(*exclude_none*=True))

        # 没有工具调用，说明模型给出了最终回答 （这里只是简单的通用Agent说法-真实的感觉还是得AI判断结果是否符合QUery）
        if not msg.tool_calls:
            return msg.content or ""    

        # 模型可能一轮里请求多个工具，每个都必须回应
        for call in msg.tool_calls:
            name = call.function.name
            *log*(f"[tool] {name}")
            try:
                result = handle_call(call, *session*.perms)
            except Exception as error:
                result = f"Error: 处理工具调用失败（{type(error).__name__}）"
            if name != "load_skill" or not result.startswith(KEEP_PREFIXES):
                result = truncate_tool_result(result)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })

    return "（达到最大步数，已停止）"

第六步：开始接入Skill和Mcp

接入 Skill - 不能通过OpenAi的Tool返回

前面组装 system prompt 时，我们已经通过 skills_section() 把可用 Skill 的名称和描述放进了系统消息。

但模型看到名称和描述，只是知道“有哪些技能可以使用”，还没有拿到每个技能的完整操作说明。如果把所有 Skill 的正文都提前放进上下文，会占用大量 token；因此项目采用的是先展示目录，再按需读取正文的方式。

Skill 长什么样？

在本项目中，每个 Skill 是工作区 skills/ 目录下的一个子目录，其中的 SKILL.md 保存技能说明。例如：

▼
Plain
复制代码
skills/
├── git-commit/
│   └── SKILL.md
└── weekly-report/
    ├── SKILL.md
    └── template.md

SKILL.md 的开头是元信息，后面是具体的操作说明。以 weekly-report 为例：

▼
Plain
复制代码
---
name: weekly-report
description: 根据用户提供或可访问的工作记录整理周报，并区分完成、进行中、风险和计划。
---

# 工作周报

## 收集与核实

1. 确定报告周期、对象和格式。
2. 使用用户提供或当前环境中可访问的工作记录。
3. 区分已完成、进行中、受阻和计划。

上面是便于说明的正文节选。`name` 和 description 用来生成技能目录；正文则告诉模型匹配到该任务后应该怎么做。

需要注意：Skill 本身是一份说明文档，不是会自动运行的 Python 函数。 模型读到说明后，仍需要通过已有的文件工具、命令工具等完成实际操作，这些工具调用也仍然要经过权限检查。

第一步：扫描并生成技能目录

discover() 负责查找工作区中的 skills/*/SKILL.md，读取文件并解析头部信息：

▼
Plain
复制代码
def discover(skills_dir=None):
    skills_dir = skills_dir or SKILLS_DIR
    found = {}

    if not skills_dir.is_dir() or not skills_dir.resolve().is_relative_to(WORKSPACE):
        return found

    for f in sorted(skills_dir.glob("*/SKILL.md")):
        if not f.resolve().is_relative_to(skills_dir.resolve()):
            continue

        meta, _ = parse_skill(f.read_text(encoding="utf-8"))
        name = meta.get("name") or f.parent.name

        if not meta.get("description"):
            print(f"[skills] 跳过 {f}：缺少 description")
            continue

        found[name] = {
            "description": meta["description"],
            "path": f,
        }

    return found

这里做了几件事：

1. 只查找工作区 skills/ 下形如 技能目录/SKILL.md 的文件。

2. 通过 parse_skill() 拆出头部元信息和正文。

3. 取出 name 和 description；没有 name 时使用目录名，没有 description 则跳过。

4. 把技能描述和文件路径记录下来，供生成目录和后续加载使用。

这里虽然会读取 SKILL.md 文件来解析内容，但放进 system prompt 的只有名称和描述，正文不会在这一阶段发送给模型。

parse_skill() 当前只处理简单的 key: value 头部，不是完整的 YAML 解析器，因此元信息应保持简单。

接着，skills_section() 把扫描结果转成系统提示词中的技能目录：

▼
Plain
复制代码
def skills_section(found=None):
    found = discover() if found is None else found
    if not found:
        return ""

    lines = [
        "",
        "## 可用技能",
        "下面只列出技能的名字和用途。当任务和某个技能的用途匹配时，"
        "先调用 load_skill 读取完整说明，再按说明执行。",
    ]
    lines += [f"- {n}：{s['description']}" for n, s in found.items()]
    return "\n".join(lines)

结合前面已经介绍的 build_system()，模型在会话初始化后会看到类似这样的目录：

▼
Plain
复制代码
## 可用技能
下面只列出技能的名字和用途。当任务和某个技能的用途匹配时，
先调用 load_skill 读取完整说明，再按说明执行。

- weekly-report：根据用户提供或可访问的工作记录整理周报，并区分完成、进行中、风险和计划。

这里为了展示，只截取了一个技能。实际显示哪些技能，取决于工作区中扫描到的文件。

第二步：按需读取技能正文

当用户的任务与某个技能匹配时，模型可以调用 load_skill 工具。这个工具和第三步中注册的其他工具一样，通过 @tool 注册：

▼
Plain
复制代码
@tool(
    name="load_skill",
    description="读取一个技能的完整说明。任务和「可用技能」列表里某个技能的用途匹配时调用。",
    parameters={
        "type": "object",
        "properties": {
            "name": {
                "type": "string",
                "description": "技能名",
            }
        },
        "required": ["name"],
    },
)
def load_skill(name):
    found = discover()

    if name not in found:
        return f"Error: 没有名为 {name!r} 的技能，可用：{', '.join(found) or '无'}"

    path = found[name]["path"]
    _, body = parse_skill(path.read_text(encoding="utf-8"))
    extras = sorted(p.name for p in path.parent.iterdir() if p.name != "SKILL.md")
    rel = path.parent.relative_to(WORKSPACE)
    note = f"\n\n（技能目录 {rel}/ 下还有：{', '.join(extras)}）" if extras else ""

    return f"{SKILL_MARK} {name}\n{body}{note}"

load_skill 根据模型提供的 name 查找技能，读取对应 SKILL.md 的正文，并以 [skill] 开头返回。

例如，用户说“帮我整理本周工作周报”，模型可能发现 weekly-report 与任务匹配，于是请求：

▼
Plain
复制代码
load_skill(name="weekly-report")

程序执行后，把技能正文作为工具结果追加到 messages。下一轮模型就能看到这份说明，再按照说明收集资料、整理周报。

如果技能目录下还有其他文件，`load_skill` 会在结果中列出这些文件的名称，但不会自动读取它们的内容；需要时，模型还得继续调用相应工具读取。

第三步：让技能结果进入 Agent 循环

load_skill 能被模型调用，是因为 agent.py 导入了 skills 模块，模块中的 @tool 装饰器会将它注册到工具表。之后每次请求模型时，`get_schemas()` 都会把它的工具说明一起提供给模型。

整个过程串起来就是：

▼
Plain
复制代码
创建 Session
    ↓
扫描 Skill，将名称和描述放入 system prompt
    ↓
用户提出任务
    ↓
模型判断是否需要某个 Skill
    ↓
模型请求调用 load_skill
    ↓
程序读取 SKILL.md 正文并作为工具结果返回
    ↓
模型根据正文继续调用工具、完成任务

加载成功的技能结果以 [skill] 开头。项目会让这类结果跳过普通的单条工具结果截断，也不会在清理旧工具结果时直接替换掉它们。但如果之后触发了历史对话的摘要压缩，较早的技能内容仍可能被压缩成摘要，并不是永久完整保留。

还有两个实际使用时要注意的地方：

• 技能目录在创建 Session 时加入 system prompt。运行期间新增或修改技能后，已有会话的目录不会自动刷新，重新创建 Session 才会重新组装。

• 项目查找的是工作区根目录下的 ****skills/。仓库中的示例位于 init_project/skills/，因此要扫描到它们，工作区需要指向 init_project；如果工作区是仓库根目录，程序会去查找仓库根目录下的 skills/。

最后，系统提示词只是要求模型在任务匹配时调用 load_skill；模型是否实际调用、是否遵循正文，仍取决于模型表现，不能把它当成程序强制执行的规则。

接入MCP - 通过OpenAi的Tool返回

前面我们通过装饰器注册了 read_file、`write_file` 等本地工具。这些工具的实现都在当前项目里，Agent 可以直接调用对应的 Python 函数。

但有些能力可能由其他程序提供，例如查询数据库、访问项目管理系统或调用外部服务。如果每接入一个服务都手写一套通信代码，工具越来越多时就不容易管理。

MCP（Model Context Protocol）提供了一套统一的通信协议，让 Agent 可以连接提供能力的 MCP Server，发现并调用其中的工具。

这里要区分一下：MCP 本身是一套协议，不是单个工具。 一个 MCP Server 可以提供多个工具；本节只讨论其中的工具调用。MCP 官方架构说明

对于我们的 Agent，接入 MCP 后的流程可以分成三步：发现工具、交给模型、执行并返回结果。

第一步：连接 MCP Server，发现工具

Agent 需要先通过** MCP Client 连接 MCP Server**，然后发送 tools/list 请求，获取服务提供的工具列表。

假设某个天气服务返回了下面这样的工具说明：

▼
Plain
复制代码
{
  "name": "get_weather",
  "description": "查询指定城市的天气",
  "inputSchema": {
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "城市名称"
      }
    },
    "required": ["city"]
  }
}

这和第三步中的本地工具说明很相似，都包含三件事：工具名、工具描述、参数结构。区别在于，本地工具的说明来自项目里的注册表，MCP 工具的说明来自 MCP Server。

第二步：把 MCP 工具交给模型

当前项目发起模型请求时，通过 tools 参数传入本地工具说明：

▼
Plain
复制代码
resp = client.chat.completions.create(
    model=model,
    messages=messages,
    tools=get_schemas(),
)

因此，接入 MCP 后，需要先把 MCP Server 返回的工具说明转换成模型接口需要的格式，再与本地工具列表合并。

例如，MCP 工具的 inputSchema 需要放到当前接口格式中的 parameters 位置：

▼
Plain
复制代码
# 示意：mcp_tool 是从 MCP Server 获取的工具说明
model_tool = {
    "type": "function",
    "function": {
        "name": "weather_get_weather",
        "description": mcp_tool["description"],
        "parameters": mcp_tool["inputSchema"],
    },
}

再把转换后的 MCP 工具说明与本地工具一起发送：

▼
Plain
复制代码
# 示意：mcp_tool_schemas 是已经转换好的 MCP 工具列表
all_schemas = get_schemas() + mcp_tool_schemas

resp = client.chat.completions.create(
    model=model,
    messages=messages,
    tools=all_schemas,
)

这里需要注意，工具说明是通过 tools 参数发送的；它和前面放入 system_prompt 的 Skill 名称、描述不是同一个位置。

上面的 weather_get_weather 是 Agent 给模型使用的名称。实际接入时，还需要记录它对应哪个 MCP Server 以及 Server 上的原始工具名 get_weather，这样后面才能把请求转发到正确的位置。给不同服务的工具加上区分名称，也有助于避免工具重名。

第三步：执行 MCP 工具并回填结果

当用户提出问题后，模型可能选择调用 MCP 工具，并返回工具名和参数，例如：

▼
Plain
复制代码
工具名：weather_get_weather
参数：{"city": "杭州"}

这时 Agent 仍然需要先解析参数、检查权限。通过后，根据工具名称判断它来自哪个 MCP Server，再通过 MCP Client 发送 tools/call 请求。

MCP Server 执行后，会把结果返回给 Agent。例如：

▼
Plain
复制代码
{
  "content": [
    {
      "type": "text",
      "text": "杭州：22°C，晴"
    }
  ],
  "isError": false
}

上面的天气数据只是示意。Agent 拿到结果后，需要将其转换成模型接口可接受的内容，再像第三步处理本地工具那样追加到 messages：

▼
Plain
复制代码
# 示意：result_text 是从 MCP 工具结果中提取并转换后的文本
messages.append({
    "role": "tool",
    "tool_call_id": call.id,
    "content": result_text,
})

这里的 call.id 是模型返回的工具调用 ID，用于让模型知道这条结果对应哪次工具调用。追加后，Agent 再次请求模型，模型就可以根据工具结果继续处理或回答用户。

整体流程如下：

▼
Plain
复制代码
MCP Server 提供工具
        ↓
Agent 连接 Server，获取工具列表
        ↓
转换工具说明，与本地工具一起传给模型
        ↓
模型返回工具名和参数
        ↓
Agent 检查权限，通过 MCP Client 调用工具
        ↓
MCP Server 返回结果
        ↓
Agent 将结果追加到 messages
        ↓
模型读取结果，继续下一轮

需要注意，MCP 返回的结果不一定只有文本，还可能包含结构化数据、图片或错误信息。接入时需要根据结果类型做转换，并处理调用失败、超时等情况，不能一律把结果当作成功的普通字符串。MCP 官方工具规范

多 Agent 之间的协作

一、Subagents 和 Multi-Agent 是什么关系？

聊到多 Agent 协作，经常会听到两个词：**Subagents（子 Agent）**和 Multi-Agent（多 Agent）。

它们描述的是同一个领域中的不同层面：

• Multi-Agent 是更广义的概念。在本文讨论的协作场景中，多个 Agent 通过分工、信息交换或任务交接，共同完成一个目标，就可以构成一个多 Agent 系统。

• Subagents 是其中一种常见的组织方式：主 Agent 把一部分任务委派给子 Agent，子 Agent 完成后返回结果，再由主 Agent 继续处理和整合。

可以把它理解为：

多 Agent 描述的是“多个 Agent 如何协作”；子 Agent 描述的是其中一种“主 Agent 委派任务”的关系。

因此，主 Agent 带领多个子 Agent 的层级结构，本身就是一种多 Agent 系统。

还有一点容易混淆：多个 Agent 不一定意味着多个模型。 同一个底层模型，也可以在不同的角色指令、上下文、工具和执行过程中，承担多个 Agent 的任务。

二、Subagents：主 Agent 委派任务

假设用户提出一个需求：

“帮我看看这个项目哪里出了问题，顺手修一下。”

主 Agent 可以先把任务拆开：

1. 派一个子 Agent 阅读相关代码，定位可能的问题。

2. 派另一个子 Agent 检查现有测试，分析哪些行为缺少验证。

3. 收到结果后，由主 Agent 综合判断、修改代码并执行验证。

在这个过程中，主 Agent 负责整体目标、任务分配和结果整合；子 Agent 负责范围相对明确的具体任务。

但有两个边界需要分清。

第一，子 Agent 不一定并行执行。

如果两个任务相互独立，可以并行；如果后一个任务依赖前一个任务的结果，就需要串行。例如，先定位问题，再让另一个子 Agent 根据定位结果设计测试。

第二，看到多个 Agent 同时工作，也不一定意味着它们之间是主从关系。

它们可能是被同一个主 Agent 调度的子 Agent，也可能是通过其他方式协作的独立 Agent。界面上的呈现形式，不能直接决定底层的组织方式。

三、Multi-Agent：不止主 Agent 和子 Agent

把范围扩大，多个 Agent 的协作方式还可以更灵活。

比如一个内容生产流程：

1. 研究 Agent 收集资料、整理来源。

2. 写作 Agent 根据资料撰写初稿。

3. 审核 Agent 检查事实、逻辑和表达。

4. 发现问题后，将内容退回研究或写作环节，修订后再次审核。

这个流程可以由主 Agent 统一调度，也可以由预设程序控制交接顺序；某些系统还允许 Agent 之间直接通信和协作。

因此，设计多 Agent 系统时，重点不只是“用了几个 Agent”，还包括：

• 每个 Agent 负责什么？

• 任务和上下文如何交接？

• 谁决定下一步？

• 出现分歧或失败时如何处理？

• 到什么程度，任务才算完成？

例如，OpenAI Agents SDK 区分了由管理 Agent 调用其他 Agent，以及把控制权交给另一个 Agent 两类编排方式。实际系统也可以组合使用它们。 OpenAI Agents SDK：Agent orchestration

四、为什么要使用子 Agent：上下文隔离与任务分工

子 Agent 的一个重要价值，是让主 Agent 不必亲自消化所有中间过程。

例如，网页搜索、资料检索和代码排查可能产生大量工具调用结果。如果全部放进主 Agent 的上下文，主对话会很快变长，后续推理也更容易受到无关信息干扰。

这时，可以把检索工作交给子 Agent：

• 子 Agent 在自己的上下文中搜索、筛选和分析。

• 完成后，向主 Agent 返回结论、关键依据、来源和未解决的问题。

• 主 Agent 根据这些结果继续决策，必要时再查阅原始材料。

这样做能让主 Agent 的上下文更加聚焦。不过，结果也不能只剩一句结论：来源、证据、不确定性和关键限制如果被压缩掉，主 Agent 就可能无法判断结论是否可靠。

上下文变短，不等于总成本一定更低

这里需要区分两个问题：

1. 主 Agent 的上下文占用是否减少。

2. 整个任务的 Token 消耗和费用是否减少。

子 Agent 虽然只把精简结果交回主 Agent，但它自己搜索、阅读和推理的过程同样会消耗 Token。再加上任务分配、重复读取、结果汇总和失败重试，多 Agent 的总成本可能更高。

Anthropic 在其研究系统的实践中，也提到多 Agent 能并行处理大量信息，但会带来更高的 Token 消耗；是否值得取决于任务价值和具体场景。 Anthropic：多 Agent 研究系统

所以，子 Agent 更适合那些任务边界清晰、信息量较大，或确实需要独立调查的工作。简单问题不一定值得额外委派。

五、Fork：让子 Agent 从已有上下文继续工作

前面说的子 Agent，通常是由主 Agent 重新写一段任务说明，让它从相对独立的上下文开始工作。

另一种方式是 Fork（分叉）：从已有会话的某个状态创建一个新的执行分支。

如果框架支持继承完整会话，Fork 出来的分支就能看到当时保留的历史消息、工具调用和结果。主 Agent 不需要重新整理全部背景，子 Agent 就能从同一个起点继续探索。

例如，团队正在分析一个复杂故障，主会话已经积累了大量日志、代码线索和讨论。这时可以从当前状态分叉：

• 一个分支继续调查数据库问题。

• 另一个分支检查缓存和并发逻辑。

• 主 Agent 汇总结果，决定下一步。

但 Fork 的具体继承范围取决于产品和配置：有的可能继承完整历史，有的可能只传递部分上下文。Claude Code 的文档就区分了从独立上下文启动的子 Agent 和继承当前会话的 Fork。 Claude Code：Subagents

Fork 不等于环境隔离

会话上下文被分叉，并不意味着文件系统、数据库或其他外部环境也被复制。

如果多个分支仍然使用同一个工作目录，它们依然可能互相覆盖文件。需要隔离时，还要由具体框架或运行环境提供对应机制，例如独立工作目录或 Git worktree。

Fork 和 Prompt Cache

如果 Fork 后的请求保留了相同的输入前缀，并且满足模型服务的缓存条件，Prompt Cache 可能降低重复输入的处理成本。

但缓存命中并不保证发生：它取决于服务商、模型、前缀是否匹配、缓存有效期等条件。而且，缓存不会让已经继承进来的历史从子 Agent 的上下文窗口中消失。 OpenAI：Prompt caching

Fork 的主要价值，仍然是让多个分支从已有背景出发，分别探索问题或方案。

六、多 Agent 设计中的四个选择

了解概念之后，可以考虑怎样设计一个实际可控的系统。

下面这四点是一套偏稳妥的设计方案，适合作为起点；具体框架和业务场景也可以采用不同做法。

1. 默认让子 Agent 使用独立上下文

主 Agent 创建子 Agent 时，只向它提供完成任务所需的信息，例如：

• 任务目标和范围。

• 必要的背景与约束。

• 可使用的工具。

• 期望的输出格式。

这样做可以减少无关信息，但代价是：主 Agent 必须把任务说明写清楚。

需要注意，独立上下文不代表子 Agent 只能看到一段提示词。它是否还能读取项目文件、规则、记忆或其他资源，取决于系统配置和权限。

2. 区分调查型和执行型子 Agent

可以按职责将子 Agent 分为两类：

• Explore：负责查资料、阅读代码、分析问题，主要返回调查结果。

• Worker：负责修改文件、执行命令或完成其他具体工作。

这有助于控制职责和权限。但“叫 Explore 就一定安全”并不成立：真正的约束需要由工具权限和运行环境执行，而不是只靠角色名称。

Worker 能做哪些操作、是否需要用户确认，也取决于产品的权限策略和用户配置。

3. 默认只允许一层委派

在这套方案中，主 Agent 可以创建子 Agent，但子 Agent 不再继续创建子 Agent。

这样做的好处是：任务来源、调用链和成本更容易追踪，也更容易控制责任边界。对于定时任务等敏感能力，也可以选择只交给主 Agent。

这仍然是一项设计取舍。有些框架允许多层委派；当任务足够复杂时，多层结构可能有用，但也会增加协调、追踪和成本控制的难度。

4. 优先并行独立的只读任务

多个互不依赖的调查任务，可以同时执行。例如，让不同的子 Agent 分别调查代码、测试和文档。

对于写入任务，这套方案默认一次只执行一个，以降低文件覆盖和结果冲突的风险。

不过，并行写入并非天然不可行。如果任务边界明确、工作环境经过隔离，并且有合并与验证机制，也可以并行执行。关键在于是否能控制共享状态和冲突。

七、Task系统，AgentTool派活

在 Agent 内部的单个会话里，想开多个 Agent，本质上也是靠 Tool 系统实现的——这个 Tool 就是 AgentTool。它和我们日常用的工具没有本质区别，DeepSeek Harness 里的内置插件就是最好的例子。从下图可以看到，AgentTool 分两种模式：普通模式和 Fork 模式，对应前面讲过的两种子 Agent 策略。

除了子 Agent，shell 命令也可以放到后台运行。在系统层面，Task 是所有后台运行单元的统一抽象——不管是子 Agent 还是后台 shell，都被纳入同一个 Task 模型来管理。

每个 Task 都有标准的生命周期：pending → running → completed / failed / cancelled，系统会持续追踪进度信息。

大部分时候，主 Agent 发出 Task 之后就"撒手不管"了——因为 Task 执行完毕后，结果会自动以消息形式返回主 Agent。但为了以防万一，主 Agent 仍然配备了一整套工具来监控和管理后台任务，并不是真的就放任不管。

工具	作用
TaskGetTool	查看任务状态和结果
TaskListTool	列出所有后台任务
TaskStopTool	终止任务
TaskOutputTool	读取任务的完整输出

总结

这大概是我第一次认认真真做一次完整的技术总结。当然里面还有些很多细节还没写出来，其实但是可以当作入门教学了，在这里力推一个X的博主@antiAIvo 还有一个是叶天帝@yetone 前者做的是技术分享、后者是我想法Idea的缔造者同样它的magpie也是目前我觉得很牛逼的产品它拥有庞大的社群同时也拥有好的自循环Loop更新迭代！简直是我的偶像

之前学东西，多半是为了应付面试和工作需要。但真正进入职场后你会发现，身边的人都在用 AI——这时候，理解市面上 coding Agent 的底层逻辑就变得很有必要。搞懂了内核，你才能更好地驾驭 AI 帮你干活，也知道过程中该关注哪些关键信息、参数该怎么调。

当然，学这些也有很现实的考量：为了之后找更好的工作，以及也为了打造第一个我真正想长期维护的开源产品（全自动化运维的一整套AI驱动的软件开发）

以上就是上周 11.2 - 11.6 这五天，我上班摸鱼学完的全部内容。说起来有点惭愧，其实应该早点开始的——快毕业这段时间在学校里，一直没能沉下心来学点什么，内心焦虑又难受。现在的工资很低，就算转正也才 8K，这家公司对我来说只是一块垫脚石。想往上走，得先把地基打牢。我也不知道还有半年时间我能否达到我的目标月薪15K，可能最后真的很难吧

Agent 这个主题就先告一段落（面试被问到细节再回来补充）。接下来要开始深入学习 DeepSeek 的 Cordis 了——我很好奇他们是怎么搭建这套生态的，据说连上下文都能做成插件。这种细粒度的工程化和生态设计，从目前的反响来看，确实做得不错。

另外，最近也看了下 SuperAgent 这个概念框架，整体感觉还行，实际用下来还有不少地方需要打磨。不过官方也说了这是概念验证性质，所以倒也不用太较真。

上周学的 SuperAgent 内容，看时间整理一下也可以发出来。不过说实话，我没怎么看源码——就目前阶段而言，逐行读代码可能是性价比最低的事情。利用 AI 帮你梳理主要链路，搞懂内部是怎么运转的，就够了。

毕竟在我看来，未来的开发者可能都要往"决策者"的角色转变：不是亲手写每一行代码，而是判断方向、拆解任务、指挥 AI 执行。可能是最后是最懂业务的产品、又或者是最懂架构的架构师，谁又知道呢

（注：部分内容由豆包工作 AI 生成）