1. 项目概述：为什么我们需要一个统一的AI代理接口？

最近在折腾AI应用开发的朋友，估计都有一个共同的痛点：模型太多了。今天客户要求用GPT-4o做推理，明天老板说为了降本增效得切到国产大模型，后天测试又发现某个特定任务用Claude 3.5效果更好。每次切换，代码里就得改一遍API Key、Endpoint、调用方式，甚至参数命名都不同。这不仅仅是麻烦，更让整个系统的架构变得脆弱，维护成本直线上升。

这就是“CUA Agent框架”要解决的核心问题。CUA，你可以理解为“统一代理架构”的缩写，它的目标不是创造一个新的、更强大的AI模型，而是构建一个 中间层 ，一个 适配器 。它让你能用一套统一的代码和接口，去调用背后五花八门的AI模型（无论是OpenAI的GPT系列、Anthropic的Claude，还是国内的文心一言、通义千问，甚至是本地的Llama、Qwen）。你不再需要关心每个模型特有的SDK和参数细节，框架帮你处理了所有的差异。

想象一下，你开发了一个智能客服Agent，核心逻辑是处理用户query、调用LLM生成回复、再执行后续动作。如果没有CUA，你的代码里会充斥着 if model_type == “gpt-4”: … elif model_type == “claude-3”: … 这样的分支。而有了CUA，你只需要告诉框架：“给我一个能处理对话的Agent”，然后像调用一个标准函数一样使用它。底层是GPT还是Claude，只是一个配置文件里的一个参数。这对于追求 灵活性 和 成本效益 的团队来说，价值巨大。你可以根据响应速度、费用、任务类型，随时在多个模型间做A/B测试或无缝切换，而业务代码几乎无需改动。

2. CUA Agent框架的核心设计思路拆解

一个设计良好的统一接口框架，绝不是简单地把不同API封装一下。它需要一套深思熟虑的架构，来平衡易用性、扩展性和性能。CUA Agent框架的设计，通常围绕以下几个核心原则展开。

2.1 抽象与适配器模式：隔离变化的核心

这是整个框架的基石。其核心思想是： 定义一套稳定的、面向业务的抽象接口，让变化的部分（具体模型实现）通过“适配器”来接入。

1. 定义抽象层（Abstract Layer） ：首先，我们需要抽象出AI代理的核心能力。无论底层是什么模型，一个Agent通常需要提供诸如“生成文本”、“处理对话”、“理解指令”等功能。在CUA中，我们会定义一个顶层的 Agent 抽象类或接口。这个接口只声明业务方需要的方法，例如 generate(prompt: str, **kwargs) -> str 或 chat(messages: List[Dict]) -> Dict 。业务代码只依赖这个抽象接口编程，因此对底层实现一无所知，也就不会受到影响。

2. 实现适配器（Adapter） ：针对每一个需要接入的AI模型（如OpenAI、Anthropic、百度文心等），我们创建一个对应的适配器类（例如 OpenAIAdapter 、 ClaudeAdapter ）。这个适配器继承或实现上述抽象接口，但其内部封装了与该模型API交互的所有细节：包括HTTP客户端的初始化、认证（API Key）、请求体的构造（将通用参数映射为模型特定参数）、响应的解析（从模型返回的原始数据中提取出统一的格式），以及错误处理。

3. 工厂模式（Factory Pattern）动态创建 ：我们不会在代码里硬编码 new OpenAIAdapter() 。相反，会有一个 AgentFactory 。业务代码只需要告诉工厂：“我需要一个模型标识为 gpt-4 的Agent”。工厂根据这个标识，从配置或注册表中找到对应的适配器类，实例化并返回一个具体的适配器对象给业务方。业务方拿到的依然是 Agent 抽象接口，它并不知道自己手里拿的是OpenAIAdapter的实例。

这样，当需要新增一个模型时，开发者的工作变得极其单纯：1）为这个新模型编写一个适配器类；2）将这个适配器注册到工厂或配置中。所有现有的业务代码 零修改 就能开始使用新模型。这完美符合“开闭原则”（对扩展开放，对修改关闭）。

