AI 模型配置常见问题

模型选择和配置

Q: Sira AI 支持哪些 AI 模型？

A: 我们支持主流的 AI 模型供应商：

国际模型：

• OpenAI：GPT-4o、GPT-4 Turbo、GPT-3.5 Turbo
• Anthropic：Claude 3.5 Sonnet、Claude 3 Haiku、Claude 3 Opus
• Google：Gemini Pro、Gemini Pro Vision
• Microsoft：Azure OpenAI 系列

国产模型：

• 阿里巴巴：通义千问（Qwen）系列
• 百度：文心一言（ERNIE）系列
• DeepSeek：DeepSeek Chat、DeepSeek Coder
• 智谱AI：GLM-4、ChatGLM 系列
• 月之暗面：Kimi Chat（Moonshot）
• 字节跳动：豆包（Doubao）系列

开源模型：

• Meta：Llama 2、Llama 3 系列
• Mistral：Mistral 7B、Mixtral 8x7B
• 科大讯飞：星火认知大模型

---

Q: 如何选择合适的 AI 模型？

A: 根据不同场景选择模型：

1. 通用对话场景：

• 推荐：GPT-3.5 Turbo、Claude 3 Haiku
• 特点：响应快，成本低，适合日常咨询

2. 复杂问题处理：

• 推荐：GPT-4o、Claude 3.5 Sonnet
• 特点：理解能力强，逻辑推理好，适合专业咨询

3. 代码和技术问题：

• 推荐：DeepSeek Coder、GPT-4 Turbo
• 特点：编程能力强，技术理解准确

4. 图像和多媒体处理：

• 推荐：GPT-4o、Claude 3.5 Sonnet、Gemini Pro Vision
• 特点：支持视觉识别，图片分析能力强

5. 中文场景优化：

• 推荐：通义千问、文心一言、GLM-4
• 特点：中文理解更准确，文化背景理解好

6. 成本敏感场景：

• 推荐：GPT-3.5 Turbo、Claude 3 Haiku、开源模型
• 特点：性价比高，满足基础需求

选择建议表：

使用场景	首选模型	备选模型	说明
客服咨询	GPT-3.5 Turbo	通义千问	快速响应，成本可控
技术支持	DeepSeek Coder	GPT-4 Turbo	技术准确性高
内容创作	Claude 3.5 Sonnet	GPT-4o	创意能力强
数据分析	GPT-4o	Claude 3 Opus	逻辑推理能力强
图片识别	GPT-4o	Gemini Pro Vision	视觉识别准确

---

Q: 如何配置 AI 模型的 API 密钥？

A: API 密钥配置步骤：

1. 在模型管理页面：

• 进入「模型管理」→「添加模型」
• 选择对应的模型供应商
• 填入必要信息

2. OpenAI 配置示例：

{
  "provider": "openai",
  "model_name": "gpt-4o",
  "api_key": "sk-xxx",
  "base_url": "https://api.openai.com/v1",
  "max_tokens": 4000,
  "temperature": 0.7
}

3. Claude 配置示例：

{
  "provider": "anthropic",
  "model_name": "claude-3-5-sonnet-20241022",
  "api_key": "sk-ant-xxx",
  "base_url": "https://api.anthropic.com",
  "max_tokens": 4000,
  "temperature": 0.7
}

4. 国产模型配置示例：

{
  "provider": "qianwen",
  "model_name": "qwen-turbo",
  "api_key": "sk-xxx",
  "base_url": "https://dashscope.aliyuncs.com/api/v1",
  "max_tokens": 2000,
  "temperature": 0.8
}

5. 测试连接：

• 配置完成后点击「测试连接」
• 系统会发送测试请求验证配置
• 确认能正常返回响应

---

Q: 模型参数如何调整？

A: 主要参数说明和调整建议：

1. Temperature（创造性）：

• 范围：0.0 - 1.0
• 0.1-0.3：适合事实性问答，回复一致性高
• 0.5-0.7：适合一般对话，平衡创造性和准确性
• 0.8-1.0：适合创意写作，回复更有变化

2. Max Tokens（最大长度）：

• 短回复：500-1000 tokens
• 中等回复：1000-2000 tokens
• 长回复：2000-4000 tokens
• 注意：tokens 数量影响成本

3. Top-p（多样性）：

