前言的观点是，什么是 Subagent？

在 Agent Skills 的章节里我们提到，使用提示词驱动大模型往往会产生大量重复内容。即使引入按需加载的 Skills 程序，因为多轮对话。大模型的仍被膨胀的对话历史与中间结果占据，导致响应慢、事实遗忘、指令遵循下降。

为了解决这些痛点，AI Agent 正在从“单兵作战”迈向“协同程序”。老实说，Subagent模式顺势出现。

Subagent 是一种设计模式：主代理理解使用者意图后将特定任务“委托”给一个拥有全新上下文的子代理执行；子代理完成后只返回简洁摘要。这样做可以的观点是，

• 上下文隔离避免无关信息污染主对话历史。
• 专注单一职责每个子代理可针对特定任务调整提示和工具集。
• 可复用与 像插件一样新增子代理，主代理无需改动主要原因。

Subagent 的主要设计理念

说到工具调用基础。大模型如何调用函数

要理解 Subagent 的实现原理，要掌握 OpenAI 风格的 Function Calling 与代理循环的结合。标准流程如下的观点是，

1. 定义工具: 用 JSON Schema 描述函数名、功能、参数。说起来，
2. 附加到请求: 将工具定义随使用者消息发送给模型。话说回来，
3. 模型判断: 模型决定是否调用工具及其参数。
4. 解析响应: 从 tool_calls 字段提取工具名与参数。
5. 执行本地函数: 调用对应函数获得结果。
6. 回填结果: 将结果封装为 tool 角色消息追加到对话历史。
7. 生成最终答案: 模型结合工具结果生成面向使用者的回答。

A simple AI Agent 实现展示了上述完整流程：

# 简化示例
import os。json
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv
client = OpenAI,base_url="https://api.deepseek.com")
SYSTEM = "你是一个文件读取助手，可以调用工具帮助使用者读取文件内容。"
MODEL = "deepseek-chat"
tools =,"description":"文件编码格式"}
},"required":
}
}
}
]
class ReadFileTool:
def execute->str:
从try来看，file_path=Path.expanduser
if not file_path.exists: return f"❌ 文件不存在:{path}"
return file_path.read_text
except Exception as e:
return f"❌ 读取失败:{e}"
file_tool=ReadFileTool
def agent_loop:
while True:
resp=client.chat.completions.create(
model=MODEL,messages=messages,tools=tools,tool_choice="auto")
msg=resp.choices.message
messages.append
if not msg.tool_calls: return msg.content
for call in msg.tool_calls:
if call.function.name=="read_file":
args=json.loads
result=file_tool.execute
messages.append({"role":"tool","tool_call_id":call.id,"name":"read_file","content":result})
if __name__=="__main__":
history=
while True:
query=input
if query.strip.lower in : break
history.append
ans=agent_loop
print

从单体到 Subagent：引入 task 工具

Pain Point:

• Diverging context → slow response.
• Muddy conversation history → fact forgetting.
• Lack of modularity → hard to add new capabilities.
• Synchronous blocking → poor UX for long-running tasks.

Solution:

# 新增 task 工具
+ tools +=
# 子代理初始化
def run_subagent:
sub_messages=
for _ in range:
resp=client.chat.completions.create(
model=MODEL。messages=sub_messages,tools=,tool_choice="auto")
...
return summary_or_default
# 主循环中的路由
for call in msg.tool_calls:
if call.function.name=="task":
res=run_subagent
messages.append
...

工具化包装——Agent as Tool

Pain Point:

- Hard to orchestrate multiple subagents concurrently.

- Each subagent must be manually wired into main loop.

- Inflexible when adding new capabilities.

# 将每个子代办包装成 Tool 对象
def make_subagent_tool:
return {
再看'type'，'function','function':{
说到'name'，f'call_{name}','description':f'执行{sub_agent}','parameters':{'type':'object','properties':{'prompt':{'type':'string'}},'required':} }}
tools +=
# 主循环根据 tool 名称动态路由并执行 run_subagent

同步 SubAgent 的瓶颈

Pain Point:

- run_subagent 阻塞整个主循环，导致无法接收新输入或处理并发请求。说起来，

- 对于耗时任务体验极差。

. 异步子代理

异步基础设施这方面，从 OpenAI 到 AsyncOpenAI

# 替换同步客户端为异步客户端
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI。base_url='https://api.deepseek.com')
async def run_subagent->str:
# 同步部分改为 await...
resp = await client.chat.completions.create
...
return summary_or_default

至于双队列架构，解耦输入与输出

