📚 项目1：智能问答知识库 Agent

难度: ⭐⭐⭐ Claude Code ⏱ 预计 4-5 小时

1. 需求分析

业务场景

小明有一个本地文件夹，存放了大量技术文档、学习笔记、项目报告（.txt/.md/.pdf）。每次查找信息都要逐个打开文件搜索，效率极低。他希望有一个 AI 助手，能"理解"所有文档内容，用自然语言提问就能得到精准答案。

核心需求

• 📥 支持本地文档批量导入（txt, md, pdf文本提取）
• 🔍 基于文档内容进行智能问答（不是关键词搜索，是语义理解）
• 📝 回答时引用来源文档，消除幻觉
• 💾 记住之前的问答，支持追问
• 📊 显示答案的可信度/相关度

为什么不用普通搜索？

普通搜索（Ctrl+F / grep）只能做关键词匹配。你搜"怎么提高性能"，文档里写的是"优化响应速度"，关键词不匹配就找不到。AI Agent 能做语义搜索——理解你的意图，找到意思相近的内容。

2. 架构设计

这是经典的 RAG（Retrieval-Augmented Generation）架构：检索增强生成。先把文档"消化"（向量化存库），用户提问时"检索"最相关的内容片段，然后"喂"给大模型让其基于真实文档回答——从而减少幻觉。

3. 完整代码

创建 qa_agent.py：

#!/usr/bin/env python3
"""
QAAgent — 智能问答知识库 Agent
================================
基于 RAG 架构：文档导入→向量化→语义检索→增强回答

依赖安装：
  pip install anthropic python-dotenv numpy sentence-transformers

运行：python3 qa_agent.py
"""

import json
import os
import re
import math
from datetime import datetime
from pathlib import Path
from typing import Generator

import numpy as np
from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()

# ================================================================
# 第1部分：文本处理与分块
# ================================================================

class DocumentProcessor:
    """文档加载、清洗、分块"""

    SUPPORTED_EXTENSIONS = {".txt", ".md", ".py", ".json", ".csv", ".log", ".rst"}

    @staticmethod
    def load_file(filepath: Path) -> str | None:
        """加载单个文件"""
        try:
            content = filepath.read_text("utf-8")
            # 基本清洗：移除多余空行
            content = re.sub(r'\n{3,}', '\n\n', content)
            return content
        except UnicodeDecodeError:
            print(f"  ⚠️ 跳过二进制文件: {filepath.name}")
            return None
        except Exception as e:
            print(f"  ⚠️ 读取失败: {filepath.name} — {e}")
            return None

    @staticmethod
    def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[dict]:
        """
        将文本分割为重叠的块
        - chunk_size: 每块最大字符数
        - overlap: 相邻块之间的重叠字符数（保证上下文不丢失）
        """
        chunks = []
        start = 0
        chunk_id = 0

        while start < len(text):
            end = start + chunk_size
            chunk_content = text[start:end]

            # 尽量按句子边界分割（找最后一个句号/换行）
            if end < len(text):
                last_period = max(
                    chunk_content.rfind("。"),
                    chunk_content.rfind("\n"),
                    chunk_content.rfind(". "),
                )
                if last_period > chunk_size * 0.5:
                    end = start + last_period + 1
                    chunk_content = text[start:end]

            chunks.append({
                "id": chunk_id,
                "content": chunk_content.strip(),
                "start": start,
                "end": end,
                "length": len(chunk_content),
            })

            start = end - overlap
            chunk_id += 1

        return chunks

    def process_directory(self, directory: Path) -> list[dict]:
        """处理整个目录，返回所有文档块"""
        all_chunks = []
        file_count = 0

        for filepath in directory.rglob("*"):
            if not filepath.is_file():
                continue
            if filepath.name.startswith("."):
                continue
            if filepath.suffix.lower() not in self.SUPPORTED_EXTENSIONS:
                continue

            content = self.load_file(filepath)
            if content is None:
                continue

            chunks = self.chunk_text(content)
            for chunk in chunks:
                chunk["source_file"] = str(filepath.relative_to(directory))
            all_chunks.extend(chunks)
            file_count += 1
            print(f"  ✅ {filepath.name}: {len(chunks)} 块")

        print(f"\n📊 共处理 {file_count} 个文件，生成 {len(all_chunks)} 个文本块")
        return all_chunks


