AI Agent与RAG结合:构建知识增强型智能体 AI Agent与RAG结合:构建知识增强型智能体大语言模型(LLM)虽然知识广博,但在面对特定领域问题时往往力不从心——要么产生幻觉,要么知识已经过时。检索增强生成(RAG)技术恰好为 AI Agent è¡¥å äº†å®žæ—¶ç²¾å‡†çŸ¥è¯†è¿™ä¸€å ³é”®èƒ½åŠ›ã€‚æœ¬æ–‡å°†ç³»ç»Ÿè®²è§£ RAG åŽŸç†ã€å‘é‡æ£€ç´¢æœºåˆ¶ã€æ–‡æ¡£å¤„ç†ç­–ç•¥ä»¥åŠä¸€ä¸ªå®Œæ•´çš„ä»£ç å®žçŽ°ã€‚ä¸€ã€RAG æ ¸å¿ƒåŽŸç†1.1 什么是 RAGRAG(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