2026 AI Agent 知识管理：RAG vs 向量数据库 vs MCP 工具生态——架构对比与实战指南

from dataclasses import dataclass, field
from typing import List, Optional
import hashlib
import numpy as np
from collections import defaultdict

# ========== 1. 文档加载与分块 ==========

@dataclass
class Document:
    id: str
    content: str
    source: str
    metadata: dict = field(default_factory=dict)

class TextChunker:
    """智能文本分块器：按语义边界切分，避免截断关键信息"""

    def __init__(self, chunk_size: int = 500, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def split(self, text: str) -> List[str]:
        # 先按段落分割
        paragraphs = text.split("\n\n")
        chunks = []
        current_chunk = ""

        for para in paragraphs:
            if len(current_chunk) + len(para) > self.chunk_size:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                # 保留重叠部分
                if self.overlap > 0 and chunks:
                    last_words = chunks[-1].split()[-self.overlap//6:]
                    current_chunk = " ".join(last_words) + " " + para
                else:
                    current_chunk = para
            else:
                current_chunk = current_chunk + "\n\n" + para if current_chunk else para

        if current_chunk.strip():
            chunks.append(current_chunk.strip())
        return chunks

# ========== 2. 本地向量化（无需外部API） ==========

class LocalEmbedder:
    """轻量级本地嵌入器（模拟实现，实际使用 sentence-transformers）"""

    def __init__(self, dim: int = 384):
        self.dim = dim
        # 生产环境: from sentence_transformers import SentenceTransformer
        # self.model = SentenceTransformer('all-MiniLM-L6-v2')

    def encode(self, text: str) -> List[float]:
        # 模拟嵌入：实际场景中替换为真实模型推理
        seed = int(hashlib.md5(text.encode()).hexdigest()[:8], 16)
        rng = np.random.RandomState(seed)
        vec = rng.randn(self.dim).tolist()
        norm = sum(v*v for v in vec) ** 0.5
        return [v / norm for v in vec]

# ========== 3. 向量检索（余弦相似度） ==========

class VectorRetriever:
    """基于余弦相似度的向量检索器"""

    def __init__(self):
        self.vectors: List[List[float]] = []
        self.documents: List[Document] = []
        self.embedder = LocalEmbedder()

    def add(self, doc: Document):
        self.documents.append(doc)
        self.vectors.append(self.embedder.encode(doc.content))

    def search(self, query: str, top_k: int = 3) -> List[tuple]:
        query_vec = self.embedder.encode(query)
        scores = []
        for i, vec in enumerate(self.vectors):
            # 余弦相似度
            dot = sum(a*b for a, b in zip(query_vec, vec))
            norm_q = sum(v*v for v in query_vec) ** 0.5
            norm_v = sum(v*v for v in vec) ** 0.5
            similarity = dot / (norm_q * norm_v) if norm_q * norm_v > 0 else 0
            scores.append((similarity, self.documents[i]))
        scores.sort(key=lambda x: x[0], reverse=True)
        return scores[:top_k]

# ========== 4. RAG 管道组装 ==========

class RAGPipeline:
    """完整的 RAG 问答管道"""

    def __init__(self):
        self.chunker = TextChunker(chunk_size=500, overlap=80)
        self.retriever = VectorRetriever()

    def ingest(self, doc: Document):
        """加载并索引文档"""
        chunks = self.chunker.split(doc.content)
        for i, chunk in enumerate(chunks):
            chunk_doc = Document(
                id=f"{doc.id}_chunk_{i}",
                content=chunk,
                source=doc.source,
                metadata={**doc.metadata, "chunk_index": i, "total_chunks": len(chunks)}
            )
            self.retriever.add(chunk_doc)
        print(f"  ✓ 索引完成: {len(chunks)} 个分块")

    def query(self, question: str) -> dict:
        """执行 RAG 查询"""
        results = self.retriever.search(question, top_k=3)
        context = "\n\n---\n\n".join(
            f"[来源: {doc.source}]\n{doc.content}"
            for score, doc in results
        )
        return {
            "question": question,
            "context": context,
            "sources": [doc.source for _, doc in results],
            "scores": [round(score, 3) for score, _ in results]
        }

# ========== 5. 使用示例 ==========

if __name__ == "__main__":
    rag = RAGPipeline()

    # 加载示例文档
    docs = [
        Document(
            id="doc1", source="Python教程",
            content="Python的装饰器是一种高级函数，它接受一个函数作为输入并返回一个新函数。"
                    "装饰器使用@语法糖，可以在不修改原函数代码的情况下增强其功能。"
                    "常见的内置装饰器包括@staticmethod、@classmethod和@property。"
                    "装饰器的核心原理是闭包和高阶函数。"
        ),
        Document(
            id="doc2", source="AI基础",
            content="Transformer架构是2017年提出的神经网络架构，它完全基于自注意力机制。"
                    "与RNN不同，Transformer可以并行处理整个序列，大大提高了训练速度。"
                    "GPT和BERT都是基于Transformer的变体。"
                    "注意力机制允许模型在处理序列时关注最相关的部分。"
        ),
        Document(
            id="doc3", source="机器学习",
            content="梯度下降是一种优化算法，用于最小化损失函数。"
                    "它通过计算损失函数关于参数的梯度，并沿梯度反方向更新参数。"
                    "学习率是梯度下降中最关键的超参数，过大可能导致发散，"
                    "过小则收敛缓慢。Adam优化器结合了动量和自适应学习率的优点。"
        ),
    ]

    for doc in docs:
        rag.ingest(doc)

    # 查询示例
    result = rag.query("Python如何在不修改原函数的情况下增强功能？")
    print(f"\n问题: {result['question']}")
    print(f"相关来源: {result['sources']}")
    print(f"相似度: {result['scores']}")
    print(f"\n上下文:\n{result['context']}")

from dataclasses import dataclass, field
from typing import List, Optional, Callable
from enum import Enum
import json
import time

# ========== 1. 查询策略定义 ==========

class QueryStrategy(Enum):
    SEMANTIC = "语义搜索"
    FILTER = "元数据过滤"
    HYBRID = "混合搜索"
    MULTIHOP = "多跳检索"

@dataclass
class QueryPlan:
    """Agent 生成的查询计划"""
    strategy: QueryStrategy
    query_text: str
    filters: dict = field(default_factory=dict)
    top_k: int = 5
    follow_ups: List[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "strategy": self.strategy.value,
            "query": self.query_text,
            "filters": self.filters,
            "top_k": self.top_k,
            "follow_ups": self.follow_ups,
        }

# ========== 2. 轻量级向量数据库（带过滤能力） ==========

class LightVectorDB:
    """模拟向量数据库，支持语义搜索 + 元数据过滤"""

    def __init__(self):
        self.entries: List[dict] = []

    def insert(self, vector: list, metadata: dict, content: str):
        self.entries.append({
            "vector": vector,
            "metadata": metadata,
            "content": content,
            "created_at": time.time(),
        })

    def search(self, query_vector: list, strategy: QueryStrategy,
               filters: dict = None, top_k: int = 5) -> List[dict]:
        # 过滤
        candidates = self.entries
        if filters and strategy in (QueryStrategy.FILTER, QueryStrategy.HYBRID):
            for key, value in filters.items():
                candidates = [e for e in candidates
                              if e["metadata"].get(key) == value]

        # 语义评分
        scored = []
        for entry in candidates:
            dot = sum(a*b for a, b in zip(query_vector, entry["vector"]))
            nq = sum(v*v for v in query_vector) ** 0.5
            ne = sum(v*v for v in entry["vector"]) ** 0.5
            score = dot / (nq * ne) if nq * ne > 0 else 0
            scored.append((score, entry))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [{"score": round(s, 3), **e} for s, e in scored[:top_k]]

    def bulk_insert(self, items: List[dict]):
        for item in items:
            self.insert(item["vector"], item["metadata"], item["content"])

# ========== 3. Agent 查询规划器 ==========

class AgentQueryPlanner:
    """模拟 Agent 自主生成查询策略"""

    def __init__(self, db: LightVectorDB):
        self.db = db

    def plan_query(self, user_question: str) -> QueryPlan:
        """根据问题类型自动生成查询计划（模拟 LLM 推理）"""
        # 关键词启发式策略选择（实际场景由 LLM 决策）
        lower_q = user_question.lower()

        if any(k in lower_q for k in ["对比", "比较", "区别", "vs"]):
            return QueryPlan(
                strategy=QueryStrategy.HYBRID,
                query_text=user_question,
                filters={},
                top_k=6,
                follow_ups=["分别查找每个概念的定义", "查找两者的对比分析"]
            )
        elif any(k in lower_q for k in ["所有", "列表", "有哪些"]):
            return QueryPlan(
                strategy=QueryStrategy.FILTER,
                query_text=user_question,
                filters={"type": "overview"},
                top_k=10
            )
        elif any(k in lower_q for k in ["为什么", "如何", "怎么"]):
            return QueryPlan(
                strategy=QueryStrategy.MULTIHOP,
                query_text=user_question,
                top_k=3,
                follow_ups=["查找基础概念定义", "查找相关案例"]
            )
        else:
            return QueryPlan(
                strategy=QueryStrategy.SEMANTIC,
                query_text=user_question,
                top_k=3
            )

    def execute(self, plan: QueryPlan, embedder) -> dict:
        """执行查询计划，支持多跳检索"""
        results = []

        # 第一轮检索
        first_results = self.db.search(
            embedder.encode(plan.query_text),
            plan.strategy,
            plan.filters,
            plan.top_k
        )
        results.extend(first_results)

        # 多跳：使用第一轮结果触发第二轮检索
        if plan.strategy == QueryStrategy.MULTIHOP and plan.follow_ups:
            for follow_up in plan.follow_ups:
                follow_results = self.db.search(
                    embedder.encode(follow_up),
                    QueryStrategy.SEMANTIC,
                    top_k=2
                )
                results.extend(follow_results)

        # 去重并排序
        seen = set()
        unique_results = []
        for r in results:
            content_key = r.get("content", "")[:50]
            if content_key not in seen:
                seen.add(content_key)
                unique_results.append(r)

        return {
            "plan": plan.to_dict(),
            "results": unique_results[:plan.top_k],
            "total_found": len(unique_results),
        }

# ========== 4. 使用示例 ==========

if __name__ == "__main__":
    import numpy as np

    db = LightVectorDB()
    planner = AgentQueryPlanner(db)
    embedder = LocalEmbedder()

    # 模拟嵌入（简化）
    def quick_embed(text: str) -> list:
        import hashlib
        seed = int(hashlib.md5(text.encode()).hexdigest()[:8], 16)
        rng = np.random.RandomState(seed)
        vec = rng.randn(384).tolist()
        norm = sum(v*v for v in vec) ** 0.5
        return [v / norm for v in vec]

    # 插入测试数据
    test_data = [
        {"vector": quick_embed("Python装饰器"), "metadata": {"type": "concept", "lang": "python"}, "content": "Python装饰器是高阶函数..."},
        {"vector": quick_embed("Transformer注意力"), "metadata": {"type": "concept", "domain": "nlp"}, "content": "Transformer的自注意力机制..."},
        {"vector": quick_embed("梯度下降优化"), "metadata": {"type": "concept", "domain": "ml"}, "content": "梯度下降用于最小化损失..."},
        {"vector": quick_embed("RAG vs 微调对比"), "metadata": {"type": "comparison", "domain": "llm"}, "content": "RAG通过检索增强上下文..."},
    ]
    db.bulk_insert(test_data)

    # Agent 自主查询
    question = "RAG和微调有什么区别？"
    plan = planner.plan_query(question)
    result = planner.execute(plan, embedder)
    print(f"查询策略: {plan.strategy.value}")
    print(f"找到 {result['total_found']} 条结果")
    for r in result['results']:
        print(f"  [{r['score']}] {r['content']}")

from dataclasses import dataclass, field
from typing import List, Optional, Any, Callable
from abc import ABC, abstractmethod
import json
import time

# ========== 1. MCP 协议核心定义 ==========

@dataclass
class MCPResource:
    """MCP 资源描述"""
    uri: str
    name: str
    description: str
    mime_type: str = "text/plain"

@dataclass
class MCPTool:
    """MCP 工具描述"""
    name: str
    description: str
    parameters: dict  # JSON Schema
    handler: Callable  # 实际执行函数

@dataclass
class MCPToolCall:
    name: str
    arguments: dict

@dataclass
class MCPToolResult:
    content: List[dict]
    is_error: bool = False

# ========== 2. MCP Server 实现 ==========

class MCPServer:
    """MCP Server：注册并提供工具和资源"""

    def __init__(self, name: str):
        self.name = name
        self.tools: dict[str, MCPTool] = {}
        self.resources: dict[str, MCPResource] = {}

    def register_tool(self, tool: MCPTool):
        self.tools[tool.name] = tool

    def register_resource(self, resource: MCPResource):
        self.resources[resource.uri] = resource

    def list_tools(self) -> List[dict]:
        return [
            {"name": t.name, "description": t.description,
             "parameters": t.parameters}
            for t in self.tools.values()
        ]

    def call_tool(self, name: str, args: dict) -> MCPToolResult:
        if name not in self.tools:
            return MCPToolResult(
                content=[{"type": "text", "text": f"工具 {name} 不存在"}],
                is_error=True
            )
        try:
            result = self.tools[name].handler(**args)
            return MCPToolResult(
                content=[{"type": "text", "text": str(result)}]
            )
        except Exception as e:
            return MCPToolResult(
                content=[{"type": "text", "text": f"错误: {str(e)}"}],
                is_error=True
            )

# ========== 3. 构建知识管理工具集 ==========

def create_knowledge_mcp_servers() -> List[MCPServer]:
    """创建 MCP 知识管理工具服务器集群"""

    # --- Server 1: 文档搜索 ---
    doc_server = MCPServer("document-search")
    documents_db = {
        "python-decorators": "Python装饰器使用@语法，在不修改原函数的情况下增强功能...",
        "transformer-attention": "Transformer的自注意力机制允许模型并行处理序列...",
        "rag-architecture": "RAG通过检索相关文档片段，将其作为上下文提供给LLM...",
    }

    def search_documents(query: str, limit: int = 3) -> str:
        """在文档库中搜索"""
        results = []
        for key, content in documents_db.items():
            if any(word in content.lower() for word in query.lower().split()):
                results.append(f"[{key}] {content}")
        return "\n".join(results[:limit]) if results else "未找到相关文档"

    doc_server.register_tool(MCPTool(
        name="search_documents",
        description="在技术文档库中搜索",
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"},
                "limit": {"type": "integer", "description": "返回数量"}
            },
            "required": ["query"]
        },
        handler=search_documents
    ))

    # --- Server 2: 代码示例检索 ---
    code_server = MCPServer("code-examples")
    code_db = {
        "decorator_pattern": '''def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__}: {time.time()-start:.3f}s")
        return result
    return wrapper''',
        "attention_mechanism": '''def scaled_dot_product_attention(Q, K, V, mask=None):
    d_k = Q.size(-1)
    scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k)
    if mask is not None:
        scores = scores.masked_fill(mask == 0, -1e9)
    attention_weights = F.softmax(scores, dim=-1)
    return torch.matmul(attention_weights, V)''',
    }

    def get_code_example(name: str) -> str:
        """获取代码示例"""
        return code_db.get(name, f"未找到代码示例: {name}")

    code_server.register_tool(MCPTool(
        name="get_code_example",
        description="获取特定技术主题的可运行代码示例",
        parameters={
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "代码示例名称"}
            },
            "required": ["name"]
        },
        handler=get_code_example
    ))

    # --- Server 3: 知识图谱查询 ---
    graph_server = MCPServer("knowledge-graph")
    concept_relations = {
        "decorator": ["高阶函数", "闭包", "Python语法"],
        "attention": ["Transformer", "序列模型", "并行计算"],
        "rag": ["向量数据库", "embedding", "prompt工程"],
    }

    def query_relations(concept: str) -> str:
        """查询概念关联的知识"""
        relations = concept_relations.get(concept, [])
        if relations:
            return f"{concept} 关联概念: {', '.join(relations)}"
        return f"未找到 {concept} 的关联概念"

    graph_server.register_tool(MCPTool(
        name="query_knowledge_graph",
        description="查询概念之间的关联关系",
        parameters={
            "type": "object",
            "properties": {
                "concept": {"type": "string", "description": "要查询的概念"}
            },
            "required": ["concept"]
        },
        handler=query_relations
    ))

    return [doc_server, code_server, graph_server]