# ================================================================
# 第2部分：向量化与检索
# ================================================================

class SimpleVectorStore:
    """
    简易向量存储（基于 NumPy + 余弦相似度）
    学习用，生产环境请用 FAISS/Chroma/Milvus
    """

    def __init__(self):
        self.vectors: np.ndarray | None = None
        self.chunks: list[dict] = []
        self.embedding_dim = 384  # sentence-transformers 默认维度

    def add(self, chunks: list[dict], embeddings: np.ndarray):
        """批量添加向量和对应的文档块"""
        start_idx = len(self.chunks)
        for i, chunk in enumerate(chunks):
            chunk["vector_idx"] = start_idx + i
        self.chunks.extend(chunks)

        if self.vectors is None:
            self.vectors = embeddings
        else:
            self.vectors = np.vstack([self.vectors, embeddings])

    def search(self, query_embedding: np.ndarray, top_k: int = 5) -> list[dict]:
        """余弦相似度搜索，返回最相关的 top_k 个文档块"""
        if self.vectors is None or len(self.vectors) == 0:
            return []

        # 计算余弦相似度: cos(a,b) = dot(a,b) / (|a| * |b|)
        dot_product = np.dot(self.vectors, query_embedding.T).flatten()
        norms = np.linalg.norm(self.vectors, axis=1) * np.linalg.norm(query_embedding)
        similarities = dot_product / (norms + 1e-8)

        # 取 top_k
        top_indices = np.argsort(similarities)[::-1][:top_k]

        results = []
        for idx in top_indices:
            if similarities[idx] > 0.1:  # 过滤低相关度
                results.append({
                    **self.chunks[idx],
                    "similarity": float(similarities[idx]),
                })

        return results

    def save(self, path: str):
        """持久化向量库"""
        np.save(f"{path}_vectors.npy", self.vectors)
        with open(f"{path}_chunks.json", "w", encoding="utf-8") as f:
            json.dump(self.chunks, f, ensure_ascii=False, indent=2)

    def load(self, path: str):
        """加载向量库"""
        vectors_path = Path(f"{path}_vectors.npy")
        chunks_path = Path(f"{path}_chunks.json")
        if vectors_path.exists() and chunks_path.exists():
            self.vectors = np.load(vectors_path)
            self.chunks = json.loads(chunks_path.read_text("utf-8"))
            return True
        return False


# ================================================================
# 第3部分：嵌入模型
# ================================================================

class Embedder:
    """文本向量化（将文本转为数字向量）"""

    def __init__(self):
        self.model = None
        self._init_model()

    def _init_model(self):
        """初始化嵌入模型"""
        try:
            from sentence_transformers import SentenceTransformer
            # all-MiniLM-L6-v2: 轻量、快速、中文尚可（学习用）
            # 生产环境建议: BAAI/bge-small-zh-v1.5（中文专用）
            self.model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
            print(f"✅ 嵌入模型加载成功 (维度: {self.model.get_sentence_embedding_dimension()})")
        except ImportError:
            print("❌ 请安装 sentence-transformers: pip install sentence-transformers")
            raise
        except Exception as e:
            print(f"⚠️ 模型加载失败: {e}")
            print("  使用简易词袋模型作为后备方案...")
            self.model = None

    def encode(self, texts: list[str]) -> np.ndarray:
        """将文本列表转为向量矩阵"""
        if self.model:
            return self.model.encode(texts, show_progress_bar=True)
        # 后备：简单的词袋模型（非常简陋，仅用于教学）
        return self._bag_of_words(texts)

    def _bag_of_words(self, texts: list[str]) -> np.ndarray:
        """简易词袋模型（后备方案）"""
        # 构建词汇表
        vocab = set()
        for text in texts:
            for char in text:
                vocab.add(char)
        vocab_list = sorted(vocab)

        vectors = np.zeros((len(texts), len(vocab_list)))
        for i, text in enumerate(texts):
            for char in text:
                if char in vocab_list:
                    vectors[i, vocab_list.index(char)] += 1
        return vectors


# ================================================================
# 第4部分：知识库 Agent
# ================================================================

