AI Agent与RAG结合：构建知识增强型智能体

大语言模型（LLM）虽然知识广博，但在面对特定领域问题时往往力不从心——要么产生幻觉，要么知识已经过时。检索增强生成（RAG）技术恰好为 AI Agent 补充了"实时精准知识"这一关键能力。本文将系统讲解 RAG 原理、向量检索机制、文档处理策略以及一个完整的代码实现。

一、RAG 核心原理

1.1 什么是 RAG

RAG（Retrieval-Augmented Generation）的核心思想是：在生成回答前，先从外部知识库检索相关信息，将检索结果作为上下文输入大语言模型，引导其生成更准确的回答。

相比微调 LLM，RAG 具有三个优势：

• 知识实时性：无需重新训练即可更新知识库
• 可溯源：每个回答都可追溯到具体来源
• 幻觉抑制：检索到的真实信息有效约束模型输出

1.2 标准流程

一个典型的 RAG 系统包含：文档加载 → 文档切分 → 向量嵌入 → 向量存储 → 检索 → 重排序 → 提示构建 → 生成回答。

二、文档切分策略

2.1 切分策略对比

| 策略 | 说明 | 适用场景 | |------|------|----------| | 固定字符数 | 按固定长度切分 | 通用场景，实现简单 | | 递归字符切分 | 先按段落，再按句子切分 | 结构化文档 | | 重叠切分 | 相邻 chunk 保留重叠区域 | 避免边界信息丢失 |

2.2 切分实现

import re
from typing import List

class DocumentChunker:
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 100):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_text(self, text: str) -> List[str]:
        paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
        chunks = []
        for p in paragraphs:
            if len(p) <= self.chunk_size:
                chunks.append(p)
            else:
                chunks.extend(self._split_by_sentences(p))
        return self._merge_small_chunks(chunks)
    
    def _split_by_sentences(self, text: str) -> List[str]:
        sentences = re.split(r'(?<=[。．.?!])\s+', text)
        sentences = [s.strip() for s in sentences if s.strip()]
        chunks, current = [], ""
        for s in sentences:
            if len(current) + len(s) <= self.chunk_size:
                current += s
            else:
                if current:
                    chunks.append(current)
                overlap = current[-self.chunk_overlap:] if self.chunk_overlap > 0 else ""
                current = overlap + s
        if current:
            chunks.append(current)
        return chunks
    
    def _merge_small_chunks(self, chunks: List[str]) -> List[str]:
        merged, current = [], ""
        for chunk in chunks:
            if len(current) + len(chunk) <= self.chunk_size:
                current += chunk
            else:
                if current:
                    merged.append(current)
                current = chunk
        if current:
            merged.append(current)
        return merged

三、向量检索与存储

3.1 向量数据库

文本通过嵌入模型转换为高维向量，存储在向量数据库中。常见的开源方案包括 BGE-M3、E5-Mistral 等。下面用纯 Python 实现一个内存向量库（生产环境建议使用 Milvus 或 Chroma）：

import numpy as np
from typing import List, Tuple, Dict
from dataclasses import dataclass

@dataclass
class Document:
    id: str
    content: str
    embedding: np.ndarray
    metadata: dict

class SimpleVectorStore:
    def __init__(self, embedding_dim: int =