• 0.1-0.5：更保守，适合专业咨询
• 0.6-0.9：平衡，适合一般对话
• 0.9-1.0：更多样，适合创意内容

4. Frequency Penalty（重复惩罚）：

• 0.0：允许重复
• 0.5：适度避免重复
• 1.0：强烈避免重复

参数调优示例：

# 客服场景配置
customer_service_config = {
    "temperature": 0.3,
    "max_tokens": 1000,
    "top_p": 0.6,
    "frequency_penalty": 0.3
}

# 创意写作配置
creative_writing_config = {
    "temperature": 0.8,
    "max_tokens": 3000,
    "top_p": 0.9,
    "frequency_penalty": 0.5
}

# 技术咨询配置
technical_support_config = {
    "temperature": 0.2,
    "max_tokens": 2000,
    "top_p": 0.5,
    "frequency_penalty": 0.2
}

---

Q: 为什么 AI 模型响应很慢？

A: 响应慢的常见原因及解决方案：

1. 模型选择问题：

• 问题：使用了大型模型（如 GPT-4 Opus）
• 解决：切换到快速模型（如 GPT-3.5 Turbo、Claude Haiku）
• 对比：GPT-4o（2-5秒）vs GPT-3.5（0.5-2秒）

2. 参数设置问题：

• 问题：max_tokens 设置过高
• 解决：根据实际需要调整长度限制
• 建议：日常对话设置 1000-1500 tokens

3. 网络延迟问题：

• 问题：服务器与 API 端点距离远
• 解决：使用国内代理或专线
• 优化：选择就近的 API 端点

4. 并发限制：

# 使用连接池优化
import httpx
from asyncio import Semaphore

class AIModelClient:
    def __init__(self, max_concurrent=10):
        self.semaphore = Semaphore(max_concurrent)
        self.client = httpx.AsyncClient(
            limits=httpx.Limits(max_connections=20)
        )
    
    async def generate(self, prompt):
        async with self.semaphore:
            response = await self.client.post(
                url=self.api_url,
                json={"prompt": prompt},
                timeout=30.0
            )
            return response.json()

5. 缓存策略：

# 实现智能缓存
import hashlib
from functools import lru_cache

def cache_key(prompt, model_config):
    content = f"{prompt}#{json.dumps(model_config, sort_keys=True)}"
    return hashlib.md5(content.encode()).hexdigest()

@lru_cache(maxsize=1000)
def get_cached_response(cache_key):
    # 从缓存获取响应
    return redis_client.get(f"ai_response:{cache_key}")

性能监控：

import time
import logging

