《Hermes AI Agent 框架核心剖析：基于事件溯源与 Saga 模式的高确定性多智能体编排实践》

graph TD
    User[用户请求] --> Gateway[接入层鉴权]
    Gateway --> Planner[规划器-生成执行计划树]
    Planner --> VectorDB[工具语义检索增强]
    VectorDB --> Saga[Saga协调器-开启全局事务]
    Saga --> Exec1[执行工具A-扣库存] 
    Exec1 --成功事件--> EventStore[(Kafka Event Log)]
    Exec1 --失败事件--> Compensator[补偿回滚器]
    Compensator --> Exec2[执行工具B-退款补偿]

from pymilvus import Collection
from openai import OpenAI
import json

class HermesToolRouter:
    def __init__(self):
        self.collection = Collection("tool_embeddings")
        self.client = OpenAI()
        
    def retrieve_tools(self, user_query: str, top_k: int = 5):
        # 1. 用户请求向量化
        query_embedding = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=user_query
        ).data[0].embedding
        
        # 2. 向量相似度检索
        search_params = {"metric_type": "IP", "params": {"nprobe": 10}}
        results = self.collection.search(
            data=[query_embedding],
            anns_field="embedding",
            param=search_params,
            limit=top_k,
            output_fields=["tool_name", "schema"]
        )
        
        # 3. 返回压缩后的工具定义，极大节省LLM上下文
        tool_definitions = []
        for hit in results[0]:
            tool_definitions.append({
                "type": "function",
                "function": json.loads(hit.entity.get('schema'))
            })
        return tool_definitions

from hermes.core import HermesAgent, SagaStep, compensate

@HermesAgent(name="travel_agent")
class TravelAgent:
    
    @SagaStep(order=1, on_fail="refund_flight")
    def book_flight(self, user_id, flight_no):
        # 执行机票预订
        return {"booking_id": "F123", "amount": 800}
    
    @compensate(for_step="book_flight")
    def refund_flight(self, context):
        # 若后续步骤失败，自动调用此补偿逻辑
        booking_id = context.get("booking_id")
        print(f"Executing refund for {booking_id}")
        return {"status": "refunded"}

// 协调器维护一个双向栈（Forward Stack / Backward Stack）
func (s *SagaCoordinator) Execute(ctx context.Context, saga *SagaDef) error {
    executedSteps := []string{}
    for _, step := range saga.Steps {
        // 执行正向操作
        if err := step.Execute(ctx); err != nil {
            log.Error("Step failed, starting compensation", "step", step.Name)
            // 逆序执行补偿
            for i := len(executedSteps) - 1; i >= 0; i-- {
                comp := saga.GetCompensation(executedSteps[i])
                if compErr := comp.Execute(ctx); compErr != nil {
                    // 补偿失败 => 写入死信队列(DLQ)，人工介入
                    s.dlq.Push(executedSteps[i], compErr)
                }
            }
            return err
        }
        executedSteps = append(executedSteps, step.Name)
    }
    return nil
}

hermes:
  event-store:
    type: kafka
    brokers: ["kafka-1:9092"]
    consumer-group: "hermes-replay-group"
    replay:
      enabled: true
      # 重放时自动启用时间戳过滤器
      time-filter: "2026-07-31T10:00:00Z"