# ========== 4. MCP Router（Agent 的工具调度中心） ==========

class MCPRouter:
    """MCP 路由器：根据 Agent 意图选择并调用合适的 MCP 工具"""

    def __init__(self, servers: List[MCPServer]):
        self.servers = servers
        self.tool_catalog: dict[str, MCPServer] = {}
        for server in servers:
            for tool_name in server.tools:
                self.tool_catalog[tool_name] = server

    def discover_tools(self) -> List[dict]:
        """发现所有可用工具"""
        tools = []
        for server in self.servers:
            for tool in server.list_tools():
                tools.append({**tool, "server": server.name})
        return tools

    def route_and_execute(self, tool_calls: List[MCPToolCall]) -> List[MCPToolResult]:
        """路由并执行工具调用（支持并行）"""
        results = []
        for call in tool_calls:
            server = self.tool_catalog.get(call.name)
            if server:
                result = server.call_tool(call.name, call.arguments)
                results.append(result)
            else:
                results.append(MCPToolResult(
                    content=[{"type": "text", "text": f"未知工具: {call.name}"}],
                    is_error=True
                ))
        return results

# ========== 5. 使用示例 ==========

if __name__ == "__main__":
    servers = create_knowledge_mcp_servers()
    router = MCPRouter(servers)

    print("=== 可用的 MCP 工具 ===")
    for tool in router.discover_tools():
        print(f"  [{tool['server']}] {tool['name']}: {tool['description']}")

    print("\n=== 模拟 Agent 调用 ===")
    calls = [
        MCPToolCall("search_documents", {"query": "Python 装饰器", "limit": 2}),
        MCPToolCall("get_code_example", {"name": "decorator_pattern"}),
        MCPToolCall("query_knowledge_graph", {"concept": "attention"}),
    ]

    for call in calls:
        results = router.route_and_execute([call])
        for r in results:
            status = "✅" if not r.is_error else "❌"
            print(f"{status} {call.name}: {r.content[0]['text'][:100]}...")