def monitor_ai_performance(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            response_time = time.time() - start_time
            
            # 记录性能指标
            logging.info(f"AI响应时间: {response_time:.2f}s")
            
            if response_time > 5.0:
                logging.warning(f"AI响应过慢: {response_time:.2f}s")
            
            return result
        except Exception as e:
            logging.error(f"AI调用失败: {e}")
            raise
    return wrapper

---

Q: 如何控制 AI 模型的使用成本？

A: 成本控制策略：

1. 模型选择策略：

# 成本分层策略
cost_tiers = {
    "basic": {
        "model": "gpt-3.5-turbo",
        "max_tokens": 1000,
        "use_cases": ["简单问答", "基础客服"]
    },
    "standard": {
        "model": "gpt-4o-mini", 
        "max_tokens": 2000,
        "use_cases": ["复杂对话", "分析任务"]
    },
    "premium": {
        "model": "gpt-4o",
        "max_tokens": 4000,
        "use_cases": ["专业咨询", "创意写作"]
    }
}

2. 使用配额管理：

# 配额控制实现
class UsageQuotaManager:
    def __init__(self):
        self.daily_limits = {
            "department_A": 10000,  # tokens per day
            "department_B": 5000,
            "default": 2000
        }
    
    def check_quota(self, user_id, requested_tokens):
        department = get_user_department(user_id)
        used_today = get_daily_usage(department)
        limit = self.daily_limits.get(department, self.daily_limits["default"])
        
        if used_today + requested_tokens > limit:
            raise QuotaExceededException("今日使用量已达上限")
        
        return True

3. 智能缓存：

# 相似问题缓存
def similarity_cache(new_question, threshold=0.8):
    # 计算与历史问题的相似度
    for cached_q, cached_a in question_cache:
        similarity = calculate_similarity(new_question, cached_q)
        if similarity > threshold:
            return cached_a
    return None

4. 成本监控看板：

# 成本统计
def generate_cost_report():
    return {
        "daily_cost": calculate_daily_cost(),
        "monthly_cost": calculate_monthly_cost(),
        "cost_by_department": get_department_costs(),
        "cost_by_model": get_model_costs(),
        "usage_trends": get_usage_trends()
    }

5. 自动降级策略：

def smart_model_selection(question_complexity, user_tier):
    if question_complexity == "simple" and user_tier == "basic":
        return "gpt-3.5-turbo"
    elif question_complexity == "medium":
        return "gpt-4o-mini" 
    else:
        return "gpt-4o"

---

模型性能和质量

Q: 如何提高 AI 模型回复的质量？

A: 质量提升方法：

1. 提示词工程：

# 高质量提示词模板
system_prompt_template = """
你是一个专业的{role}，具有{expertise}经验。

你的职责：
- {responsibility_1}
- {responsibility_2}
- {responsibility_3}

回复要求：
1. 准确性：确保信息准确无误
2. 专业性：使用专业术语，体现专业水平
3. 完整性：提供完整的解决方案
4. 友好性：保持礼貌和耐心的语调

请根据用户问题提供专业、准确、有帮助的回复。
"""

# 示例：客服助手
customer_service_prompt = system_prompt_template.format(
    role="客服专员",
    expertise="5年客户服务",
    responsibility_1="解答客户疑问",
    responsibility_2="提供解决方案", 
    responsibility_3="记录问题反馈"
)

2. 上下文管理：

# 智能上下文截取
def manage_context(conversation_history, max_tokens=3000):
    # 保留系统提示词
    system_messages = [msg for msg in conversation_history if msg['role'] == 'system']
    
    # 保留最近的对话
    recent_messages = conversation_history[-10:]  # 最近10轮对话
    
    # 重要信息提取
    important_messages = extract_important_context(conversation_history)
    
    # 组合上下文
    context = system_messages + important_messages + recent_messages
    
    return truncate_to_token_limit(context, max_tokens)

3. 回复质量检查：

def quality_check(response):
    checks = {
        "length": len(response) > 10,  # 回复不能太短
        "relevance": check_relevance(response),  # 相关性检查
        "toxicity": not detect_toxic_content(response),  # 有害内容检测
        "factuality": verify_facts(response),  # 事实准确性
        "completeness": is_complete_answer(response)  # 回答完整性
    }
    
    quality_score = sum(checks.values()) / len(checks)
    
    if quality_score < 0.8:
        # 质量不达标，重新生成
        return regenerate_response()
    
    return response

4. A/B 测试优化：

# 回复质量 A/B 测试
def ab_test_prompts():
    test_groups = {
        "A": "简洁风格提示词",
        "B": "详细风格提示词", 
        "C": "问答风格提示词"
    }
    
    for group, prompt in test_groups.items():
        responses = generate_test_responses(prompt)
        quality_scores = evaluate_responses(responses)
        
        print(f"组{group}平均质量分数: {quality_scores['average']}")

---

Q: AI 模型出现幻觉（生成不准确信息）怎么办？

A: 幻觉问题解决方案：

1. 事实核查机制：

# 事实验证工具
def fact_check_response(response):
    # 提取可验证的声明
    claims = extract_factual_claims(response)
    
    verified_claims = []
    for claim in claims:
        # 使用搜索引擎验证
        search_results = web_search(claim)
        confidence = calculate_confidence(claim, search_results)
        
        if confidence < 0.7:
            verified_claims.append({
                "claim": claim,
                "status": "待验证",
                "confidence": confidence
            })
    
    if verified_claims:
        return add_uncertainty_markers(response, verified_claims)
    
    return response

2. 知识库集成：

# RAG（检索增强生成）实现
def rag_enhanced_generation(question):
    # 从知识库检索相关信息
    relevant_docs = knowledge_base.search(question, top_k=5)
    
    # 构建增强提示词
    enhanced_prompt = f"""
    基于以下可靠信息回答问题：
    
    参考资料：
    {format_documents(relevant_docs)}
    
    问题：{question}
    
    请仅基于提供的参考资料回答，如果参考资料中没有相关信息，请明确说明。
    """
    
    return ai_model.generate(enhanced_prompt)

3. 不确定性表达：

# 添加不确定性标识
def add_uncertainty_markers(response):
    uncertainty_phrases = [
        "根据我了解的信息",
        "据我所知",
        "一般情况下", 
        "建议您进一步确认",
        "这个信息可能需要核实"
    ]
    
    # 智能插入不确定性表达
    if contains_factual_claims(response) and confidence_score < 0.9:
        response = f"据我了解，{response}。建议您通过官方渠道进一步确认具体细节。"
    
    return response

4. 多模型验证：

# 多模型交叉验证
async def multi_model_verification(question):
    models = ["gpt-4o", "claude-3-sonnet", "qwen-max"]
    responses = []
    
    # 并行调用多个模型
    tasks = [call_model(model, question) for model in models]
    model_responses = await asyncio.gather(*tasks)
    
    # 寻找一致性答案
    consensus = find_consensus(model_responses)
    
    if consensus['agreement_rate'] > 0.8:
        return consensus['answer']
    else:
        return f"对于这个问题，不同模型给出了不同的答案，建议您咨询专业人士：\n{format_alternative_answers(model_responses)}"

5. 用户反馈循环：

# 用户纠错机制
def handle_user_correction(original_response, correction, question):
    # 记录错误案例
    error_case = {
        "question": question,
        "incorrect_response": original_response,
        "correct_answer": correction,
        "timestamp": datetime.now(),
        "user_id": get_current_user_id()
    }
    
    # 存储到错误知识库
    error_knowledge_base.add(error_case)
    
    # 更新模型提示词
    update_system_prompt_with_correction(error_case)
    
    # 感谢用户
    return "感谢您的指正！我已经记录这个信息，会在今后提供更准确的回复。"

---

Q: 如何让 AI 模型更好地理解企业专业术语？

A: 专业术语理解优化：

1. 专业词汇库构建：

{
  "enterprise_glossary": {
    "CRM": {
      "full_name": "客户关系管理",
      "definition": "用于管理企业与客户关系的系统",
      "context": "销售、市场、客服部门常用",
      "examples": ["我们的CRM系统记录了所有客户信息", "请在CRM中更新客户状态"]
    },
    "SLA": {
      "full_name": "服务水平协议", 
      "definition": "服务提供方与客户之间的服务质量承诺",
      "context": "IT服务、客户支持",
      "examples": ["我们承诺99.9%的SLA", "这个问题违反了SLA标准"]
    }
  }
}

2. 上下文感知提示词：

def create_domain_specific_prompt(domain, base_prompt):
    domain_knowledge = {
        "finance": {
            "terms": ["ROI", "EBITDA", "现金流", "资产负债表"],
            "context": "你是一位资深财务专家，熟悉各种财务指标和报表分析。"
        },
        "it": {
            "terms": ["API", "数据库", "云计算", "微服务"],
            "context": "你是一位经验丰富的IT专家，精通系统架构和技术解决方案。"
        },
        "hr": {
            "terms": ["KPI", "OKR", "绩效考核", "人才盘点"],
            "context": "你是一位专业的HR专家，了解人力资源管理的各个方面。"
        }
    }
    
    if domain in domain_knowledge:
        domain_info = domain_knowledge[domain]
        enhanced_prompt = f"""
        {domain_info['context']}
        
        专业术语理解：
        {', '.join(domain_info['terms'])}
        
        {base_prompt}
        
        请使用准确的专业术语，确保回复的专业性和准确性。
        """
        return enhanced_prompt
    
    return base_prompt

3. 术语消歧义：

def disambiguate_terms(text, company_context):
    # 术语多义性处理
    ambiguous_terms = {
        "应用": {
            "it_context": "软件应用程序",
            "hr_context": "工作申请", 
            "general_context": "使用、运用"
        },
        "流程": {
            "business_context": "业务流程",
            "it_context": "系统流程",
            "manufacturing_context": "生产流程"
        }
    }
    
    # 根据上下文判断正确含义
    for term, meanings in ambiguous_terms.items():
        if term in text:
            context = detect_context(text, company_context)
            if context in meanings:
                # 在提示词中明确术语含义
                text = text.replace(term, f"{term}({meanings[context]})")
    
    return text

4. 领域知识注入：

# 动态知识注入
def inject_domain_knowledge(prompt, user_department):
    knowledge_base = load_department_knowledge(user_department)
    
    relevant_knowledge = knowledge_base.search(
        prompt, 
        top_k=3,
        similarity_threshold=0.7
    )
    
    if relevant_knowledge:
        enhanced_prompt = f"""
        相关企业知识：
        {format_knowledge(relevant_knowledge)}
        
        用户问题：{prompt}
        
        请结合上述企业知识回答问题。
        """
        return enhanced_prompt
    
    return prompt

5. 持续学习机制：

# 专业术语学习
class TermLearningSystem:
    def __init__(self):
        self.term_usage_stats = {}
        self.context_patterns = {}
    
    def learn_from_conversation(self, conversation):
        # 提取专业术语
        terms = extract_professional_terms(conversation)
        
        for term in terms:
            # 统计使用频率
            self.term_usage_stats[term] = self.term_usage_stats.get(term, 0) + 1
            
            # 学习上下文模式
            context = extract_context(conversation, term)
            if term not in self.context_patterns:
                self.context_patterns[term] = []
            self.context_patterns[term].append(context)
    
    def get_term_definition(self, term):
        if term in self.context_patterns:
            # 基于历史上下文推断定义
            contexts = self.context_patterns[term]
            return generate_definition_from_contexts(contexts)
        
        return None

---

模型集成和部署

Q: 如何添加新的 AI 模型供应商？

A: 新供应商集成步骤：

1. 供应商适配器开发：

# 创建新的模型适配器
class NewProviderAdapter(BaseModelAdapter):
    def __init__(self, api_key, base_url=None):
        self.api_key = api_key
        self.base_url = base_url or "https://api.newprovider.com/v1"
        self.client = httpx.AsyncClient()
    
    async def generate(self, messages, **kwargs):
        """生成回复的核心方法"""
        payload = self._build_request_payload(messages, **kwargs)
        
        try:
            response = await self.client.post(
                f"{self.base_url}/chat/completions",
                json=payload,
                headers=self._get_headers(),
                timeout=30.0
            )
            
            response.raise_for_status()
            return self._parse_response(response.json())
            
        except Exception as e:
            raise ModelAPIException(f"API调用失败: {e}")
    
    def _build_request_payload(self, messages, **kwargs):
        return {
            "model": kwargs.get("model_name"),
            "messages": self._convert_messages(messages),
            "max_tokens": kwargs.get("max_tokens", 1000),
            "temperature": kwargs.get("temperature", 0.7),
            "stream": kwargs.get("stream", False)
        }
    
    def _get_headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "User-Agent": "Sira-AI/1.0"
        }
    
    def _parse_response(self, response_data):
        return {
            "content": response_data["choices"][0]["message"]["content"],
            "usage": response_data.get("usage", {}),
            "model": response_data.get("model")
        }

