深度拆解 HermesAI Agent 框架：打造函数调用优先的高性能多智能体协同系统

from dataclasses import dataclass
from typing import Any, List, Dict
from datetime import datetime
import json

@dataclass(frozen=True)  # 不可变
class AgentContext:
    session_id: str
    user_query: str
    plan: List[str]                    # 步骤列表
    current_step: int
    tool_calls: List[Dict]             # 已执行的工具调用记录
    observations: List[str]            # 观察结果
    final_answer: str = ""
    
    def append_tool_call(self, tool_name: str, args: Dict, result: Any):
        new_calls = self.tool_calls + [{"name": tool_name, "args": args, "result": result, "ts": datetime.utcnow().isoformat()}]
        new_obs = self.observations + [f"{tool_name} -> {str(result)[:100]}"]
        return AgentContext(
            session_id=self.session_id,
            user_query=self.user_query,
            plan=self.plan,
            current_step=self.current_step + 1,
            tool_calls=new_calls,
            observations=new_obs,
            final_answer=self.final_answer
        )

from pydantic import BaseModel, Field
from typing import Callable, Dict, Type, Any

class ToolDefinition:
    def __init__(self, func: Callable, schema: Type[BaseModel], description: str):
        self.func = func
        self.schema = schema
        self.description = description

class ToolRegistry:
    _tools: Dict[str, ToolDefinition] = {}
    
    @classmethod
    def register(cls, name: str, description: str, schema: Type[BaseModel]):
        def decorator(func: Callable):
            cls._tools[name] = ToolDefinition(func, schema, description)
            return func
        return decorator

# ---------- 业务示例 ----------
class GetSlowQueryArgs(BaseModel):
    time_range: str = Field(..., description="时间范围，如 'last_30min'")
    threshold: float = Field(1.0, description="慢查询阈值（秒）")

@ToolRegistry.register("get_slow_queries", "获取指定时间段的慢查询列表", GetSlowQueryArgs)
def fetch_slow_logs(time_range: str, threshold: float = 1.0) -> List[Dict]:
    # 伪代码：实际查询 performance_schema
    return [{"sql": "select * from orders...", "duration": 5.2}]

import asyncio
from openai import AsyncOpenAI

class HermesAgent:
    def __init__(self, client: AsyncOpenAI, registry: ToolRegistry, max_steps: int = 5):
        self.client = client
        self.registry = registry
        self.max_steps = max_steps

    async def run(self, query: str) -> str:
        ctx = AgentContext(
            session_id="sess_001",
            user_query=query,
            plan=[],
            current_step=0,
            tool_calls=[],
            observations=[]
        )
        
        messages = self._build_initial_messages(query)
        
        for step in range(self.max_steps):
            # 1. 调用 LLM
            response = await self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                tools=self._get_openai_tools(),
                tool_choice="auto"
            )
            
            msg = response.choices[0].message
            
            # 2. 检查是否结束
            if not msg.tool_calls:
                ctx = AgentContext(
                    **{**ctx.__dict__, "final_answer": msg.content}  # 简易更新，生产用专用方法
                )
                return msg.content
            
            # 3. 执行工具（并行优化）
            tool_results = await asyncio.gather(*[
                self._execute_tool_call(tc) for tc in msg.tool_calls
            ])
            
            # 4. 追加工具结果到消息上下文
            messages.append(msg)  # 助手消息（带 tool_calls）
            for tc, result in zip(msg.tool_calls, tool_results):
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": json.dumps(result, ensure_ascii=False)
                })
            
            # 5. 更新上下文（用于观测）
            # ... (省略 ctx 更新逻辑)
        
        raise TimeoutError(f"Agent 执行超过最大步数 {self.max_steps}")

@ToolRegistry.register("call_worker", "异步调用子智能体", CallWorkerArgs)
async def call_worker(worker_type: str, task: str):
    if worker_type == "sql_optimizer":
        return await SqlOptimizerAgent.run(task)
    elif worker_type == "report_generator":
        return await ReportAgent.run(task)
    ...