Agent Core & Optimization

AIAgent 内核决策流与 Prompt 缓存保护

深度透视 run_agent.py 会话循环、Cache Invariant 守则与轨迹自适应压缩机制
AIAgent 决策流与缓存优化
📊 图 2-1:AIAgent 核心生命周期循环与 Prompt Cache 优化全景图

🧠 1. AIAgent 核心配置与生命周期控制

主大脑 AIAgent 类定义在 run_agent.py 中,其 __init__ 构造器包含多个参数。初学者重点需要掌握以下几个核心配置属性:

  • max_iterations (默认 90):控制单个 Turn 内智能体连续调用工具的最大循环次数,到达该值将强行终止以防止失控的死循环。
  • iteration_budget:定义本次请求的 Token 计算预算(包含输入、输出和思考),一旦预算耗尽则拒绝后续的大模型请求。
  • providermodel:指定当前主会话的推理后端服务(如 openai, anthropic, deepseek)和具体的模型名称。
  • enabled_toolsets:工具箱白名单。仅允许大模型感知和调度该名单内的工具。
  • _budget_grace_call (布尔型):优雅退出的“余次保证”标记。当 Token 预算在工具执行中耗尽时,系统仍会预留最后一次单 Turn 交互,允许大模型输出一句对用户的妥善回复。

🔄 run_conversation 会话主循环详解

每次用户发送提问,系统都会调用 AIAgent.run_conversation()。它是在 agent/conversation_loop.py 支撑下运转的一个循环:

# run_conversation 伪代码逻辑演示
def run_conversation(self, user_message, ...):
    # Step 1: 从 DB 加载会话历史,构建 System Message 与 Caching 锚点
    messages = self._prepare_session_history(user_message)
    
    while api_call_count < self.max_iterations and self.iteration_budget.remaining > 0:
        # Step 2: 发送请求给 LLM
        response = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=self.get_tool_schemas()
        )
        
        # Step 3: 若是大模型给出的回复直接退出
        if not response.tool_calls:
            self._save_to_db(response.content)
            return response.content
            
        # Step 4: 循环拦截并串行分派 tool_calls
        for tool_call in response.tool_calls:
            result = self.handle_function_call(tool_call.name, tool_call.args)
            messages.append(self.format_tool_result_message(tool_call.id, result))
            
        api_call_count += 1

🔒 2. Prompt 缓存保护(Caching 保护三铁律)

为避免缓存击穿带来的高延迟和双倍 Token 计费,Hermes 在代码设计上通过三大防御机制来锁死缓存前缀的稳定性:

🛡️ 铁律一:Byte-stable System Prompt(有序的 JSON schema)

agent/system_prompt.pyformat_tools_for_system_message() 方法中,每次在把工具 Schema 组装进 System Prompt 时,系统绝不使用无序的 Python 字典,而是进行硬性升序排序并强行去除缩进空格:

# 字节级稳定提示词实现
def format_tools_for_system_message(tools_list: list) -> str:
    # 强制排序以规避因 Python dict 哈希随机化导致的字节流改变
    sorted_tools = sorted(tools_list, key=lambda x: x["name"])
    return json.dumps(sorted_tools, sort_keys=True, separators=(',', ':'))

🔄 铁律二:严格角色校直(Role Alternation Control)

某些 LLM 厂商对消息队列中的 role 交替非常严苛。如果开发者在使用 Skills 或 Plugin 时意外塞入了两个连续的 user 消息,或者将 tool 消息插在了错误的序列里,都会引发服务端接口报错。在 agent/agent_runtime_helpers.py 中,系统自动对历史消息数组进行检查与校直:若出现连续的 user 消息,底层自动将其内容在文本层面通过换行符合并为一条大 user 消息;若发现缺失 assistant 响应的 tool 结果,自动在中间插入一个合成的空白回复。

🚫 铁律三:禁止动态 System Prompt 注入

任何包含动态时间戳(如“当前时间是 10:21:55”)、运行负载或瞬时路径的系统元数据严禁直接注入 System Prompt 头部。这些动态数值会使缓存瞬间作废。Hermes 要求所有这类动态数值必须作为 user 消息或者作为工具调用返回值(如 get_system_time())提供给模型,保证 System Prompt 字节序列坚固如初。

📉 3. 轨迹自适应压缩细节 (Context Compaction)

随着 Turn 轮次叠加,会话 Token 数量会暴涨。为了防止爆上下文并节省 Caching 开销,系统在 agent/conversation_compression.py 中实现了一套压缩算法。流程由以下时序驱动:

轨迹压缩与数据库裁剪
📊 图 2-2:自适应轨迹压缩 compaction 及 DB 裁剪时序流转图
  • 触发阈值条件:当会话 Token 数量达到大模型主窗口的 threshold_percent 时,主循环会自动触发压缩阶段。
  • 压缩与 Summary 生成:系统调用 auxiliary.compression.model 指向的辅助模型,把即将被裁剪的历史轮次发给它进行要点提炼,生成 concise markdown 格式的 Summary 块。
  • System Prompt 头部拼接:生成完毕后,系统将该 Summary 插入到系统提示词头部的固定锚点位置:
    [Historical Summary]
    {Summary}
    [End of Historical Summary]
  • 数据库会话截断:新会话开始前,系统把旧历史做分割存储,然后直接在数据库中删除被总结的中间轮次,仅在内存和 state.db 中保留最新的尾部消息,使得首字延迟暴跌。

🤝 4. 多智能体派发与并发截断保护

当大任务被分解时,AIAgent 会在 Turn 中决策出调用 delegate_task。这一过程包含如下关键路径:

  • 分派路径:在 run_agent.py 中,_dispatch_delegate_task 方法会拦截这一动作,它不作为一个简单的 Python 函数执行,而是动态从 tools/delegate_tool.py 导入 delegate_task 模块,为子任务拉起全新的 AIAgent 线程。
  • 子智能体限流防死锁 (`_cap_delegate_task_calls`): 为了规避模型失控发起的并行子智能体风暴导致费用激增,系统在 run_agent.py_cap_delegate_task_calls() 中会强制对 tool_calls 列表进行清洗:
    # 并发截断处理
    delegate_count = sum(1 for tc in tool_calls if tc.function.name == "delegate_task")
    if delegate_count > MAX_CONCURRENT_CHILDREN:
        # 强行截断超出限制的子智能体调用,超出的部分被丢弃,待后续轮次重新调度
        tool_calls = [tc for tc in tool_calls if ...]
        logger.warning("Truncated excess delegate_task calls to enforce concurrency limits.")
  • Kanban 协作:子智能体会注册到项目 plugins/kanban/ 状态中,主智能体可以通过看板状态轮询子进程执行,并整合产出最终结果。