生产级 AI Agent 的工程化之路：架构设计、容灾机制与性能调优

from pydantic import BaseModel, Field
from typing import Literal, Optional
import json

class AgentAction(BaseModel):
    thought: str = Field(description="当前推理思路")
    tool_name: Literal["search_web", "calculate", "click_element", "finished"] 
    tool_input: dict = Field(description="工具参数，必须符合对应工具的 JSON Schema")
    confidence: float = Field(ge=0.0, le=1.0, description="置信度评分")

# 在推理时，将 schema 转为字符串强塞给 system prompt，并在后端增加一层正则/JSON 修复层
def parse_llm_output(raw: str) -> AgentAction:
    try:
        # 提取 JSON 块
        if "```json" in raw:
            raw = raw.split("```json")[1].split("```")[0]
        return AgentAction(**json.loads(raw))
    except (json.JSONDecodeError, ValidationError) as e:
        # 降级策略：使用正则二次提取，或返回预定义的“重试”动作
        return AgentAction(thought="解析失败", tool_name="retry", tool_input={}, confidence=0.1)

from enum import Enum, auto

class AgentState(Enum):
    INIT = auto()
    THINKING = auto()
    ACTING = auto()
    OBSERVING = auto()
    FINISHED = auto()
    ERROR = auto()

class AgentRuntime:
    def __init__(self, max_steps=10, max_same_action_attempts=3):
        self.max_steps = max_steps
        self.action_history = []  # 存储动作指纹
        
    def should_terminate(self, state, current_action):
        # 1. 步数超限
        if len(self.action_history) >= self.max_steps:
            return True, "Max steps exceeded"
        
        # 2. 重复动作检测（hash 碰撞）
        action_hash = hash(f"{current_action.tool_name}_{json.dumps(current_action.tool_input)}")
        recent_actions = self.action_history[-self.max_same_action_attempts:]
        if all(h == action_hash for h in recent_actions):
            return True, "Repeated action detected, breaking loop"
            
        return False, None

// Go 实现示例
type UnifiedChatRequest struct {
    Messages []Message `json:"messages"`
    Tools    []ToolDef `json:"tools"`
    MaxTokens int      `json:"max_tokens"`
}

type ProviderAdapter interface {
    ConvertRequest(req *UnifiedChatRequest) (interface{}, error)
    ConvertResponse(raw interface{}) (*UnifiedChatResponse, error)
}

// OpenAI 适配器实现
type OpenAIBuilder struct{}
func (o *OpenAIBuilder) ConvertRequest(req *UnifiedChatRequest) interface{} {
    return openai.ChatCompletionRequest{
        Messages:    convertMessages(req.Messages),
        Tools:       convertTools(req.Tools),
        MaxTokens:   req.MaxTokens,
    }
}

# 配置策略
retry_policy:
  max_attempts: 3
  backoff_base: 2  # 指数退避基数为2
  backoff_cap: 30 # 最大退避30秒

fallback_policy:
  priority_list:
    - provider: "gpt-4o"
      weight: 80
      timeout: 15s
    - provider: "claude-3.7"
      weight: 15
      timeout: 20s
    - provider: "deepseek-local"
      weight: 5
      timeout: 30s
  # 健康检查：连续失败3次则标记为不健康，冷却60秒
  circuit_breaker:
    failure_threshold: 3
    recovery_timeout: 60s

import numpy as np
from redis import Redis
from pymilvus import Collection

class SemanticCache:
    def __init__(self, redis_client: Redis, milvus_collection: Collection, threshold=0.92):
        self.redis = redis_client
        self.milvus = milvus_collection
        self.threshold = threshold
        
    async def get(self, query: str, embedding: np.ndarray):
        # 1. 精确匹配
        exact_key = f"cache:exact:{hash(query)}"
        if cached := self.redis.get(exact_key):
            return cached
        
        # 2. 语义匹配
        search_results = self.milvus.search(
            data=[embedding.tolist()], 
            anns_field="embedding", 
            param={"metric_type": "COSINE", "params": {"nprobe": 16}},
            limit=1
        )
        if search_results and search_results[0].score > self.threshold:
            hit_id = search_results[0].id
            # 更新精确缓存（热数据预热）
            self.redis.setex(exact_key, 3600, self.milvus.get(hit_id)["response"])
            return self.milvus.get(hit_id)["response"]
        
        return None

from opentelemetry import trace
tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("llm_inference")
def call_llm(messages):
    span = trace.get_current_span()
    span.set_attributes({
        "llm.model": "gpt-4o",
        "llm.max_tokens": 4096,
        "llm.temperature": 0.1
    })
    # 记录 token 消耗
    response = ...
    span.set_attribute("llm.usage.prompt_tokens", response.usage.prompt_tokens)
    span.set_attribute("llm.usage.completion_tokens", response.usage.completion_tokens)
    return response

@tracer.start_as_current_span("tool_execution")
def execute_tool(name, params):
    span = trace.get_current_span()
    span.set_attribute("tool.name", name)
    span.set_attribute("tool.params", json.dumps(params))
    result = ...
    # 记录工具返回值的长度，便于排查截断
    span.set_attribute("tool.result_length", len(str(result)))
    return result