2. 模型注册：

# 在模型管理器中注册新供应商
class ModelManager:
    def __init__(self):
        self.providers = {
            "openai": OpenAIAdapter,
            "anthropic": AnthropicAdapter,
            "qianwen": QianwenAdapter,
            "newprovider": NewProviderAdapter,  # 新增
        }
    
    def create_model_instance(self, config):
        provider = config.get("provider")
        if provider not in self.providers:
            raise UnsupportedProviderException(f"不支持的供应商: {provider}")
        
        adapter_class = self.providers[provider]
        return adapter_class(**config)

3. 配置界面更新：

// 前端配置表单
const providerConfigs = {
  newprovider: {
    name: "新供应商",
    fields: [
      { name: "api_key", label: "API密钥", type: "password", required: true },
      { name: "base_url", label: "API地址", type: "url", required: false },
      { name: "model_name", label: "模型名称", type: "select", 
        options: ["model-v1", "model-v2"] }
    ],
    testConnection: async (config) => {
      return await testNewProviderConnection(config);
    }
  }
};

4. 流式响应支持：

async def stream_generate(self, messages, **kwargs):
    """流式生成支持"""
    payload = self._build_request_payload(messages, stream=True, **kwargs)
    
    async with self.client.stream(
        "POST",
        f"{self.base_url}/chat/completions",
        json=payload,
        headers=self._get_headers()
    ) as response:
        async for line in response.aiter_lines():
            if line.startswith("data: "):
                data = line[6:]  # 去除 "data: " 前缀
                
                if data.strip() == "[DONE]":
                    break
                
                try:
                    chunk = json.loads(data)
                    content = chunk["choices"][0]["delta"].get("content", "")
                    if content:
                        yield content
                except json.JSONDecodeError:
                    continue