2.2 统一的消息与参数格式：让交互标准化

不同模型的API输入输出格式差异很大。例如，OpenAI的ChatCompletion接口使用 role 为 user / assistant / system 的message列表，而Claude的message结构略有不同，一些国内模型可能使用完全不同的字段名。

CUA框架内部必须定义一套 中立、通用 的消息格式和参数集。例如，定义一个 Message 类，包含 role （发言者）、 content （内容）等字段。所有业务代码和框架内部的核心流程都使用这套通用格式。

当调用具体适配器时，适配器的职责之一就是将通用的 Message 列表， 转换 成目标模型API所要求的特定格式。同样，对于参数，框架会定义一套“超集”参数（如 temperature 、 max_tokens 、 stream 等），适配器需要将通用参数映射到模型支持的参数上，对于模型不支持的参数，可以选择忽略或提供默认行为。

注意 ：统一格式的设计需要有一定的前瞻性和包容性，要考虑多模态（图像、音频）输入、函数调用（Tool Calling）、JSON模式等高级特性的占位设计，避免日后扩展时伤筋动骨。

2.3 上下文管理与会话保持

一个智能Agent往往不是“一问一答”就结束，它需要维护对话的历史上下文，才能进行连贯的多轮对话。CUA框架需要内置上下文管理机制。

1. 上下文存储 ：框架需要提供一个上下文存储（Context Store）的抽象。最简单的实现是内存存储（Memory Store），将当前会话的所有消息保存在内存的一个列表中。更复杂的实现可以支持基于数据库或Redis的持久化存储，以便支持跨会话、长窗口的对话。

2. 上下文窗口与修剪 ：大模型都有上下文长度限制。框架需要智能地管理上下文窗口。当历史消息的总token数接近模型上限时，需要有一套策略来修剪（trim）历史。常见的策略包括：丢弃最早的消息、总结早期对话内容、或优先保留系统指令和最近几轮对话。这个策略应该是可配置的，并由框架在调用适配器前自动执行。

3. 会话标识 ：为了支持多用户并发，框架需要通过一个唯一的 session_id 来区分不同的对话会话。上下文存储根据 session_id 来存取对应的历史消息列表。

这样，业务开发者只需关注当前轮次用户输入的内容，框架会自动将历史上下文组装好，发送给模型，从而实现连贯的对话体验。

3. 核心模块详解与实操要点

理解了设计思路，我们来看看如何动手实现一个最小可用的CUA Agent框架。我们将分模块拆解，并附上关键的Python代码示例和注意事项。

3.1 抽象层与基础接口定义

这是框架的“宪法”，定义了所有参与者都必须遵守的契约。

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional, AsyncIterator

class Message:
    """统一消息格式"""
    def __init__(self, role: str, content: str, **kwargs):
        self.role = role  # 如：'system', 'user', 'assistant', 'tool'
        self.content = content
        self.metadata = kwargs  # 预留扩展字段，如图像URL、工具调用结果等

class Agent(ABC):
    """Agent抽象基类。所有具体模型的适配器都必须实现这些方法。"""
    
    @abstractmethod
    def generate(self, prompt: str, **kwargs) -> str:
        """单轮文本生成（兼容传统Completion模式）"""
        pass
    
    @abstractmethod
    async def generate_async(self, prompt: str, **kwargs) -> str:
        """异步单轮文本生成"""
        pass
    
    @abstractmethod
    def chat(self, messages: List[Message], **kwargs) -> Message:
        """多轮对话（Chat模式），返回助理消息"""
        pass
    
    @abstractmethod
    async def chat_async(self, messages: List[Message], **kwargs) -> Message:
        """异步多轮对话"""
        pass
    
    @abstractmethod
    def chat_stream(self, messages: List[Message], **kwargs) -> AsyncIterator[str]:
        """流式多轮对话，返回一个异步生成器，逐块产出内容"""
        pass