from dataclasses import dataclass, field
from typing import List, Optional
import time

@dataclass
class KnowledgeQuery:
    source: str  # "rag" | "vector_db" | "mcp_tool" | "knowledge_graph"
    query: str
    filters: dict = field(default_factory=dict)

@dataclass
class KnowledgeResult:
    source: str
    content: str
    confidence: float  # 0-1
    metadata: dict = field(default_factory=dict)

class HybridKnowledgeSystem:
    """混合知识管理系统：整合 RAG + 向量DB + MCP"""

    def __init__(self):
        self.rag_pipeline = None      # RAGPipeline
        self.vector_db = None         # LightVectorDB
        self.mcp_router = None        # MCPRouter
        self._initialized = False

    def initialize(self, rag, vdb, router):
        """注入依赖"""
        self.rag_pipeline = rag
        self.vector_db = vdb
        self.mcp_router = router
        self._initialized = True

    def analyze_intent(self, question: str) -> List[KnowledgeQuery]:
        """分析问题意图，生成多路查询（模拟 LLM 推理）"""
        queries = []
        lower_q = question.lower()

        # 策略：总是先尝试 RAG（非结构化文档检索）
        queries.append(KnowledgeQuery(
            source="rag", query=question
        ))

        # 如果问题涉及概念关系，添加知识图谱查询
        if any(k in lower_q for k in ["关系", "关联", "区别", "对比"]):
            queries.append(KnowledgeQuery(
                source="knowledge_graph", query=question
            ))

        # 如果问题涉及具体操作，添加 MCP 工具查询
        if any(k in lower_q for k in ["代码", "示例", "怎么实现"]):
            queries.append(KnowledgeQuery(
                source="mcp_tool", query=question,
                filters={"tool_type": "code_example"}
            ))

        # 默认添加向量数据库直连作为补充
        queries.append(KnowledgeQuery(
            source="vector_db", query=question
        ))

        return queries

    def query(self, question: str) -> dict:
        """执行混合知识查询"""
        start = time.time()
        queries = self.analyze_intent(question)
        results = []

        for q in queries:
            try:
                if q.source == "rag" and self.rag_pipeline:
                    rag_result = self.rag_pipeline.query(q.query)
                    results.append(KnowledgeResult(
                        source="rag",
                        content=rag_result["context"][:500],
                        confidence=max(rag_result["scores"], default=0),
                        metadata={"sources": rag_result.get("sources", [])}
                    ))
                elif q.source == "vector_db" and self.vector_db:
                    # 由 Agent 规划查询
                    # （此处简化为直接语义搜索）
                    results.append(KnowledgeResult(
                        source="vector_db",
                        content="向量数据库检索结果（详见方案二实现）",
                        confidence=0.75,
                        metadata={"strategy": "semantic"}
                    ))
                elif q.source == "knowledge_graph" and self.mcp_router:
                    results.append(KnowledgeResult(
                        source="knowledge_graph",
                        content="知识图谱关联查询结果",
                        confidence=0.80,
                        metadata={"type": "relations"}
                    ))
            except Exception as e:
                results.append(KnowledgeResult(
                    source=q.source,
                    content=f"查询失败: {str(e)}",
                    confidence=0.0,
                    metadata={"error": str(e)}
                ))

        # 结果融合与排序
        results.sort(key=lambda r: r.confidence, reverse=True)
        elapsed = time.time() - start

        return {
            "question": question,
            "queries_planned": len(queries),
            "results": [
                {"source": r.source, "confidence": r.confidence,
                 "preview": r.content[:100]}
                for r in results
            ],
            "total_time_ms": round(elapsed * 1000, 1),
        }

# ========== 使用示例 ==========

if __name__ == "__main__":
    system = HybridKnowledgeSystem()
    # 实际使用时需要 initialize(rag, vdb, router)

    question = "Python 装饰器和 Rust 宏有什么区别？它们分别怎么实现？"
    queries = system.analyze_intent(question)
    print(f"问题: {question}")
    print(f"计划查询 {len(queries)} 个知识源:")
    for q in queries:
        print(f"  - {q.source}: {q.query}")