---

Q: 如何实现模型故障切换？

A: 故障切换策略实现：

1. 健康检查机制：

import asyncio
from datetime import datetime, timedelta

class ModelHealthChecker:
    def __init__(self):
        self.health_status = {}
        self.failure_counts = {}
        self.last_check_time = {}
    
    async def check_model_health(self, model_config):
        """检查单个模型的健康状态"""
        try:
            # 发送测试请求
            test_prompt = "你好，这是一个连接测试。"
            start_time = datetime.now()
            
            response = await self._test_model_connection(model_config, test_prompt)
            
            response_time = (datetime.now() - start_time).total_seconds()
            
            # 更新健康状态
            self.health_status[model_config['id']] = {
                "status": "healthy",
                "response_time": response_time,
                "last_check": datetime.now(),
                "error": None
            }
            
            # 重置失败计数
            self.failure_counts[model_config['id']] = 0
            
            return True
            
        except Exception as e:
            # 增加失败计数
            model_id = model_config['id']
            self.failure_counts[model_id] = self.failure_counts.get(model_id, 0) + 1
            
            self.health_status[model_id] = {
                "status": "unhealthy",
                "response_time": None,
                "last_check": datetime.now(),
                "error": str(e)
            }
            
            return False
    
    async def periodic_health_check(self, models, interval=60):
        """周期性健康检查"""
        while True:
            tasks = [self.check_model_health(model) for model in models]
            await asyncio.gather(*tasks, return_exceptions=True)
            await asyncio.sleep(interval)