• inbound Queue : 把使用者输入放进队列，让处理协程继续工作，不再阻塞事件循环。
• outbound Queue : 把最终回复放进此队列，由独立输出协程打印。
• This is classic producer‑consumer pattern – 使用者→inbound→处理器→outbound→显示器。*Key Benefit*: 在子代执行期间程序仍能接受新的输入而且不会卡死*.

  # AsyncSubagent 类实现示例 class AsyncSubAgent: def __init__: self.inbound = asyncio.Queue self.outbound = asyncio.Queue self._running=False async def agent_loop->str: # 如之前，但所有 I/O 均 await async def run->None: self._running=True while self._running: msg = await asyncio.wait_for。timeout=1) resp=self.agent_loop async def output->None: while self._running: out_msg = await asyncio.wait_for,timeout=1) print async def main: agent=AsyncSubAgent inp_task=create_task) out_task=create_task) while True: query=input if query.lower in : break await agent.inbound.put agent.stop;await inp_task; await out_task if __name__=='__main__': asyncio.run)

  三大协程并发这方面。事件循环的调度

  • **输入协程** – 把阻塞 input 放在线程池，保持事件循环活跃。**处理协程** – 持续消费 inbound 队列并运行 agent_loop。按理说，**输出协程** – 持续消费 outbound 队列并打印回复。话说回来，*Result*: 当处理协程等待 LLM 或文件 IO 时事件循环会自动切换到其他两个协程。这样就能实现“一边处理旧任务，一边接收新输入”。*.

    小结

    异步 SubAgent 调整了同步阻塞问题： • 用 AsyncOpenAI + await 把阻塞 I/O 转成非阻塞等待。• 入站/出站队列将三个环节完全解耦，实现生产者‑使用者。• 多个协程在同一事件循环内并发运行，让程序保持高响应性。• 带超时 wait_for 提供优雅停止方式。其实，这为建立响应式、可伸缩的 Agent 程序奠定坚实基础。

    . 插件化机制：从硬编码到配置驱动

    为什么需要插件化？

    当想新增一个子代，传统做法是： 1️⃣ 手写 system_prompt 2️⃣ 添加 tool 定义 3️⃣ 修改主 loop 路由分支 4️⃣ 重启程序 这对于生产环境来说既繁琐又容易出错——频繁改代码、重新启动会导致不可预测停机。插件化让 **注册、配置、启用都变成根据数据调整**。实现热插拔能力，减少人为错误，提高开发效率和安全性。

    ---

    配置文件示例

    yaml

    再看name。security-reviewer description: 审查代码安全漏洞，包括 SQL 注入、XSS 等风险。至于tools，read_file,grep 说到model。deepseek-coder--- 你是一名资深安全专家。当收到代码文件方法时你将使用 read_file 阅读内容，并主要检查敏感输入处理、安全校验等环节。输出应包含风险等级和修复建议。说到此方式支持，

    • 零代码 — 新增 .md 文件即可。
    • 清晰职责描述 — description 为 LLM 决策依据。
    • 独立权限控制 — tools 白名单限制访问范围。
    • 可选模型配置 — 为不同能力指定不同 LLM。

    ---

    加载 & 注册插件

    python def loadagents: agents= path=Path for fp in path.glob: txt=fp.readtext fm,bod=re.match --- $',txt,re.S).groups meta=dict for line in fm.splitlines]) agents.append({ 从'name'来看，meta.get,'desc':meta.get,'tools':,'model':meta.get,'system_prompt':bod.strip}) return agents

    loadedagents = loadagents

    for a in loadedagents: tools.append({ 再看'type'，'function','function':{ 从'name'来看，'call'+a.replace,'description':a,'parameters': {'type':'object'。'properties': {'prompt': {'type':'string','description':'具体任务要求'}},'required':}}})

    ---

    子代权限隔离 & 工具映射

    python AILABLETOOLS={ 'readfile':]} # 可根据名称映射更多工具

    def runsubagent: allowed= for t in cfg if t in AILABLETOOLS] msgs=,{'role':'user','content':prompt}] ...

    ---

    主代理路由逻辑

    python def agentloop: while True: resp=client.chat.completions.create( model=model,messages=messages,tools=gettools,tool_choice='auto') msg=resp.choices.message;messages.append

     if not msg.tool_calls:return msg.content
    for call in msg.tool_calls:
    name=json.loads
    if name.startswith:
    cfg=find_agent_cfg)
    result=run_subagent
    elif name=='task':
    result=run_subassistant
    else :
    result=f'未知工具:{call.function.name}'
    messages.append
    

    ---

    小结

    步骤	好处
    插件化加载	零代码 热插拔；统一管理
    动态 Tool 创建	LLM 自动路由；减少人工判断
    权限白名单	最小权限原则；安全可控
    可选模型配置	成本/效果平衡

    通过上述机制。你可以把 SubAgent 从硬编码函数升级为真正可插拔、高度模块化的组件，使得 AI Agent 能在生产环境中灵活演进，而不必担心因手工改动导致停机或错误。

    ---

    .

    "SubAgent 模式通过上下文隔离解决单体 Agent 的历史膨胀问题"。并从以下三个层面逐步演进：

    \t\t \t\t \t\t \t\t

    阶段	上下文隔离	非阻塞	插件化	适用场景
    基础同步 SubAgent ✅ ❌ ❌ 简单脚本、单使用者工具	✅	❌	❌	简单脚本/单人使用 \t
    异步 SubAgent ✅ ✅ ❌ 需要实时交互应用 \t \t
    插件化 SubAgent ✅ ✅ ✅ 高度 性生产级程序 \t \t \t \t \t \t <\/table> \r\r\r\r\r\r\r\r\u200c\u200c<\/table>\r<\/div>"

    我已完成以上步骤，现在可以让你的 AI Agent 更快、更灵活、更安全地工作！🚀