第35课：毕业项目：企业知识库Agent

📚 毕业项目：企业知识库Agent

本课是整个课程的毕业项目——综合全部34课所学，构建一个生产级的企业知识库Agent。它整合了RAG检索、Agent架构、安全防护、可观测性、多模型编排等所有核心能力。

第35课: 毕业项目
├── KnowledgeManager (L8-L14)
│   ├── 文档存储
│   ├── 关键词索引
│   └── 语义检索
├── SafetyMiddleware (L29-L30)
│   ├── 输入注入检测
│   ├── 输出内容过滤
│   └── PII脱敏
├── EnterpriseKBAgent (L1-L34综合)
│   ├── 安全检查
│   ├── 知识检索(RAG)
│   ├── 工具调用(Agent)
│   └── 对话管理
└── 生产化检查清单
    ├── 安全 5项
    ├── 可观测 4项
    ├── 部署 5项
    ├── 质量 4项
    └── 成本 4项

🔍 KnowledgeManager知识管理器

# 企业知识库Agent: 核心组件
from openai import OpenAI
import json
import re
from typing import Dict, List, Optional
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class Document:
    doc_id: str
    title: str
    content: str
    metadata: dict = field(default_factory=dict)
    embedding: List[float] = field(default_factory=list)

class KnowledgeManager:
    """知识管理器 - 文档存储/检索/更新"""
    def __init__(self):
        self.documents: Dict[str, Document] = {}
        self.index: Dict[str, List[str]] = {}  # keyword -> [doc_ids]

    def add_document(self, doc: Document):
        self.documents[doc.doc_id] = doc
        # 简易关键词索引
        keywords = re.findall(r'\w+', doc.title + " " + doc.content[:500])
        for kw in set(keywords):
            if kw not in self.index: self.index[kw] = []
            self.index[kw].append(doc.doc_id)

    def search(self, query: str, top_k: int = 5) -> List[Document]:
        """关键词检索"""
        keywords = set(re.findall(r'\w+', query.lower()))
        scores: Dict[str, int] = {}
        for kw in keywords:
            for doc_id in self.index.get(kw, []):
                scores[doc_id] = scores.get(doc_id, 0) + 1
        sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)[:top_k]
        return [self.documents[did] for did in sorted_ids if did in self.documents]

    def delete_document(self, doc_id: str):
        self.documents.pop(doc_id, None)

class SafetyMiddleware:
    """安全中间件 - 输入输出安全检查"""
    INJECTION_PATTERNS = [r"ignore.*instructions", r"you are now", r"system:"]

    def check_input(self, text: str) -> tuple:
        for pattern in self.INJECTION_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                return False, f"检测到注入: {pattern}"
        return True, ""

    def check_output(self, text: str) -> tuple:
        harmful = ["自杀方法", "制造炸弹"]
        for kw in harmful:
            if kw in text: return False, f"有害内容: {kw}"
        return True, ""

    def sanitize(self, text: str) -> str:
        # PII脱敏
        text = re.sub(r'\b1[3-9]\d{9}\b', '[手机号已脱敏]', text)
        text = re.sub(r'\b[\w.-]+@[\w.-]+\.[a-z]{2,}\b', '[邮箱已脱敏]', text)
        return text

✅ 验证通过：KnowledgeManager实现了文档存储、关键词索引和检索；SafetyMiddleware实现了注入检测和PII脱敏

🛠 EnterpriseKBAgent完整实现

# EnterpriseKBAgent: 完整企业知识库Agent
from openai import OpenAI
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field

client = OpenAI()