2. 故障切换逻辑：

class FailoverManager:
    def __init__(self, health_checker):
        self.health_checker = health_checker
        self.failover_rules = {
            "max_failures": 3,  # 最大失败次数
            "failover_timeout": 300,  # 故障切换超时（秒）
            "primary_recovery_check": 60  # 主模型恢复检查间隔
        }
    
    async def get_available_model(self, preferred_models):
        """获取可用的模型"""
        for model_config in preferred_models:
            model_id = model_config['id']
            
            # 检查模型健康状态
            if self._is_model_available(model_id):
                return model_config
        
        # 所有首选模型都不可用，抛出异常
        raise NoAvailableModelException("没有可用的模型")
    
    def _is_model_available(self, model_id):
        """检查模型是否可用"""
        health_info = self.health_checker.health_status.get(model_id)
        failure_count = self.health_checker.failure_counts.get(model_id, 0)
        
        if not health_info:
            return True  # 新模型，尚未检查，假设可用
        
        # 检查是否超过最大失败次数
        if failure_count >= self.failover_rules["max_failures"]:
            # 检查是否过了故障切换超时时间
            time_since_failure = (datetime.now() - health_info["last_check"]).total_seconds()
            if time_since_failure < self.failover_rules["failover_timeout"]:
                return False
        
        return health_info.get("status") == "healthy"

3. 智能模型选择：

class SmartModelSelector:
    def __init__(self, failover_manager):
        self.failover_manager = failover_manager
        self.performance_history = {}
    
    async def select_optimal_model(self, context):
        """根据上下文选择最优模型"""
        # 获取候选模型列表
        candidate_models = self._get_candidate_models(context)
        
        # 按性能和可用性排序
        ranked_models = self._rank_models(candidate_models, context)
        
        # 选择最优可用模型
        for model_config in ranked_models:
            try:
                if await self.failover_manager.get_available_model([model_config]):
                    return model_config
            except NoAvailableModelException:
                continue
        
        raise NoAvailableModelException("没有合适的模型可用")
    
    def _rank_models(self, models, context):
        """模型排序算法"""
        scored_models = []
        
        for model in models:
            score = self._calculate_model_score(model, context)
            scored_models.append((score, model))
        
        # 按分数降序排列
        scored_models.sort(key=lambda x: x[0], reverse=True)
        
        return [model for score, model in scored_models]
    
    def _calculate_model_score(self, model, context):
        """计算模型分数"""
        base_score = model.get('performance_rating', 0.5)
        
        # 响应时间权重
        health_info = self.failover_manager.health_checker.health_status.get(model['id'])
        if health_info and health_info.get('response_time'):
            response_time_score = min(1.0, 5.0 / health_info['response_time'])
            base_score += response_time_score * 0.3
        
        # 任务适配性权重
        task_compatibility = self._check_task_compatibility(model, context)
        base_score += task_compatibility * 0.4
        
        # 成本效益权重
        cost_efficiency = 1.0 / max(model.get('cost_per_token', 1), 0.001)
        base_score += cost_efficiency * 0.2
        
        return base_score

