深入解析 HermesAI Agent 框架：构建高扩展性智能代理系统

┌─────────────────────────────────────────┐
│            Application Layer            │  (用户业务逻辑)
├─────────────────────────────────────────┤
│          Orchestrator (状态机)           │  (循环调度、中断恢复)
├─────────────────────────────────────────┤
│   Perception │ Planner │ Executor │ Memory │  (四大核心模块)
├─────────────────────────────────────────┤
│      Message Bus (Event-driven)         │  (异步事件管道)
├─────────────────────────────────────────┤
│   Tool Registry │ Model Gateway │ Storage │  (基础设施)
└─────────────────────────────────────────┘

class PerceptionAdapter(ABC):
    @abstractmethod
    async def perceive(self, raw_input: Any) -> Observation:
        pass

class TextPerception(PerceptionAdapter):
    async def perceive(self, raw_input: str) -> Observation:
        # 注入 token 截断策略、敏感词过滤、语义压缩
        compressed = await self.compressor.compress(raw_input, max_tokens=4096)
        return Observation(type="text", content=compressed, metadata={"source": "user"})

class DynamicPlanner:
    def __init__(self, llm_gateway: LLMGateway):
        self.graph = nx.DiGraph()
        self.llm = llm_gateway

    async def plan(self, objective: str, context: List[Observation]) -> Plan:
        # 1. 使用 LLM 生成候选子目标列表
        candidates = await self._generate_subgoals(objective, context)
        # 2. 通过 LLM 推理建立依赖边
        edges = await self._infer_dependencies(candidates)
        self.graph.add_nodes_from(candidates)
        self.graph.add_edges_from(edges)
        # 3. 检查环路，若存在则请求 LLM 重规划
        if not nx.is_directed_acyclic_graph(self.graph):
            return await self._repair_cycle()
        # 4. 返回拓扑排序后的执行序列
        return Plan(sequence=list(nx.topological_sort(self.graph)))

@tool(name="weather_query", description="获取指定城市的实时天气")
async def get_weather(city: str, unit: Literal["celsius", "fahrenheit"] = "celsius") -> dict:
    # 实际调用外部 API
    return await http_client.get(f"https://api.weather.com/v1/{city}?unit={unit}")

class HierarchicalMemory:
    async def retrieve(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        # 先检索短期缓存
        short_term = self.short_term_cache.get(query)
        if short_term:
            return short_term
        # 长期向量检索
        long_term = await self.vector_store.search(query, top_k)
        # 根据相关度合并并排序
        merged = self._merge_and_rank(short_term, long_term)
        return merged[:top_k]

# 总线底层使用 Redis Streams 或内存队列
class MessageBus:
    async def publish(self, topic: str, message: Message):
        await self.redis.xadd(topic, message.dict())
    
    async def request(self, topic: str, payload: dict, timeout: float = 5.0) -> Message:
        # 生成唯一 correlation_id，订阅临时响应队列
        cid = str(uuid.uuid4())
        reply_topic = f"reply.{cid}"
        await self.subscribe(reply_topic, self._temp_handler)
        await self.publish(topic, Message(correlation_id=cid, reply_to=reply_topic, payload=payload))
        # 等待响应
        return await self._wait_for_reply(cid, timeout)

class SmartRouter:
    async def select_model(self, task: Task) -> str:
        complexity = self._estimate_complexity(task)
        if complexity < 0.3:
            return "llama3-8b"
        elif complexity < 0.7:
            return "gpt-4o-mini"
        else:
            return "gpt-4o"

sem = asyncio.Semaphore(5)  # 最大并发 5
async def call_with_limit(tool_func, *args):
    async with sem:
        return await tool_func(*args)

from hermesai import Agent, Tool, Memory

@Tool(name="calculator", description="执行四则运算")
async def calc(expr: str) -> float:
    return eval(expr)  # 注意：生产环境应使用安全沙箱

agent = Agent(
    tools=[calc],
    memory=Memory(type="vector", backend="milvus"),
    model="gpt-4o",
    planner="dynamic"
)

result = await agent.run("计算 (3 + 5) * 2，并记住这个结果用于后续")
print(result.final_answer)