实操心得 ：在抽象层就考虑同步和异步两种接口非常重要。现代AI应用多为高并发IO密集型，异步接口能极大提升吞吐量。同时，流式输出（Streaming）对于改善用户体验（减少等待感知）至关重要，必须在一开始就设计支持。

3.2 适配器实现：以OpenAI和Claude为例

接下来，我们实现两个具体适配器。这里会看到如何处理模型间的差异。

import openai
from anthropic import Anthropic
from .base_agent import Agent, Message

class OpenAIAdapter(Agent):
    """OpenAI GPT系列模型适配器"""
    
    def __init__(self, api_key: str, base_url: Optional[str] = None, model: str = "gpt-4o"):
        self.client = openai.OpenAI(api_key=api_key, base_url=base_url)
        self.model = model
    
    def _convert_messages(self, messages: List[Message]) -> List[Dict]:
        """将通用Message列表转换为OpenAI API格式"""
        openai_messages = []
        for msg in messages:
            # 基础转换
            openai_msg = {"role": msg.role, "content": msg.content}
            # 处理可能的工具调用（Function Calling）或图像内容等扩展
            if hasattr(msg, 'tool_calls') and msg.tool_calls:
                openai_msg['tool_calls'] = msg.tool_calls
            # 可以继续处理其他metadata...
            openai_messages.append(openai_msg)
        return openai_messages
    
    def chat(self, messages: List[Message], **kwargs) -> Message:
        # 1. 消息格式转换
        openai_messages = self._convert_messages(messages)
        # 2. 参数映射与合并。框架通用参数（如temperature）覆盖默认值。
        params = {
            "model": self.model,
            "messages": openai_messages,
            "temperature": kwargs.get('temperature', 0.7),
            "max_tokens": kwargs.get('max_tokens', 2000),
            # ... 其他OpenAI特有参数可以从kwargs中提取
        }
        # 3. 调用原生API
        response = self.client.chat.completions.create(**params)
        # 4. 将响应解析为统一的Message格式返回
        choice = response.choices[0]
        return Message(
            role=choice.message.role,
            content=choice.message.content,
            # 如果需要，可以保存finish_reason、usage等信息到metadata
            finish_reason=choice.finish_reason,
            usage=response.usage.dict() if response.usage else None
        )
    
    # 实现其他抽象方法：generate, chat_stream, 异步版本等...
    # chat_stream 需要处理SSE（Server-Sent Events）流式响应

class ClaudeAdapter(Agent):
    """Anthropic Claude系列模型适配器"""
    
    def __init__(self, api_key: str, model: str = "claude-3-5-sonnet-20241022"):
        self.client = Anthropic(api_key=api_key)
        self.model = model
    
    def _convert_messages(self, messages: List[Message]) -> List[Dict]:
        """Claude的消息格式与OpenAI略有不同，需要单独处理。"""
        claude_messages = []
        for msg in messages:
            # Claude的role通常是'user'或'assistant'，没有'system'（系统提示单独传）
            # 这里需要做一次映射，或者约定ClaudeAdapter使用特定的消息结构
            # 这是一个需要处理的差异点！
            claude_role = 'user' if msg.role == 'user' else 'assistant'
            claude_messages.append({"role": claude_role, "content": msg.content})
        return claude_messages
    
    def chat(self, messages: List[Message], **kwargs) -> Message:
        # Claude需要单独提取system message
        system_messages = [m.content for m in messages if m.role == 'system']
        system_prompt = "\n".join(system_messages) if system_messages else None
        
        # 过滤掉system消息，剩下的转换为Claude格式
        chat_messages = [m for m in messages if m.role != 'system']
        claude_messages = self._convert_messages(chat_messages)
        
        params = {
            "model": self.model,
            "max_tokens": kwargs.get('max_tokens', 1024),
            "temperature": kwargs.get('temperature', 0.7),
            "messages": claude_messages,
        }
        if system_prompt:
            params["system"] = system_prompt
            
        response = self.client.messages.create(**params)
        # 解析Claude的响应格式
        content_block = response.content[0]  # Claude响应内容在content列表中
        return Message(
            role='assistant',
            content=content_block.text,
            # Claude特有的响应字段
            stop_reason=response.stop_reason,
            model=response.model
        )

