传统方式：单一模型处理所有请求
import openai

def answer_question(user_query: str):
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_query}]
    )
    return response.choices[0].message.content

 问题示例：查询企业内部的员工手册
result = answer_question("公司2025年的年假政策是什么？")
 结果：模型不知道（训练数据不包含企业内部信息），要么编造，要么说不知道

 RAG简化代码示例
from sentence_transformers import SentenceTransformer
import chromadb

class SimpleRAG:
    def __init__(self, llm, embed_model, vector_db):
        self.llm = llm               大语言模型
        self.embed = embed_model     嵌入模型
        self.db = vector_db          向量数据库
    
    def answer(self, query):
         1️⃣ 检索：将用户问题转换为向量，从知识库中检索相关内容
        query_vec = self.embed.encode(query)
        retrieved_docs = self.db.query(query_vec, top_k=3)
        
         2️⃣ 增强：构建包含检索结果的提示词
        context = "\n".join(retrieved_docs)
        prompt = f"基于以下资料回答问题：\n{context}\n问题：{query}"
        
         3️⃣ 生成：大模型基于检索内容生成答案
        return self.llm.generate(prompt)

 效果对比
 无RAG：“公司年假政策？” → “抱歉，我无法获取您公司的内部信息。”
 有RAG：“公司年假政策？” → 检索员工手册PDF → “根据公司2024年员工手册，年假政策为...”

 MCP架构的简单模拟
class MCPServer:
    """MCP服务器：暴露工具能力"""
    def __init__(self):
        self.tools = {}
    
    def register_tool(self, name, handler, description):
        self.tools[name] = {"handler": handler, "desc": description}
    
    def list_tools(self):
        return [{"name": k, "description": v["desc"]} for k, v in self.tools.items()]
    
    def call_tool(self, tool_name, params):
        return self.tools[tool_name]["handler"](params)

 示例：构建MCP工具服务器
mcp_server = MCPServer()
mcp_server.register_tool("web_search", web_search_api, "互联网信息")
mcp_server.register_tool("query_database", db_query, "查询企业内部数据库")
mcp_server.register_tool("send_email", email_sender, "发送邮件")

 智能体通过标准协议发现并调用工具
tools = mcp_server.list_tools()    发现阶段
result = mcp_server.call_tool("web_search", {"query": "2026 AI trends"})   调用阶段

 hybrid_ai_assistant.py - 混合AI助手简化实现
from typing import Dict, List
import asyncio

class HybridAIAssistant:
    """
    混合AI助手 - 支持智能路由、RAG检索、工具调用
    """
    
    def __init__(self, small_llm, large_llm, embed_model, vector_db):
        self.small_llm = small_llm     轻量模型（处理简单问题）
        self.large_llm = large_llm     重量模型（处理复杂问题）
        self.embed = embed_model        嵌入模型
        self.vector_db = vector_db      向量数据库
        self.tools = {}                 工具注册表（MCP风格）
    
    def register_tool(self, name: str, handler, description: str):
        """注册工具（模拟MCP Server）"""
        self.tools[name] = {"handler": handler, "desc": description}
    
    def classify_intent(self, query: str) -> str:
        """步骤1：意图识别 - 判断问题类型"""
         实际生产中可以用分类模型或LLM判断
        if "查询" in query and "文档" in query:
            return "knowledge_base"    需要检索知识库
        elif "" in query:
            return "web_search"        需要联网
        elif len(query) < 20:
            return "simple_qa"         简单问答
        else:
            return "complex_task"      复杂任务
    
    async def answer(self, user_query: str) -> str:
         步骤1：路由决策 - 根据意图选择处理路径
        intent = self.classify_intent(user_query)
        
        if intent == "simple_qa":
             简单问题：轻量模型直接回答
            return self.small_llm.generate(user_query)
        
        elif intent == "knowledge_base":
             知识库查询：RAG流程
            query_vec = self.embed.encode(user_query)
            docs = self.vector_db.query(query_vec, top_k=3)
            context = "\n".join(docs)
            prompt = f"基于以下资料回答：\n{context}\n问题：{user_query}"
            return self.large_llm.generate(prompt)
        
        elif intent == "web_search":
             联网：调用工具（MCP方式）
            search_result = self.tools["web_search"]["handler"](user_query)
            return self.large_llm.generate(f"结果：{search_result}\n问题：{user_query}")
        
        elif intent == "complex_task":
             复杂任务：多智能体协作（简化为多步处理）
             步骤2：任务分解
            sub_tasks = self._decompose_task(user_query)
            results = []
            for task in sub_tasks:
                 步骤3：子任务路由 + 执行
                result = await self.answer(task)
                results.append(result)
             步骤4：结果聚合
            return self.large_llm.generate(f"请综合以下子任务结果：{results}")
    
    def _decompose_task(self, task: str) -> List[str]:
        """任务分解（演示逻辑）"""
         实际生产中可以用LLM进行思维链分解
        return [task]   简化版直接返回原任务

 使用示例
async def main():
     初始化助手（参数简化）
    assistant = HybridAIAssistant(small_llm, large_llm, embed_model, vector_db)
    
     注册MCP工具
    assistant.register_tool("web_search", search_api, "互联网")
    
     测试不同类型的问题
    print(await assistant.answer("你好，今天天气不错"))         simple_qa → 小模型
    print(await assistant.answer("查询公司年假政策"))           knowledge_base → RAG
    print(await assistant.answer("2026年AI趋势"))           web_search → 工具调用