4. 故障恢复：

class FailureRecoveryManager:
    def __init__(self, health_checker):
        self.health_checker = health_checker
    
    async def attempt_recovery(self, failed_model_id):
        """尝试恢复故障模型"""
        # 等待一段时间后重试
        await asyncio.sleep(30)
        
        # 重新检查模型健康状态
        model_config = get_model_config_by_id(failed_model_id)
        if await self.health_checker.check_model_health(model_config):
            logger.info(f"模型 {failed_model_id} 已恢复正常")
            return True
        
        return False
    
    async def gradual_recovery_test(self, model_id):
        """渐进式恢复测试"""
        # 逐步增加测试负载
        test_loads = [1, 5, 10, 20]  # 并发请求数
        
        for load in test_loads:
            success_rate = await self._test_with_load(model_id, load)
            
            if success_rate < 0.9:  # 成功率低于90%
                logger.warning(f"模型 {model_id} 在负载 {load} 下成功率为 {success_rate}")
                return False
            
            await asyncio.sleep(10)  # 等待间隔
        
        logger.info(f"模型 {model_id} 通过了渐进式恢复测试")
        return True

---

Q: 如何监控 AI 模型的使用情况？

A: 综合监控方案：

1. 实时监控指标：

from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List
import asyncio

@dataclass
class ModelMetrics:
    model_id: str
    total_requests: int
    successful_requests: int
    failed_requests: int
    average_response_time: float
    tokens_used: int
    cost: float
    last_updated: datetime

class ModelMonitor:
    def __init__(self):
        self.metrics = {}
        self.real_time_data = {}
        
    def record_request(self, model_id: str, request_data: dict):
        """记录请求指标"""
        if model_id not in self.metrics:
            self.metrics[model_id] = ModelMetrics(
                model_id=model_id,
                total_requests=0,
                successful_requests=0,
                failed_requests=0,
                average_response_time=0.0,
                tokens_used=0,
                cost=0.0,
                last_updated=datetime.now()
            )
        
        metrics = self.metrics[model_id]
        metrics.total_requests += 1
        
        if request_data.get('success'):
            metrics.successful_requests += 1
            
            # 更新平均响应时间
            response_time = request_data.get('response_time', 0)
            metrics.average_response_time = (
                (metrics.average_response_time * (metrics.successful_requests - 1) + response_time)
                / metrics.successful_requests
            )
            
            # 累计使用量
            tokens = request_data.get('tokens_used', 0)
            metrics.tokens_used += tokens
            metrics.cost += self._calculate_cost(model_id, tokens)
        else:
            metrics.failed_requests += 1
        
        metrics.last_updated = datetime.now()
    
    def get_performance_summary(self) -> Dict:
        """获取性能总结"""
        summary = {
            "total_models": len(self.metrics),
            "models": []
        }
        
        for model_id, metrics in self.metrics.items():
            success_rate = (
                metrics.successful_requests / metrics.total_requests * 100
                if metrics.total_requests > 0 else 0
            )
            
            model_summary = {
                "model_id": model_id,
                "success_rate": round(success_rate, 2),
                "average_response_time": round(metrics.average_response_time, 3),
                "total_cost": round(metrics.cost, 4),
                "requests_per_hour": self._calculate_requests_per_hour(metrics)
            }
            
            summary["models"].append(model_summary)
        
        return summary

2. 成本分析仪表板：