踩坑记录 ：不同模型对 system 角色的处理方式是天差地别的。OpenAI将其放在 messages 列表里，Claude要求作为独立的 system 参数，而有些国内模型可能根本不支持。适配器必须妥善处理这种差异。一个常见的做法是：在通用 Message 格式中保留 system 角色，但在每个适配器的 _convert_messages 方法中，按照目标API的约定进行提取和转换。这比强制统一所有模型的输入格式更可行。

3.3 工厂模式与配置化管理

工厂是解耦的关键。我们结合配置文件（如YAML）来实现灵活的模型管理。

# config/agents.yaml
agents:
  gpt-4:
    adapter: "openai"
    model: "gpt-4"
    api_key: "${OPENAI_API_KEY}"  # 支持从环境变量读取
    base_url: null
    default_params:
      temperature: 0.8
      max_tokens: 1000
      
  claude-sonnet:
    adapter: "claude"
    model: "claude-3-5-sonnet-20241022"
    api_key: "${ANTHROPIC_API_KEY}"
    default_params:
      temperature: 0.7
      max_tokens: 1024
      
  qwen-plus:
    adapter: "dashscope"  # 阿里云灵积
    model: "qwen-plus"
    api_key: "${DASHSCOPE_API_KEY}"

# agent_factory.py
import yaml
import os
from typing import Dict, Any
from .adapters.openai_adapter import OpenAIAdapter
from .adapters.claude_adapter import ClaudeAdapter
from .adapters.dashscope_adapter import DashScopeAdapter  # 假设已实现

class AgentFactory:
    _adapter_registry = {}
    _config = {}
    
    @classmethod
    def init_from_config(cls, config_path: str):
        with open(config_path, 'r') as f:
            raw_config = yaml.safe_load(f)
        cls._config = raw_config.get('agents', {})
        # 注册已知适配器类型
        cls.register_adapter('openai', OpenAIAdapter)
        cls.register_adapter('claude', ClaudeAdapter)
        cls.register_adapter('dashscope', DashScopeAdapter)
    
    @classmethod
    def register_adapter(cls, adapter_type: str, adapter_class):
        """允许运行时动态注册新的适配器类型"""
        cls._adapter_registry[adapter_type] = adapter_class
    
    @classmethod
    def get_agent(cls, agent_name: str, **override_kwargs) -> Agent:
        """根据配置名获取一个Agent实例"""
        if agent_name not in cls._config:
            raise ValueError(f"Agent '{agent_name}' not found in configuration.")
        
        agent_config = cls._config[agent_name].copy()
        # 解析环境变量
        for key, value in agent_config.items():
            if isinstance(value, str) and value.startswith('${') and value.endswith('}'):
                env_var = value[2:-1]
                agent_config[key] = os.getenv(env_var)
                if agent_config[key] is None:
                    raise EnvironmentError(f"Environment variable {env_var} not set for agent {agent_name}.")
        
        adapter_type = agent_config.pop('adapter')
        model = agent_config.pop('model')
        api_key = agent_config.pop('api_key')
        default_params = agent_config.pop('default_params', {})
        
        if adapter_type not in cls._adapter_registry:
            raise ValueError(f"Unsupported adapter type: {adapter_type}")
        
        AdapterClass = cls._adapter_registry[adapter_type]
        # 实例化适配器，传入必要参数
        agent_instance = AdapterClass(api_key=api_key, model=model, **agent_config)
        
        # 可以将默认参数绑定到实例上，方便后续调用
        agent_instance.default_params = default_params
        
        # 覆盖参数（如果有）
        if override_kwargs:
            agent_instance.default_params.update(override_kwargs)
            
        return agent_instance

使用起来就非常简单了：

# 初始化工厂
AgentFactory.init_from_config('config/agents.yaml')

