第38课：知识库Agent

📚 知识库Agent：企业知识的大脑

企业知识库Agent让组织内部的知识可搜索、可问答、可更新。它是RAG技术在企业场景的深度应用，需要考虑文档管理、权限控制、知识更新等生产级需求。

📖 企业知识库架构

企业知识库Agent
├── 知识采集
│   ├── 文档上传
│   ├── 网页抓取
│   ├── 数据库同步
│   └── API接入
├── 知识处理
│   ├── 文档解析
│   ├── 智能切分
│   ├── 实体提取
│   └── 向量化
├── 知识检索
│   ├── 混合搜索（向量+关键词）
│   ├── 语义重排序
│   ├── 权限过滤
│   └── 多轮追问
├── 知识问答
│   ├── 引用来源
│   ├── 多文档整合
│   ├── 表格/图表理解
│   └── 不确定性声明
└── 知识管理
    ├── 版本管理
    ├── 知识审核
    ├── 过期检测
    └── 使用统计

💻 代码实现：企业知识库Agent

# 企业知识库Agent
import json, re, math, hashlib, time
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field

@dataclass
class KnowledgeEntry:
    id: str
    content: str
    source: str
    category: str
    tags: List[str]
    access_level: str = "public"  # public, internal, confidential
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

class KnowledgeStore:
    # 知识存储
    def __init__(self):
        self.entries: Dict[str, KnowledgeEntry] = {}
        self.embeddings: Dict[str, List[float]] = {}
    
    def add(self, entry: KnowledgeEntry):
        self.entries[entry.id] = entry
        self.embeddings[entry.id] = self._embed(entry.content)
    
    def search(self, query, top_k=5, access_level="public", categories=None):
        q_emb = self._embed(query)
        scores = []
        for eid, emb in self.embeddings.items():
            entry = self.entries[eid]
            # 权限过滤
            levels = ["public", "internal", "confidential"]
            if levels.index(entry.access_level) > levels.index(access_level):
                continue
            # 分类过滤
            if categories and entry.category not in categories:
                continue
            sim = self._cosine_sim(q_emb, emb)
            scores.append((sim, eid))
        scores.sort(reverse=True)
        return [(s, self.entries[eid]) for s, eid in scores[:top_k]]
    
    def _embed(self, text):
        h = hashlib.md5(text.encode()).hexdigest()
        return [int(h[i:i+2], 16) / 255.0 for i in range(0, 64, 2)]
    
    def _cosine_sim(self, a, b):
        dot = sum(x * y for x, y in zip(a, b))
        na = math.sqrt(sum(x**2 for x in a)) or 1e-10
        nb = math.sqrt(sum(x**2 for x in b)) or 1e-10
        return dot / (na * nb)

class KnowledgeBaseAgent:
    # 知识库Agent
    def __init__(self):
        self.store = KnowledgeStore()
        self.qa_history = []
    
    def ingest(self, content, source, category, tags, access_level="public"):
        eid = hashlib.md5(f"{source}:{content[:50]}".encode()).hexdigest()[:8]
        entry = KnowledgeEntry(eid, content, source, category, tags, access_level)
        self.store.add(entry)
        return eid
    
    def ask(self, question, access_level="public", categories=None):
        results = self.store.search(question, top_k=3, access_level=access_level, categories=categories)
        
        if not results:
            return {"answer": "抱歉，在知识库中未找到相关信息。", "sources": [], "confidence": 0}
        
        # 整合结果
        sources = []
        context_parts = []
        for score, entry in results:
            sources.append({"source": entry.source, "category": entry.category, "relevance": f"{score:.2f}"})
            context_parts.append(f"[{entry.source}] {entry.content[:100]}")
        
        answer = f"根据知识库信息：\n\n" + "\n\n".join(context_parts)
        confidence = min(results[0][0] * 1.5, 1.0)  # 简化的置信度
        
        self.qa_history.append({"question": question, "confidence": confidence, "sources": len(sources)})
        return {"answer": answer, "sources": sources, "confidence": f"{confidence:.0%}"}
    
    def get_stats(self):
        categories = {}
        for e in self.store.entries.values():
            categories[e.category] = categories.get(e.category, 0) + 1
        return {"total_entries": len(self.store.entries), "categories": categories, "qa_count": len(self.qa_history)}

# 测试
kb = KnowledgeBaseAgent()

# 导入知识
kb.ingest("Python由Guido van Rossum于1991年创建，是高级编程语言。", "Python百科", "programming", ["python","语言"], "public")
kb.ingest("公司技术栈选型：后端使用Python FastAPI，前端React。", "技术规范", "engineering", ["技术栈","规范"], "internal")
kb.ingest("Q3营收目标1000万，已完成85%。", "季度报告", "business", ["营收","目标"], "confidential")
kb.ingest("AI Agent系统架构：感知→决策→执行→记忆。", "AI设计文档", "engineering", ["AI","架构"], "internal")
kb.ingest("公司假期政策：年假15天起，工龄每增1年加1天。", "HR手册", "hr", ["假期","政策"], "public")

# 问答
for q, level in [("Python是什么？", "public"), ("公司技术栈？", "internal"), ("营收情况？", "confidential")]:
    result = kb.ask(q, access_level=level)
    print(f"\n❓ {q} (权限:{level})")
    print(f"🤖 {result['answer'][:100]}...")
    print(f"📊 置信度:{result['confidence']}, 来源:{len(result['sources'])}个")

print(f"\n📈 知识库统计: {kb.get_stats()}")

🏋️ 实战练习

深入理解：知识库Agent核心原理

知识库Agent三层架构：交互层(问答/检索/推荐/浏览)、智能层(意图理解/检索增强/推理)、数据层(文档库/向量库/知识图谱)。数据层组件：文档库(MySQL/对象存储)、向量库(Milvus/Pinecone)、知识图谱(Neo4j)、缓存层(Redis)。知识库质量保障：入库质检(完整性/格式/重复)、检索评估(Precision@5/Recall@10/MRR)、回答评估(Faithfulness/Relevancy/Hallucination Rate)。

进阶实现：知识库Agent

以下是针对知识库Agent主题的进阶实现，包含文档管理+语义检索+问答+推荐等核心功能。代码经过实机运行验证。

# KnowledgeBaseAgent - 知识库Agent进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class KnowledgeBaseAgent:
    # 知识库Agent进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = KnowledgeBaseAgent({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

知识库Agent最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：知识库Agent的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。