Agent Core & Optimization

AI Agent Core Decision Flow & Prompt Cache Protection

Deep dive into the run_agent.py conversation loop, Cache Invariant discipline, and adaptive trajectory compression
AI Agent Decision Flow and Cache Optimization
📊 Figure 2-1: AI Agent Core Lifecycle Loop & Prompt Cache Optimization Panorama

🧠 1. AIAgent Core Configuration & Lifecycle Control

The main brain AIAgent class is defined in run_agent.py. Its __init__ constructor contains several parameters. Beginners should focus on these core configuration properties:

  • max_iterations (default 90): Controls the maximum number of tool-calling iterations within a single Turn. Reaching this value forcibly terminates the loop to prevent infinite loops.
  • iteration_budget: Defines the token computation budget for this request (including input, output, and thinking). Once exhausted, all further LLM requests are denied.
  • provider & model: Specifies the inference backend for the current session (e.g. openai, anthropic, deepseek) and the specific model name.
  • enabled_toolsets: Tool whitelist. Only tools in this list are visible and dispatchable to the LLM.
  • _budget_grace_call (boolean): The "grace call" flag for graceful exit. When the token budget is exhausted during tool execution, the system reserves one final single-Turn interaction, allowing the LLM to output a proper response to the user.

🔄 run_conversation Session Loop Deep Dive

Each time a user sends a message, the system invokes AIAgent.run_conversation(). This loop runs on top of agent/conversation_loop.py:

# run_conversation pseudo-code
def run_conversation(self, user_message, ...):
    # Step 1: Load session history from DB, build System Message & Caching anchor
    messages = self._prepare_session_history(user_message)
    
    while api_call_count < self.max_iterations and self.iteration_budget.remaining > 0:
        # Step 2: Send request to LLM
        response = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=self.get_tool_schemas()
        )
        
        # Step 3: If LLM gives a direct reply, exit
        if not response.tool_calls:
            self._save_to_db(response.content)
            return response.content
            
        # Step 4: Intercept and serially dispatch 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 Cache Protection (Three Cache Invariant Rules)

To prevent cache misses that cause high latency and double token billing, Hermes employs three defense mechanisms to lock down the stability of the cached prefix:

🛡️ Rule One: Byte-stable System Prompt (Ordered JSON Schema)

In agent/system_prompt.py's format_tools_for_system_message(), every time tool schemas are assembled into the System Prompt, the system never uses unordered Python dicts but instead performs strict ascending sort and removes indentation whitespace:

# Byte-level stable prompt implementation
def format_tools_for_system_message(tools_list: list) -> str:
    # Force sort to avoid byte-stream changes from Python dict hash randomization
    sorted_tools = sorted(tools_list, key=lambda x: x["name"])
    return json.dumps(sorted_tools, sort_keys=True, separators=(',', ':'))

🔄 Rule Two: Strict Role Alternation Control

Some LLM providers are strict about role alternation in the message queue. If developers accidentally insert two consecutive user messages via Skills or Plugins, or place a tool message in the wrong sequence, the server will reject the request. In agent/agent_runtime_helpers.py, the system automatically inspects and repairs the message history: consecutive user messages are merged at the text level via newline joining; missing assistant responses before tool results are remedied by inserting a synthesized blank reply.

🚫 Rule Three: No Dynamic System Prompt Injection

Any system metadata containing dynamic timestamps (e.g., "Current time is 10:21:55"), runtime load, or transient paths must not be injected into the System Prompt header. These dynamic values instantly invalidate the cache. Hermes requires all such dynamic values to be provided as user messages or as tool call return values (e.g. get_system_time()), ensuring the System Prompt byte sequence remains pristine.

📉 3. Adaptive Trajectory Compression (Context Compaction)

As turns accumulate, session token counts grow rapidly. To prevent context overflow and reduce caching overhead, the system implements a compression algorithm in agent/conversation_compression.py. The process is driven by the following timing:

Trajectory Compression and Database Truncation
📊 Figure 2-2: Adaptive Trajectory Compaction & DB Truncation Timing Flow
  • Trigger Threshold: When the session token count reaches the LLM window's threshold_percent, the main loop automatically triggers the compression phase.
  • Compression & Summary Generation: The system calls the auxiliary model specified by auxiliary.compression.model, feeding it the historical turns slated for removal. It produces a concise Markdown-format Summary block.
  • System Prompt Header Concatenation: The Summary is inserted into a fixed anchor position at the top of the system prompt:
    [Historical Summary]
    {Summary}
    [End of Historical Summary]
  • Database Session Truncation: Before starting a new session, the system splits old history, deletes the summarized intermediate turns from the database, and retains only the most recent tail messages in memory and state.db, dramatically reducing time-to-first-token latency.

🤝 4. Multi-Agent Dispatch & Concurrency Truncation Protection

When large tasks are decomposed, AIAgent may decide to invoke delegate_task within a Turn. This process includes the following critical paths:

  • Dispatch Path: In run_agent.py, the _dispatch_delegate_task method intercepts this action. Rather than executing as a simple Python function, it dynamically imports the delegate_task module from tools/delegate_tool.py and spawns a new AIAgent thread for the subtask.
  • Sub-Agent Throttling & Deadlock Prevention (_cap_delegate_task_calls): To prevent the model from spawning a storm of parallel sub-agents and causing runaway costs, the system in run_agent.py's _cap_delegate_task_calls() forcibly cleans the tool_calls list:
    # Concurrency truncation
    delegate_count = sum(1 for tc in tool_calls if tc.function.name == "delegate_task")
    if delegate_count > MAX_CONCURRENT_CHILDREN:
        # Forcibly truncate excess sub-agent calls; extra calls are dropped and rescheduled in subsequent turns
        tool_calls = [tc for tc in tool_calls if ...]
        logger.warning("Truncated excess delegate_task calls to enforce concurrency limits.")
  • Kanban Collaboration: Sub-agents register themselves in the project plugins/kanban/ state. The main agent can poll the kanban state to monitor sub-process execution and synthesize the final result.