# 业务代码中获取Agent，完全不需要知道底层是什么模型
def my_business_logic(user_input: str, agent_name: str = 'gpt-4'):
    agent = AgentFactory.get_agent(agent_name)
    # 使用统一的接口
    messages = [Message(role='user', content=user_input)]
    response = agent.chat(messages, temperature=0.5)  # 可以覆盖默认参数
    return response.content

3.4 上下文管理器的实现

一个简单的内存上下文管理器实现如下：

from collections import defaultdict
from typing import List, Optional
from .base_agent import Message

class MemoryContextManager:
    """基于内存的对话上下文管理器"""
    
    def __init__(self, max_tokens: int = 4000, pruning_strategy: str = 'fifo'):
        self._contexts = defaultdict(list)  # session_id -> List[Message]
        self.max_tokens = max_tokens
        self.pruning_strategy = pruning_strategy  # 'fifo', 'summary', 'priority'
        # 需要一个tokenizer来估算长度（简化起见，这里用字符数/4近似）
        # 实际应用中应集成tiktoken或模型对应的tokenizer
    
    def _estimate_tokens(self, text: str) -> int:
        """粗略估算token数。生产环境应使用准确的tokenizer。"""
        return len(text) // 4
    
    def add_message(self, session_id: str, message: Message):
        """向指定会话添加上下文消息"""
        self._contexts[session_id].append(message)
        self._prune_context(session_id)
    
    def get_messages(self, session_id: str) -> List[Message]:
        """获取指定会话的完整上下文消息"""
        return self._contexts[session_id].copy()
    
    def clear_context(self, session_id: str):
        """清空指定会话的上下文"""
        if session_id in self._contexts:
            del self._contexts[session_id]
    
    def _prune_context(self, session_id: str):
        """修剪上下文，使其token数不超过限制"""
        messages = self._contexts[session_id]
        total_tokens = sum(self._estimate_tokens(m.content) for m in messages)
        
        if total_tokens <= self.max_tokens:
            return
        
        if self.pruning_strategy == 'fifo':
            # 先进先出：丢弃最早的消息，直到满足要求
            while total_tokens > self.max_tokens and messages:
                removed_msg = messages.pop(0)
                total_tokens -= self._estimate_tokens(removed_msg.content)
        # 可以在此实现其他更智能的修剪策略，如总结前文等。

将上下文管理器集成到Agent的使用中：

class ConversationalAgent:
    """一个集成了上下文管理的高级Agent封装"""
    def __init__(self, agent_name: str, context_manager: MemoryContextManager):
        self.core_agent = AgentFactory.get_agent(agent_name)
        self.context_manager = context_manager
    
    def chat(self, session_id: str, user_input: str, **kwargs) -> str:
        # 1. 将用户输入构建为Message
        user_message = Message(role='user', content=user_input)
        # 2. 添加到上下文
        self.context_manager.add_message(session_id, user_message)
        # 3. 获取完整上下文
        history = self.context_manager.get_messages(session_id)
        # 4. 调用底层Agent
        assistant_response = self.core_agent.chat(history, **kwargs)
        # 5. 将助手回复也加入上下文
        self.context_manager.add_message(session_id, assistant_response)
        
        return assistant_response.content

4. 高级特性与扩展实践

一个基础的CUA框架搭建完成后，可以考虑引入更多生产级特性，使其更加强大和稳健。

4.1 负载均衡与故障转移

当你有多个同类型模型的API Key（例如多个OpenAI账号）或多个可用的模型终端（如GPT-4和Claude-3.5）时，可以实现负载均衡和故障转移。

1. 配置池 ：在配置中，为一个逻辑模型名（如 gpt-4-lb ）配置多个后端实例，每个实例包含自己的API Key和其他参数。

   gpt-4-pool:
     adapter: openai
     strategy: round_robin  # 或 least_connections, fallback
     instances:
       - model: gpt-4
         api_key: "${OPENAI_KEY_1}"
         weight: 5  # 权重
       - model: gpt-4
         api_key: "${OPENAI_KEY_2}"
         weight: 5
       - model: gpt-4-32k  # 甚至可以混用不同模型
         api_key: "${OPENAI_KEY_3}"
         weight: 3
   

