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

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 =