RAG(Retrieval-Augmented Generation),检索增强生成，就是在LLM生成答案前从向量库中搜索匹配度高的答案和提示词一并交给LLM，随即生成更加符合且准确的答案，本质上解决LLM不能处理上下文的问题、知识固定化的问题。
完整的RAG流程包括离线和在线过程。
离线过程中主要是对用户问题进行格式化转换，即把（PDG、word、png）转为markdown，然后对文本进行切割分块生成多个chunk，再进行chunk的向量化最终组成向量库。
在线过程主要是对用户问题query进行处理，把用户的口语化信息整理为大模型所能理解的内容，然后进行粗排，找出与问题相近的Top-k的chunk，然后进行细排，Rerank模型会将用户问题与chunk混在一起，找出它们之间的相关性，然后重新排列。
最后将用户的问题与精排的chunk一并交给prompt后，经llm处理给出用户最终答案。
这里基于hello-agent中给出具体实现
from hello_agents import SimpleAgent, HelloAgentsLLM, ToolRegistry
from hello_agents.tools import RAGTool
# 创建LLM实例
llm = HelloAgentsLLM()
# 创建RAG工具
rag_tool = RAGTool(
knowledge_base_path="./knowledge_base",
collection_name="test_collection",
rag_namespace="test"
)
#注册工具
tool_registry = ToolRegistry()
tool_registry.register_tool(rag_tool)
# 创建具有RAG能力的Agent
agent = SimpleAgent(name="知识助手", llm=llm, tool_registry=tool_registry)
# 体验RAG功能
# 添加第一个知识
result1 = rag_tool.execute("add_text",
text="Python是一种高级编程语言，由Guido van Rossum于1991年首次发布。Python的设计哲学强调代码的可读性和简洁的语法。",
document_id="python_intro")
print(f"知识1: {result1}")
# 添加第二个知识
result2 = rag_tool.execute("add_text",
text="机器学习是人工智能的一个分支，通过算法让计算机从数据中学习模式。主要包括监督学习、无监督学习和强化学习三种类型。",
document_id="ml_basics")
print(f"知识2: {result2}")
# 添加第三个知识
result3 = rag_tool.execute("add_text",
text="RAG（检索增强生成）是一种结合信息检索和文本生成的AI技术。它通过检索相关知识来增强大语言模型的生成能力。",
document_id="rag_concept")
print(f"知识3: {result3}")
print("\n=== 搜索知识 ===")
result = rag_tool.execute("search",
query="Python编程语言的历史",
limit=3,
min_score=0.1
)
print(result)
print("\n=== 知识库统计 ===")
result = rag_tool.execute("stats")
print(result)
以下是RAG_TOOL工具类代码
class RAGTool(Tool):
"""RAG工具
提供完整的 RAG 能力：
- 添加多格式文档（PDF、Office、图片、音频等）
- 智能检索与召回
- LLM 增强问答
- 知识库管理
"""
def __init__(
self,
knowledge_base_path: str = "./knowledge_base",
qdrant_url: str = None,
qdrant_api_key: str = None,
collection_name: str = "rag_knowledge_base",
rag_namespace: str = "default"
):
# 初始化RAG管道
self._pipelines: Dict[str, Dict[str, Any]] = {}
self.llm = HelloAgentsLLM()
# 创建默认管道
default_pipeline = create_rag_pipeline(
qdrant_url=self.qdrant_url,
qdrant_api_key=self.qdrant_api_key,
collection_name=self.collection_name,
rag_namespace=self.rag_namespace
)
self._pipelines[self.rag_namespace] = default_pipeline
以下介绍RAG的离线功能
格式转换
def _convert_to_markdown(path: str) -> str:
"""
Universal document reader using MarkItDown with enhanced PDF processing.
核心功能：将任意格式文档转换为Markdown文本
支持格式：
- 文档：PDF、Word、Excel、PowerPoint
- 图像：JPG、PNG、GIF（通过OCR）
- 音频：MP3、WAV、M4A（通过转录）
- 文本：TXT、CSV、JSON、XML、HTML
- 代码：Python、JavaScript、Java等
"""
if not os.path.exists(path):
return ""
# 对PDF文件使用增强处理
ext = (os.path.splitext(path)[1] or '').lower()
if ext == '.pdf':
return _enhanced_pdf_processing(path)
# 其他格式使用MarkItDown统一转换
md_instance = _get_markitdown_instance()
if md_instance is None:
return _fallback_text_reader(path)
try:
result = md_instance.convert(path)
markdown_text = getattr(result, "text_content", None)
if isinstance(markdown_text, str) and markdown_text.strip():
print(f"[RAG] MarkItDown转换成功: {path} -> {len(markdown_text)} chars Markdown")
return markdown_text
return ""
except Exception as e:
print(f"[WARNING] MarkItDown转换失败 {path}: {e}")
return _fallback_text_reader(path)
智能分块
def _split_paragraphs_with_headings(text: str) -> List[Dict]:
"""根据标题层次分割段落，保持语义完整性"""
lines = text.splitlines()
heading_stack: List[str] = []
paragraphs: List[Dict] = []
buf: List[str] = []
char_pos = 0
def flush_buf(end_pos: int):
if not buf:
return
content = "\n".join(buf).strip()
if not content:
return
paragraphs.append({
"content": content,
"heading_path": " > ".join(heading_stack) if heading_stack else None,
"start": max(0, end_pos - len(content)),
"end": end_pos,
})
for ln in lines:
raw = ln
if raw.strip().startswith("#"):
# 处理标题行
flush_buf(char_pos)
level = len(raw) - len(raw.lstrip('#'))
title = raw.lstrip('#').strip()
if level <= 0:
level = 1
if level <= len(heading_stack):
heading_stack = heading_stack[:level-1]
heading_stack.append(title)
char_pos += len(raw) + 1
continue
# 段落内容累积
if raw.strip() == "":
flush_buf(char_pos)
buf = []
else:
buf.append(raw)
char_pos += len(raw) + 1
flush_buf(char_pos)
if not paragraphs:
paragraphs = [{"content": text, "heading_path": None, "start": 0, "end": len(text)}]
return paragraphs
基于token数量进行智能分块
def _chunk_paragraphs(paragraphs: List[Dict], chunk_tokens: int, overlap_tokens: int) -> List[Dict]:
"""基于Token数量的智能分块"""
chunks: List[Dict] = []
cur: List[Dict] = []
cur_tokens = 0
i = 0
while i < len(paragraphs):
p = paragraphs[i]
p_tokens = _approx_token_len(p["content"]) or 1
if cur_tokens + p_tokens <= chunk_tokens or not cur:
cur.append(p)
cur_tokens += p_tokens
i += 1
else:
# 生成当前分块
content = "\n\n".join(x["content"] for x in cur)
start = cur[0]["start"]
end = cur[-1]["end"]
heading_path = next((x["heading_path"] for x in reversed(cur) if x.get("heading_path")), None)
chunks.append({
"content": content,
"start": start,
"end": end,
"heading_path": heading_path,
})
# 构建重叠部分
if overlap_tokens > 0 and cur:
kept: List[Dict] = []
kept_tokens = 0
for x in reversed(cur):
t = _approx_token_len(x["content"]) or 1
if kept_tokens + t > overlap_tokens:
break
kept.append(x)
kept_tokens += t
cur = list(reversed(kept))
cur_tokens = kept_tokens
else:
cur = []
cur_tokens = 0
# 处理最后一个分块
if cur:
content = "\n\n".join(x["content"] for x in cur)
start = cur[0]["start"]
end = cur[-1]["end"]
heading_path = next((x["heading_path"] for x in reversed(cur) if x.get("heading_path")), None)
chunks.append({
"content": content,
"start": start,
"end": end,
"heading_path": heading_path,
})
return chunks向量化
def index_chunks(
store = None,
chunks: List[Dict] = None,
cache_db: Optional[str] = None,
batch_size: int = 64,
rag_namespace: str = "default"
) -> None:
"""
Index markdown chunks with unified embedding and Qdrant storage.
Uses百炼 API with fallback to sentence-transformers.
"""
if not chunks:
print("[RAG] No chunks to index")
return
# 使用统一嵌入模型
embedder = get_text_embedder()
dimension = get_dimension(384)
# 创建默认Qdrant存储
if store is None:
store = _create_default_vector_store(dimension)
print(f"[RAG] Created default Qdrant store with dimension {dimension}")
# 预处理Markdown文本以获得更好的嵌入质量
processed_texts = []
for c in chunks:
raw_content = c["content"]
processed_content = _preprocess_markdown_for_embedding(raw_content)
processed_texts.append(processed_content)
print(f"[RAG] Embedding start: total_texts={len(processed_texts)} batch_size={batch_size}")
# 批量编码
vecs: List[List[float]] = []
for i in range(0, len(processed_texts), batch_size):
part = processed_texts[i:i+batch_size]
try:
# 使用统一嵌入器（内部处理缓存）
part_vecs = embedder.encode(part)
# 标准化为List[List[float]]格式
if not isinstance(part_vecs, list):
if hasattr(part_vecs, "tolist"):
part_vecs = [part_vecs.tolist()]
else:
part_vecs = [list(part_vecs)]
# 处理向量格式和维度
for v in part_vecs:
try:
if hasattr(v, "tolist"):
v = v.tolist()
v_norm = [float(x) for x in v]
# 维度检查和调整
if len(v_norm) != dimension:
print(f"[WARNING] 向量维度异常: 期望{dimension}, 实际{len(v_norm)}")
if len(v_norm) < dimension:
v_norm.extend([0.0] * (dimension - len(v_norm)))
else:
v_norm = v_norm[:dimension]
vecs.append(v_norm)
except Exception as e:
print(f"[WARNING] 向量转换失败: {e}, 使用零向量")
vecs.append([0.0] * dimension)
except Exception as e:
print(f"[WARNING] Batch {i} encoding failed: {e}")
# 实现重试机制
# ... 重试逻辑 ...
print(f"[RAG] Embedding progress: {min(i+batch_size, len(processed_texts))}/{len(processed_texts)}")
评论