2. 策略实现 ：在工厂或一个专门的 LoadBalancer 类中，根据策略（轮询、加权轮询、最少连接数等）选择下一个可用的实例。每次调用 get_agent 时，返回一个绑定到特定实例的适配器。

3. 故障转移 ：在适配器的调用方法中加入重试和异常处理逻辑。当某个实例因额度不足、网络超时等原因失败时，自动按策略切换到池中的下一个实例，并对失败实例进行冷却（circuit breaker）。

4.2 统一的功能调用（Tool Calling）支持

让不同模型的函数调用能力通过统一接口暴露，是提升Agent智能的关键。

1. 统一工具定义 ：框架定义一套描述工具（函数）的格式，例如包含 name 、 description 、 parameters （JSON Schema）的字典。

2. 调用抽象 ：在 Agent 抽象接口中增加 chat_with_tools 方法，接受消息列表和工具定义列表。

3. 适配器转换 ：每个适配器负责将统一工具定义转换为模型原生格式（如OpenAI的 tools 参数，Claude的 tools 参数格式略有不同）。当模型返回一个工具调用请求时，适配器再将其解析为框架统一的格式。

4. 执行与回调 ：框架提供一个 ToolExecutor ，负责根据工具名和参数，调用业务代码中注册的真实函数，并将执行结果格式化为 Message ，重新放回上下文，让模型进行下一步推理。

4.3 可观测性与监控

在生产环境中，必须知道Agent的运行状况。

1. 日志标准化 ：在每个适配器的关键节点（请求前、响应后、出错时）记录结构化日志。日志应包含：会话ID、模型类型、请求token数、响应token数、耗时、是否成功、错误码等。

2. Metrics埋点 ：集成监控系统（如Prometheus），暴露关键指标：每个模型的请求速率、延迟分布（P50, P95, P99）、错误率、token消耗速率等。

3. 链路追踪 ：集成OpenTelemetry等追踪工具，将一个用户请求在多个Agent或工具间的调用链路串联起来，便于排查复杂问题。

5. 常见问题、性能调优与踩坑实录

在实际开发和运维中，你会遇到各种各样的问题。这里记录一些典型场景和解决方案。

5.1 网络超时与重试策略

调用云端API，网络不稳定是常态。一个健壮的适配器必须有完善的错误处理。

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import openai

class RobustOpenAIAdapter(OpenAIAdapter):
    
    @retry(
        stop=stop_after_attempt(3),  # 最多重试3次
        wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避
        retry=retry_if_exception_type(
            (openai.APITimeoutError, openai.APIConnectionError, openai.RateLimitError)
        ),
        reraise=True  # 重试耗尽后抛出原异常
    )
    def chat(self, messages: List[Message], **kwargs) -> Message:
        # 在父类方法外包裹重试装饰器
        return super().chat(messages, **kwargs)

注意 ：并非所有错误都适合重试。对于 openai.BadRequestError （如提示词违规、参数错误），重试毫无意义，应立即失败。需要仔细区分可重试错误（网络、限流）和不可重试错误（客户端错误）。

5.2 上下文窗口与Token管理的真实挑战

前面我们用字符数简单估算了token，这在实际生产中远远不够。不同模型的token化规则不同（例如，中文在GPT和Claude中的token消耗差异很大）。

解决方案 ：

• 集成准确Tokenizer ：为每个适配器配置对应的tokenizer。OpenAI可以用 tiktoken ，Claude有官方库，国内模型可能需要使用Hugging Face的 transformers 库中的对应tokenizer。
• 实时计算与预警 ：在 ContextManager 的 add_message 和 _prune_context 方法中，使用准确的tokenizer计算消息的token数，并实时维护当前上下文的token总数。
• 智能修剪策略 ：当需要修剪时，简单的FIFO可能丢失重要信息。更优的策略是：
  • 优先级保留 ：永远保留 system 指令和最近几轮对话。
  • 总结压缩 ：当早期对话被修剪前，可以调用一个“总结Agent”（例如使用更便宜、更快的模型如 gpt-3.5-turbo ），将超出窗口的部分总结成一段简练的文字，替换掉原有的大量消息。这是一个高级但非常有效的特性。