class QAAgent:
    """智能问答知识库 Agent"""

    def __init__(self, knowledge_base_path: str = "./knowledge_base"):
        self.kb_path = Path(knowledge_base_path)
        self.kb_path.mkdir(parents=True, exist_ok=True)

        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"

        self.processor = DocumentProcessor()
        self.embedder = Embedder()
        self.vector_store = SimpleVectorStore()
        self.memory: list[dict] = []

    # --- 文档导入 ---
    def import_documents(self, source_dir: str = None) -> int:
        """导入文档到知识库"""
        source = Path(source_dir) if source_dir else self.kb_path
        if not source.exists():
            print(f"❌ 目录不存在: {source}")
            return 0

        print(f"📥 正在导入文档: {source}")
        chunks = self.processor.process_directory(source)

        if not chunks:
            print("⚠️ 未找到可导入的文档")
            return 0

        # 向量化
        print(f"🧮 正在向量化 {len(chunks)} 个文本块...")
        texts = [c["content"] for c in chunks]
        embeddings = self.embedder.encode(texts)

        # 存储
        self.vector_store.add(chunks, embeddings)
        self.vector_store.save(str(self.kb_path / "vectordb"))

        print(f"✅ 导入完成！知识库现有 {len(self.vector_store.chunks)} 个文本块")
        return len(chunks)

    # --- 核心问答 ---
    def ask(self, question: str, top_k: int = 5) -> str:
        """向知识库提问"""
        print(f"\n{'─'*60}")
        print(f"❓ 问题: {question}")

        if len(self.vector_store.chunks) == 0:
            return "⚠️ 知识库为空，请先导入文档（使用 import 命令）"

        # 步骤1：问题向量化
        print("🔍 正在检索相关文档...")
        question_embedding = self.embedder.encode([question])

        # 步骤2：语义检索
        relevant_chunks = self.vector_store.search(question_embedding, top_k=top_k)

        if not relevant_chunks:
            return "😕 在知识库中没有找到与您问题相关的内容。"

        print(f"📋 找到 {len(relevant_chunks)} 个相关片段:")
        for c in relevant_chunks[:3]:
            print(f"  • {c['source_file']} (相关度: {c['similarity']:.3f})")

        # 步骤3：构建增强 Prompt
        context_parts = []
        for i, chunk in enumerate(relevant_chunks, 1):
            context_parts.append(
                f"[来源{i}: {chunk['source_file']}]\n{chunk['content']}"
            )
        context = "\n\n---\n\n".join(context_parts)

        # 步骤4：调用 Claude 生成答案
        answer = self._generate_answer(question, context, relevant_chunks)

        # 保存到记忆
        self.memory.append({
            "question": question,
            "answer": answer,
            "sources": [c["source_file"] for c in relevant_chunks],
            "time": datetime.now().isoformat(),
        })

        return answer

    def _generate_answer(self, question: str, context: str, sources: list[dict]) -> str:
        """基于检索到的文档上下文生成答案"""
        system_prompt = """你是一个知识库问答助手。请严格基于提供的文档内容回答问题。

【重要规则】
1. 只使用下面提供的文档内容回答，不要编造任何信息
2. 如果文档内容不足以回答问题，明确说明"根据现有文档，无法确定..."
3. 回答后，标注引用了哪些来源文档
4. 使用中文回答
5. 回答要简洁、直接、有依据
6. 如果文档中有代码或技术细节，保留原文格式"""

        user_message = f"""请根据以下文档内容回答问题。

【文档内容】
{context}

【用户问题】
{question}

请在回答末尾标注使用的来源文档。"""

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1500,
                system=system_prompt,
                messages=[{"role": "user", "content": user_message}],
                temperature=0.2,
            )
            answer = response.content[0].text.strip()

            # 追加来源信息
            unique_sources = list(set(c["source_file"] for c in sources))
            answer += f"\n\n---\n📚 **参考文档**: {', '.join(unique_sources[:5])}"

            return answer
        except Exception as e:
            return f"❌ 生成答案失败: {e}"

    # --- 追问（带记忆） ---
    def follow_up(self, question: str, top_k: int = 5) -> str:
        """追问（包含之前对话上下文）"""
        if not self.memory:
            return self.ask(question, top_k)

        # 带上最近3轮对话作为上下文
        recent = self.memory[-3:]
        context = "\n".join([
            f"之前问题: {m['question']}\n之前回答: {m['answer'][:200]}"
            for m in recent
        ])

        full_question = f"对话历史:\n{context}\n\n新问题（可能和上述对话相关）: {question}"
        return self.ask(full_question, top_k)

    # --- 显示知识库状态 ---
    def show_status(self):
        print(f"\n📊 知识库状态:")
        print(f"  路径: {self.kb_path.absolute()}")
        print(f"  文档块: {len(self.vector_store.chunks)} 个")
        files = set(c.get("source_file", "未知") for c in self.vector_store.chunks)
        print(f"  源文件: {len(files)} 个")
        print(f"  对话记忆: {len(self.memory)} 轮")
        if files:
            print(f"  文件列表: {', '.join(sorted(files)[:10])}")