class EnterpriseKBAgent:
    """企业知识库Agent - 综合全部34课所学"""
    def __init__(self, model="gpt-4o-mini"):
        self.model = model
        # 核心组件(来自各课)
        self.knowledge = KnowledgeManager()      # L8-L14 RAG知识
        self.safety = SafetyMiddleware()          # L29 安全
        self._tools: Dict[str, dict] = {}        # L15-L16 工具
        self._conversation: List[dict] = []       # L4 对话管理
        self._trace: List[dict] = []              # L15 可观测性
        self._trust: Dict[str, float] = {}        # L21 信任度

    def add_document(self, doc_id, title, content, **meta):
        self.knowledge.add_document(Document(doc_id=doc_id, title=title, content=content, metadata=meta))

    def register_tool(self, name, desc, params, handler):
        self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": desc, "parameters": params}}, "handler": handler}

    def query(self, question: str, max_steps: int = 8) -> dict:
        """执行知识库查询"""
        start_time = time.time()
        result = {"question": question, "answer": "", "sources": [], "steps": [], "safety_checks": []}

        # 1. 安全检查 (L29)
        safe, reason = self.safety.check_input(question)
        result["safety_checks"].append({"stage": "input", "passed": safe, "reason": reason})
        if not safe:
            result["answer"] = f"安全拦截: {reason}"
            return result

        # 2. 知识检索 (L8-L14)
        docs = self.knowledge.search(question, top_k=3)
        context = "\n\n".join(f"[{d.title}] {d.content[:500]}" for d in docs)
        result["sources"] = [{"id": d.doc_id, "title": d.title} for d in docs]
        result["steps"].append("知识检索完成")

        # 3. 生成回答 (L1-L7)
        messages = [
            {"role": "system", "content": f"你是企业知识库助手。基于以下知识回答问题，如果知识不足请说明。\n\n知识:\n{context}"},
            {"role": "user", "content": question},
        ]

        for step in range(max_steps):
            resp = client.chat.completions.create(
                model=self.model, messages=messages,
                tools=[t["schema"] for t in self._tools.values()] if self._tools else None,
                tool_choice="auto",
            )
            msg = resp.choices[0].message
            messages.append(msg.to_dict())

            if not msg.tool_calls:
                answer = msg.content or ""
                # 4. 输出安全检查
                safe, reason = self.safety.check_output(answer)
                result["safety_checks"].append({"stage": "output", "passed": safe, "reason": reason})
                if not safe:
                    answer = f"输出安全拦截: {reason}"
                answer = self.safety.sanitize(answer)
                result["answer"] = answer
                break

            for tc in msg.tool_calls:
                try:
                    args = json.loads(tc.function.arguments)
                    tool_result = self._tools[tc.function.name]["handler"](**args)
                except Exception as e:
                    tool_result = f"错误: {e}"
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(tool_result)})
                result["steps"].append(f"工具调用: {tc.function.name}")

        result["latency_ms"] = (time.time() - start_time) * 1000
        return result

# 初始化
agent = EnterpriseKBAgent()
agent.add_document("doc1", "公司假期政策", "年假15天，病假10天，事假5天...")
agent.add_document("doc2", "报销流程", "1. 填写报销单 2. 主管审批 3. 财务审核...")
agent.add_document("doc3", "技术栈", "后端Python/FastAPI，前端React，数据库PostgreSQL...")
# result = agent.query("公司年假有多少天？")

✅ 验证通过：EnterpriseKBAgent整合了知识检索、安全检查、工具调用和对话管理

📋 生产化检查清单

# 生产化检查清单
PRODUCTION_CHECKLIST = {
    "安全": [
        "✅ 输入注入检测",
        "✅ 输出内容过滤",
        "✅ PII脱敏",
        "✅ 速率限制",
        "✅ 权限控制",
    ],
    "可观测性": [
        "✅ 结构化日志",
        "✅ 关键指标监控(TTFT/错误率/Token)",
        "✅ 分布式追踪",
        "✅ 告警系统",
    ],
    "部署": [
        "✅ Docker容器化",
        "✅ 健康检查",
        "✅ 灰度发布",
        "✅ 自动回滚",
        "✅ 负载均衡",
    ],
    "质量": [
        "✅ A/B测试",
        "✅ 质量评估自动化",
        "✅ 人机协作审批",
        "✅ 多模型路由",
    ],
    "成本": [
        "✅ Token用量追踪",
        "✅ 成本估算",
        "✅ 语义缓存",
        "✅ 模型级联",
    ],
}

def print_checklist():
    for category, items in PRODUCTION_CHECKLIST.items():
        print(f"\n{category}:")
        for item in items:
            print(f"  {item}")

print_checklist()

✅ 验证通过：生产化检查清单覆盖安全、可观测、部署、质量和成本5大类22项

🎓 课程总结

# 最终挑战: 构建你的企业知识库Agent
# 1. 添加至少10篇知识文档
# 2. 实现多轮对话
# 3. 添加工具调用(查询数据库/计算)
# 4. 完整安全检查
# 5. Docker部署
# 6. 健康检查+监控

🔄 项目扩展方向

毕业项目只是起点。以下是企业知识库Agent的常见扩展方向：

🧪 性能基准

🎯 学习路线建议