5.3 多模型混用与路由策略

有时，一个任务的不同步骤适合用不同的模型。例如，用快速便宜的模型（如 gpt-3.5-turbo ）进行意图识别和任务分解，再用强大但昂贵的模型（如 GPT-4 或 Claude-3.5 ）执行核心推理，最后用快速模型进行格式检查和润色。

CUA框架可以扩展出一个 路由层（Router） 。

class ModelRouter:
    def __init__(self):
        self.rules = [
            {"condition": self._is_simple_formatting, "agent_name": "gpt-3.5-turbo"},
            {"condition": self._needs_deep_reasoning, "agent_name": "gpt-4"},
            {"condition": self._is_creative_writing, "agent_name": "claude-sonnet"},
        ]
        self.default_agent = "gpt-3.5-turbo"
    
    def route(self, session_history: List[Message], user_input: str) -> str:
        """根据历史和当前输入，决定使用哪个Agent"""
        for rule in self.rules:
            if rule["condition"](session_history, user_input):
                return rule["agent_name"]
        return self.default_agent
    
    def _is_simple_formatting(self, history, input_text):
        # 实现一些启发式规则，例如判断是否为简单的JSON校验、翻译等
        return len(input_text) < 50 and ("json" in input_text.lower() or "format" in input_text.lower())
    
    def _needs_deep_reasoning(self, history, input_text):
        # 判断是否需要深度推理，例如包含“为什么”、“分析”、“比较”等关键词
        reasoning_keywords = ["为什么", "分析", "比较", "解释", "原理"]
        return any(keyword in input_text for keyword in reasoning_keywords)
    
    # ... 其他规则

业务代码通过 Router 获取最合适的 agent_name ，再通过 AgentFactory 拿到对应的Agent实例。这样，模型的选择就从硬编码变成了可配置、可学习的策略。

5.4 成本控制与用量统计

多模型混用后，成本核算变得复杂。需要在框架层面集成成本计算。

1. 记录每次调用 ：在适配器中，记录请求的 prompt_tokens 和 completion_tokens 。
2. 维护成本表 ：建立一个模型成本对照表（如： gpt-4o 每1000个输入token $0.005，输出token $0.015）。
3. 实时计算与报告 ：每次调用后，根据token数和成本表计算本次调用费用，并累计到按模型、按项目、按用户的维度上。可以将数据推送到监控系统或数据库，用于生成账单和优化提示。

class CostAwareAgent(Agent):
    """代理装饰器，为原有Agent增加成本计算功能"""
    def __init__(self, wrapped_agent: Agent, cost_table: Dict):
        self._agent = wrapped_agent
        self._cost_table = cost_table
        self._total_cost = 0.0
    
    def chat(self, messages: List[Message], **kwargs) -> Message:
        response = self._agent.chat(messages, **kwargs)
        # 假设response.metadata中包含了usage信息
        usage = getattr(response, 'usage', None)
        if usage and 'prompt_tokens' in usage:
            model = getattr(self._agent, 'model', 'unknown')
            cost_per_1k_input = self._cost_table.get(model, {}).get('input', 0)
            cost_per_1k_output = self._cost_table.get(model, {}).get('output', 0)
            cost = (usage['prompt_tokens']/1000)*cost_per_1k_input + (usage['completion_tokens']/1000)*cost_per_1k_output
            self._total_cost += cost
            # 可以在这里记录日志或发送到metrics系统
        return response
    
    def get_total_cost(self) -> float:
        return self._total_cost

通过这样的设计，CUA Agent框架从一个简单的API聚合器，演进成了一个具备生产级能力（高可用、可观测、可管控、成本可控）的AI应用基础设施。它让开发者能真正专注于Agent的业务逻辑和创新，而不是陷入与不同AI API打交道的泥潭中。