# ================================================================
# 第5部分：交互入口
# ================================================================

def main():
    print("=" * 60)
    print("📚 QAAgent — 智能问答知识库 Agent (RAG)")
    print("=" * 60)
    print("命令:")
    print("  import [目录]  — 导入文档（默认 ./knowledge_base）")
    print("  ask <问题>   — 向知识库提问")
    print("  follow <问题> — 带记忆的追问")
    print("  status        — 查看知识库状态")
    print("  quit          — 退出")
    print("=" * 60)

    if not os.getenv("ANTHROPIC_API_KEY"):
        print("❌ 请先配置 ANTHROPIC_API_KEY！")
        return

    # 创建知识库目录和示例文档
    kb_path = Path("./knowledge_base")
    kb_path.mkdir(exist_ok=True)
    (kb_path / "ai_intro.md").write_text(
        "# AI Agent 简介\n\nAI Agent 是一种能够自主感知环境、制定计划、"
        "使用工具并执行任务的智能程序。\n\n## 核心组件\n\n"
        "1. **大模型基座**: Agent 的\"大脑\"，负责理解和推理\n"
        "2. **记忆模块**: 存储对话历史和重要信息\n"
        "3. **工具调用**: 让 Agent 能够操作外部世界\n"
        "4. **任务规划**: 将复杂任务分解为可执行的步骤\n\n"
        "## 与传统程序的区别\n\n"
        "传统程序按固定逻辑执行，而 AI Agent 能够根据目标自主调整策略。\n"
        "比如一个文档整理 Agent，不只是机械改名，而是理解文档内容后智能分类。\n",
        encoding="utf-8"
    )
    (kb_path / "python_tips.md").write_text(
        "# Python 性能优化技巧\n\n"
        "## 列表推导式\n使用列表推导式比传统 for 循环快约 30-50%。\n\n"
        "## 生成器\n处理大数据时使用生成器可以节省内存，`yield` 关键字是关键。\n\n"
        "## 字典查找\n字典的查找复杂度是 O(1)，列表是 O(n)。\n"
        "频繁查找操作请使用字典或集合。\n\n"
        "## 字符串拼接\n大量字符串拼接使用 `''.join()` 而非 `+=`，性能可提升10倍以上。\n",
        encoding="utf-8"
    )
    print("📝 已创建示例知识库文档")

    agent = QAAgent()
    # 自动导入示例文档
    agent.import_documents()

    while True:
        try:
            user_input = input("\n👤 你: ").strip()
            if not user_input:
                continue
            if user_input.lower() in ("quit", "exit", "q"):
                break
            if user_input.lower() == "status":
                agent.show_status()
                continue

            if user_input.lower().startswith("import"):
                parts = user_input.split(maxsplit=1)
                source = parts[1] if len(parts) > 1 else None
                agent.import_documents(source)
                continue

            if user_input.lower().startswith("follow "):
                question = user_input[7:]
                result = agent.follow_up(question)
                print(f"\n🤖 Agent:\n{result}")
                continue

            if user_input.lower().startswith("ask "):
                question = user_input[4:]
            else:
                question = user_input  # 默认作为提问

            result = agent.ask(question)
            print(f"\n🤖 Agent:\n{result}")

        except KeyboardInterrupt:
            break

    print(f"\n👋 知识库 Agent 已退出。再见！")


if __name__ == "__main__":
    main()

4. 调试与优化