class CostAnalyzer:
    def __init__(self, model_monitor):
        self.monitor = model_monitor
        self.cost_by_department = {}
        self.cost_trends = []
    
    def generate_cost_report(self, time_range="daily"):
        """生成成本报告"""
        report = {
            "time_range": time_range,
            "total_cost": 0,
            "cost_by_model": {},
            "cost_by_department": {},
            "cost_trends": [],
            "recommendations": []
        }
        
        for model_id, metrics in self.monitor.metrics.items():
            model_cost = metrics.cost
            report["total_cost"] += model_cost
            report["cost_by_model"][model_id] = model_cost
        
        # 成本优化建议
        report["recommendations"] = self._generate_cost_recommendations()
        
        return report
    
    def _generate_cost_recommendations(self):
        """生成成本优化建议"""
        recommendations = []
        
        # 分析高成本模型
        high_cost_models = [
            (model_id, metrics.cost) 
            for model_id, metrics in self.monitor.metrics.items()
            if metrics.cost > 100  # 假设100为高成本阈值
        ]
        
        if high_cost_models:
            recommendations.append({
                "type": "cost_optimization",
                "title": "高成本模型优化",
                "description": f"模型 {high_cost_models[0][0]} 成本较高，建议考虑使用更经济的替代模型",
                "potential_savings": self._calculate_potential_savings(high_cost_models[0][0])
            })
        
        return recommendations

3. 性能告警系统：

class AlertManager:
    def __init__(self, model_monitor):
        self.monitor = model_monitor
        self.alert_rules = {
            "high_error_rate": {"threshold": 0.05, "window": "5m"},
            "slow_response": {"threshold": 5.0, "window": "1m"},
            "cost_spike": {"threshold": 200, "window": "1h"},
            "model_unavailable": {"threshold": 0.0, "window": "2m"}
        }
        self.alert_history = []
    
    async def check_alerts(self):
        """检查告警条件"""
        current_time = datetime.now()
        
        for model_id, metrics in self.monitor.metrics.items():
            # 检查错误率
            if metrics.total_requests > 0:
                error_rate = metrics.failed_requests / metrics.total_requests
                if error_rate > self.alert_rules["high_error_rate"]["threshold"]:
                    await self._send_alert(
                        "high_error_rate",
                        f"模型 {model_id} 错误率过高: {error_rate:.2%}",
                        severity="warning"
                    )
            
            # 检查响应时间
            if metrics.average_response_time > self.alert_rules["slow_response"]["threshold"]:
                await self._send_alert(
                    "slow_response", 
                    f"模型 {model_id} 响应时间过慢: {metrics.average_response_time:.2f}s",
                    severity="warning"
                )
    
    async def _send_alert(self, alert_type: str, message: str, severity: str):
        """发送告警"""
        alert = {
            "type": alert_type,
            "message": message,
            "severity": severity,
            "timestamp": datetime.now(),
            "acknowledged": False
        }
        
        self.alert_history.append(alert)
        
        # 发送通知（邮件、微信、短信等）
        await self._notify_administrators(alert)
        
        # 记录到日志
        logger.warning(f"AI模型告警: {message}")

4. 可视化监控面板：

# 生成监控数据的API
from fastapi import FastAPI, APIRouter

monitoring_router = APIRouter(prefix="/api/monitoring")

@monitoring_router.get("/metrics")
async def get_metrics():
    """获取实时监控指标"""
    return model_monitor.get_performance_summary()

@monitoring_router.get("/alerts")
async def get_alerts():
    """获取当前告警"""
    return {
        "active_alerts": [
            alert for alert in alert_manager.alert_history 
            if not alert["acknowledged"]
        ],
        "total_alerts": len(alert_manager.alert_history)
    }

@monitoring_router.get("/cost-analysis")
async def get_cost_analysis():
    """获取成本分析"""
    return cost_analyzer.generate_cost_report()

@monitoring_router.get("/health-status")
async def get_health_status():
    """获取模型健康状态"""
    return {
        model_id: status 
        for model_id, status in health_checker.health_status.items()
    }

---

相关链接

• 模型管理功能说明
• 多模型配置教程
• MCP工具配置指南
• 最佳实践

---

需要技术支持？

我们的AI专家团队随时为您服务：

• 🔧 技术咨询：tech@tcfusion.ai
• 📊 性能优化：performance@tcfusion.ai
• 💰 成本优化：cost@tcfusion.ai
• 🚨 紧急支持：emergency@tcfusion.ai（7x24小时）