工程实践

为什么结构化输出如此重要 在生产环境中，LLM的输出需要被程序解析和处理。非结构化的自然语言输出虽然灵活，但带来三个严重问题： 解析不可靠：正则提取容易遗漏边界情况 集成困难：下游系统需要稳定的接口契约 验证缺失：无法保证输出满足业务约束 2026年，结构化输出已从"nice to have"变为"must have"。所有主流模型都提供了原生结构化输出能力。 技术方案全景 方案对比 方案 原理 可靠性 性能 灵活性 适用场景 JSON Mode 模型内置JSON生成 95% 高 中 简单结构 Function Calling 函数签名约束 97% 高 高 API调用 Constrained Decoding 解码时约束 99% 中 最高 严格格式 Pydantic + LLM Schema验证+重试 90% 低 高 复杂校验 XML标签 标签结构化 85% 高 低 简单提取 JSON Mode 基本使用 import json from openai import OpenAI client = OpenAI() # OpenAI JSON Mode response = client.chat.completions.create( model="gpt-4o-2026", response_format={"type": "json_object"}, messages=[ { "role": "system", "content": """你是一个信息提取助手。 请将用户输入提取为JSON格式，包含以下字段： - name: 姓名 - age: 年龄（整数） - skills: 技能列表（字符串数组） - experience: 工作经验（整数，单位年） """ }, { "role": "user", "content": "张三，28岁，精通Python和JavaScript，有5年开发经验" } ] ) result = json.loads(response.choices[0].message.content) print(result) # 输出: {"name": "张三", "age": 28, "skills": ["Python", "JavaScript"], "experience": 5} JSON Schema约束 # 2026年最新：JSON Schema强化约束 response = client.chat.completions.create( model="gpt-4o-2026", response_format={ "type": "json_schema", "json_schema": { "name": "employee_info", "strict": True, "schema": { "type": "object", "properties": { "name": {"type": "string", "minLength": 1, "maxLength": 50}, "age": {"type": "integer", "minimum": 18, "maximum": 65}, "skills": { "type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 20 }, "experience": {"type": "integer", "minimum": 0, "maximum": 50}, "level": { "type": "string", "enum": ["junior", "mid", "senior", "expert"] } }, "required": ["name", "age", "skills", "experience", "level"], "additionalProperties": False } } }, messages=[...] ) 各家模型JSON Mode对比 模型 JSON可靠性 Schema支持 性能影响 特殊限制 GPT-4o 95% 完整 <5% 需提示JSON关键词 Claude 4 93% XML标签 <3% 推荐XML格式 Gemini 2 94% 部分支持 <5% Qwen 3 92% 部分 <5% Llama 4 88% 不支持 <8% 需Few-shot Function Calling 基本架构 from dataclasses import dataclass from typing import Callable import inspect @dataclass class ToolDefinition: name: str description: str parameters: dict # JSON Schema class FunctionCallingSystem: """ 2026年Function Calling最佳实践 """ def __init__(self, model_client): self.model = model_client self.tools: dict[str, ToolDefinition] = {} self.handlers: dict[str, Callable] = {} def register_function(self, func: Callable, description: str): """注册可调用函数""" # 自动从函数签名生成Schema sig = inspect.signature(func) params = {} required = [] for name, param in sig.parameters.items(): param_type = param.annotation json_type = self._python_type_to_json(param_type) params[name] = { "type": json_type, "description": self._extract_param_doc(func, name) } if param.default == inspect.Parameter.empty: required.append(name) tool = ToolDefinition( name=func.__name__, description=description or func.__doc__, parameters={ "type": "object", "properties": params, "required": required } ) self.tools[func.__name__] = tool self.handlers[func.__name__] = func async def execute_with_functions(self, user_message: str) -> str: """带函数调用的对话""" messages = [{"role": "user", "content": user_message}] tools = [t.__dict__ for t in self.tools.values()] while True: response = await self.model.chat( messages=messages, tools=tools, tool_choice="auto" ) message = response.choices[0].message messages.append(message) if not message.tool_calls: # 模型没有调用工具，返回最终回复 return message.content # 执行函数调用 for tool_call in message.tool_calls: func_name = tool_call.function.name func_args = json.loads(tool_call.function.arguments) # 参数验证 validation = self._validate_arguments(func_name, func_args) if not validation["valid"]: result = f"参数错误: {validation['errors']}" else: # 执行函数 try: handler = self.handlers[func_name] result = await handler(**func_args) except Exception as e: result = f"执行错误: {str(e)}" # 将结果返回给模型 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) 实际应用示例 # 定义工具函数 @dataclass class SearchResult: title: str url: str snippet: str async def search_web(query: str, max_results: int = 5) -> list[dict]: """搜索网络内容 Args: query: 搜索关键词 max_results: 最大返回结果数（默认5） """ # 实际搜索逻辑 results = await search_engine.search(query, limit=max_results) return [{"title": r.title, "url": r.url, "snippet": r.snippet} for r in results] async def get_weather(city: str, unit: str = "celsius") -> dict: """获取指定城市的天气信息 Args: city: 城市名称 unit: 温度单位（celsius或fahrenheit） """ weather = await weather_api.get(city, unit) return { "city": city, "temperature": weather.temp, "condition": weather.condition, "humidity": weather.humidity } async def send_email(to: str, subject: str, body: str) -> dict: """发送邮件 Args: to: 收件人邮箱 subject: 邮件主题 body: 邮件正文 """ await email_service.send(to, subject, body) return {"status": "sent", "to": to} # 注册并使用 system = FunctionCallingSystem(model_client) system.register_function(search_web, "搜索网络获取最新信息") system.register_function(get_weather, "查询天气信息") system.register_function(send_email, "发送邮件") # 执行 response = await system.execute_with_functions( "帮我查一下北京今天的天气，然后把结果发邮件给 zhangsan@example.com" ) Constrained Decoding 原理 Constrained Decoding（约束解码）在生成过程中实时约束token选择，确保输出符合预定义的语法规则。 ...

RAG 故障诊断框架 RAG 系统出问题时，症状往往模糊——“回答不对”、“答非所问”、“太慢了”。2026 年的实践表明，几乎所有 RAG 故障都可以归因到三个核心环节：检索（Retrieval）、生成（Generation）、系统（System）。 故障报告 → 症状分类 → 根因定位 → 修复方案 → 验证 ↓ ┌─────────┼─────────┐ ↓ ↓ ↓ 检索层 生成层 系统层 漏检 幻觉 延迟 误检 偏题 超时 排序错 不完整 成本高 故障1：幻觉——“系统在编造事实” 症状 答案包含上下文中不存在的信息 引用了错误的来源 捏造数据或数字 诊断流程 class HallucinationDiagnostic: def diagnose(self, question: str, answer: str, contexts: list): # 1. 拆分答案为原子论断 claims = self._decompose_claims(answer) # 2. 逐条验证 results = [] for claim in claims: evidence = self._find_evidence(claim, contexts) results.append({ "claim": claim, "has_evidence": evidence["score"] > 0.5, "evidence_text": evidence["text"], "score": evidence["score"] }) hallucinated = [r for r in results if not r["has_evidence"]] return { "total_claims": len(claims), "hallucinated_claims": len(hallucinated), "hallucination_rate": len(hallucinated) / len(claims), "details": results, "likely_causes": self._identify_causes(hallucinated, contexts) } def _identify_causes(self, hallucinated, contexts): causes = [] # 原因1：上下文不足 if len(contexts) < 3: causes.append({ "cause": "context_insufficient", "description": "检索到的上下文数量太少", "fix": "增加 top_k 参数或扩大检索范围" }) # 原因2：上下文不相关 if contexts and all(not self._is_relevant(c) for c in contexts): causes.append({ "cause": "context_irrelevant", "description": "检索到的上下文与问题不相关", "fix": "检查 Embedding 模型质量或分块策略" }) # 原因3：LLM 忽略上下文 if contexts and any(self._is_relevant(c) for c in contexts): causes.append({ "cause": "llm_ignoring_context", "description": "上下文包含相关信息但 LLM 未使用", "fix": "优化 Prompt，强调'仅基于上下文回答'" }) return causes 常见根因与修复 根因 症状 修复方案 检索结果不相关 答案与文档无关 优化 Embedding 模型、加入重排序 上下文太短 缺少关键信息 增加 top_k、扩展上下文窗口 LLM 忽略上下文 编造而非引用 加强 Prompt 约束、降低 temperature 分块切断关键信息 信息碎片化 调整分块策略、增加重叠 模型本身幻觉倾向 即使上下文完整也编造 换用更可靠的模型 Prompt 修复示例 # 修复前（容易幻觉） BAD_PROMPT = "根据以下信息回答问题：{context}\n问题：{question}" # 修复后（抑制幻觉） GOOD_PROMPT = """请严格基于以下参考信息回答问题。 规则： 1. 只使用参考信息中的内容 2. 如果参考信息不足以回答，请说"根据现有信息无法回答" 3. 每个事实陈述后用 [1], [2] 标注来源 4. 不要编造任何信息 参考信息： {context} 问题：{question} 回答：""" 故障2：漏检——“该找到的没找到” 症状 知识库中明明有相关信息，但系统说"不知道" 回答不完整，遗漏关键信息 多文档对比时只引用了一部分 诊断流程 class RetrievalDiagnostic: def diagnose(self, question: str, retrieved: list, ground_truth_docs: list): report = {} # 1. 召回率分析 retrieved_ids = {r.id for r in retrieved} gt_ids = {d.id for d in ground_truth_docs} report["recall"] = len(retrieved_ids & gt_ids) / len(gt_ids) report["missed_docs"] = list(gt_ids - retrieved_ids) # 2. 分析漏检原因 for missed_id in report["missed_docs"]: missed_doc = self._get_doc(missed_id) root_cause = self._analyze_miss(question, missed_doc) report.setdefault("miss_reasons", []).append(root_cause) return report def _analyze_miss(self, question: str, missed_doc: Document): # 原因A: Embedding 相似度过低 q_emb = self.embedder.encode(question) d_emb = self.embedder.encode(missed_doc.content) similarity = cosine_similarity(q_emb, d_emb) if similarity < 0.5: return { "doc_id": missed_doc.id, "cause": "embedding_similarity_low", "similarity": similarity, "fix": "考虑使用 HyDE 或 Query 重写" } # 原因B: 分块导致信息碎片化 if len(missed_doc.content) < 50: return { "doc_id": missed_doc.id, "cause": "chunk_too_short", "fix": "调整分块策略，增大 chunk_size" } # 原因C: 元数据过滤排除了该文档 return { "doc_id": missed_doc.id, "cause": "metadata_filter_excluded", "similarity": similarity, "fix": "检查元数据过滤条件" } 常见漏检模式 模式1: 语义鸿沟 Query: "怎么提高模型准确率" 文档: "模型精度优化方法" ← 同义词但 Embedding 可能不够近 修复: 加入 Query 重写或同义词扩展 模式2: 多跳信息分散 Query: "A和B的区别" 文档: 分别有A和B的描述，但没有对比性文档 修复: 使用多查询策略或 Agentic RAG 模式3: 长尾知识 Query: 某罕见概念 文档: 存在但 Embedding 训练时未见 修复: 加入关键词检索做补充 模式4: 版本过期 Query: 最新信息 文档: 旧版本未更新 修复: 定期增量更新 + 时间戳过滤 故障3：延迟过高——“等了10秒才出结果” 延迟分解 class LatencyProfiler: def profile(self, query: str): timings = {} # 1. Query Embedding t0 = time.time() q_emb = self.embedder.encode(query) timings["embedding"] = time.time() - t0 # 2. 向量检索 t0 = time.time() results = self.vector_store.search(q_emb, top_k=50) timings["vector_search"] = time.time() - t0 # 3. 重排序 t0 = time.time() reranked = self.reranker.rerank(query, results, top_k=5) timings["rerank"] = time.time() - t0 # 4. Prompt 构建 t0 = time.time() prompt = self._build_prompt(query, reranked) timings["prompt_build"] = time.time() - t0 # 5. LLM 生成 t0 = time.time() answer = self.llm.generate(prompt) timings["llm_generate"] = time.time() - t0 timings["total"] = sum(timings.values()) return timings 典型延迟分布 环节 优化前 优化后 优化方法 Embedding 200ms 50ms 本地部署 + 批量 向量检索 500ms 80ms HNSW 参数调优 + 缓存 重排序 800ms 200ms 批量推理 + 减小候选集 LLM 生成 3000ms 1500ms 流式输出 + 更短 Prompt 总计 4500ms 1830ms 优化代码 class OptimizedRAG: def __init__(self): self.semantic_cache = SemanticCache( threshold=0.95, ttl=3600 ) self.query_classifier = QueryClassifier() def query(self, question: str): # 0. 语义缓存检查（~1ms） cached = self.semantic_cache.get(question) if cached: return cached, "cache_hit" # 1. 查询分类：简单 vs 复杂 complexity = self.query_classifier.predict(question) if complexity < 0.3: # 快速通道：跳过重排序 results = self._fast_query(question) else: # 标准通道：完整流程 results = self._full_query(question) # 写入缓存 self.semantic_cache.set(question, results) return results def _fast_query(self, question): """快速通道：向量检索 → LLM（跳过重排序）""" q_emb = self.embedder.encode(question) results = self.vector_store.search(q_emb, top_k=5) return self.llm.generate(question, context=results, stream=True) 故障4：答案偏题——“问的是A，答的是B” 诊断 def diagnose_off_topic(question, answer, contexts): # 检查问题覆盖 question_aspects = extract_question_aspects(question) answer_coverage = check_coverage(answer, question_aspects) # 检查检索相关性 retrieval_relevance = assess_relevance(question, contexts) return { "question_aspects": question_aspects, "covered_aspects": answer_coverage["covered"], "missed_aspects": answer_coverage["missed"], "retrieval_relevance": retrieval_relevance, "root_cause": identify_cause(answer_coverage, retrieval_relevance) } 常见根因 根因 检查方法 修复 Query 理解错误 检查检索 Query 是否与用户意图一致 加入 Query 重写 检索偏向热门内容 检查是否总是检索到相同文档 调整多样性参数 LLM 上下文窗口溢出 检查 Prompt 长度 压缩上下文 多意图问题未拆分 检查问题是否包含多个子问题 Agent 拆解 故障排查清单 □ 检索阶段 □ 检索结果是否相关？ → 查看原始检索结果 □ top_k 是否足够？ → 检查召回率 □ Embedding 模型是否合适？ → 测试语义相似度 □ 分块是否合理？ → 检查块边界 □ 是否需要 Query 重写？ → 检查 Query 与文档的语义鸿沟 □ 生成阶段 □ Prompt 是否清晰约束？ → 检查 Prompt 模板 □ 上下文是否足够？ → 检查上下文长度 □ temperature 是否过高？ → 降到 0-0.3 □ 模型能力是否足够？ → 换更强模型测试 □ 系统阶段 □ 延迟在哪个环节？ → Profile 各阶段耗时 □ 缓存是否生效？ → 检查缓存命中率 □ 是否有资源瓶颈？ → 检查 GPU/CPU/内存 总结 RAG 故障排查的核心是分层诊断：先定位是检索问题还是生成问题，再深入排查具体根因。最高效的排查方法是建立一个端到端的可观测系统，记录每个环节的输入输出和耗时，让故障无处遁形。 ...

分块为何如此重要 在 RAG 系统中，分块（Chunking）是影响检索质量的第一道关卡。一个糟糕的分块策略会让最好的 Embedding 模型和最贵的 LLM 都无济于事。2026 年的分块策略已经从简单的固定长度切分进化到了语义感知、文档结构感知的智能分块。 五大分块策略对比 策略1：固定长度分块（Fixed-Size Chunking） 最基础的方式，按固定 token 数切分，带重叠窗口。 def fixed_size_chunk(text: str, chunk_size: int = 512, overlap: int = 50): tokens = tokenizer.encode(text) chunks = [] for i in range(0, len(tokens), chunk_size - overlap): chunk = tokens[i:i + chunk_size] chunks.append(tokenizer.decode(chunk)) if i + chunk_size >= len(tokens): break return chunks 优点：简单、快速、可控 缺点：可能切断语义、忽略文档结构 适用：纯文本、均匀内容 策略2：语义分块（Semantic Chunking） 基于 Embedding 相似度，在语义跳变点切分。 import numpy as np from sentence_transformers import SentenceTransformer def semantic_chunk(text: str, model, threshold: float = 0.5): # 1. 先按句子切分 sentences = split_sentences(text) # 2. 逐句计算 Embedding embeddings = model.encode(sentences) # 3. 计算相邻句子的余弦相似度 similarities = [] for i in range(len(embeddings) - 1): sim = cosine_similarity(embeddings[i], embeddings[i+1]) similarities.append(sim) # 4. 在相似度低于阈值处切分 chunks = [] current_chunk = [sentences[0]] for i, sim in enumerate(similarities): if sim < threshold: chunks.append(" ".join(current_chunk)) current_chunk = [sentences[i+1]] else: current_chunk.append(sentences[i+1]) if current_chunk: chunks.append(" ".join(current_chunk)) return chunks 优点：语义完整、边界自然 缺点：计算开销大、块大小不均匀 适用：长文本、叙述性内容 策略3：文档感知分块（Document-Aware Chunking） 基于文档结构（标题、段落、列表）进行切分。 ...

数据管道是 RAG 系统的地基 RAG 系统的效果上限由数据质量决定。再好的 Embedding 模型和 LLM，如果喂进去的是垃圾数据，出来的也是垃圾答案。2026 年的 RAG 数据管道已经发展为一套完整的工程体系。 整体架构 数据源 解析层 处理层 索引层 ┌─────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ PDF │───→│ │ │ │ │ │ │ HTML │───→│ 文档解析 │───→│ 清洗+增强 │───→│ Embedding │ │ Word │───→│ Unified │ │ Pipeline │ │ + Index │ │ DB │───→│ Parser │ │ │ │ │ │ API │───→│ │ │ │ │ │ │ Web │───→│ │ │ │ │ │ └─────────┘ └───────────┘ └───────────┘ └───────────┘ ↓ ┌───────────┐ │ 质量监控 │ │ QC Layer │ └───────────┘ 1. 多源文档解析 统一解析层 from abc import ABC, abstractmethod from typing import List, Document import fitz # PyMuPDF from bs4 import BeautifulSoup from unstructured import partition_pdf class DocumentParser(ABC): @abstractmethod def parse(self, file_path: str) -> List[Document]: pass class PDFParser(DocumentParser): def parse(self, file_path: str) -> List[Document]: # 方案1: PyMuPDF（快，但结构感知弱） # 方案2: Unstructured（慢，但结构感知强） # 方案3: LLM 辅助解析（最准，但贵） elements = partition_pdf( file_path, strategy="hi_res", # 高精度模式 infer_table_structure=True, extract_image_block_types=["Image", "Table"], extract_image_block_output_dir="./images/" ) documents = [] for elem in elements: doc = Document( content=elem.text if hasattr(elem, 'text') else str(elem), metadata={ "source": file_path, "page": elem.metadata.page_number if elem.metadata else 0, "category": elem.category, "type": self._map_type(elem.category) } ) documents.append(doc) return documents class HTMLParser(DocumentParser): def parse(self, file_path: str) -> List[Document]: with open(file_path, 'r', encoding='utf-8') as f: soup = BeautifulSoup(f.read(), 'html.parser') # 移除无关元素 for tag in soup(['script', 'style', 'nav', 'footer', 'header']): tag.decompose() documents = [] # 按标题分块 current_section = {"title": "", "content": []} for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'li', 'table', 'pre']): if element.name in ['h1', 'h2', 'h3']: if current_section["content"]: documents.append(Document( content=self._format_section(current_section), metadata={"source": file_path, "section": current_section["title"]} )) current_section = {"title": element.get_text(), "content": []} else: current_section["content"].append(self._extract_content(element)) if current_section["content"]: documents.append(Document( content=self._format_section(current_section), metadata={"source": file_path, "section": current_section["title"]} )) return documents class DatabaseParser(DocumentParser): def parse(self, conn_config: dict) -> List[Document]: """从数据库提取知识""" import sqlalchemy as sa engine = sa.create_engine(conn_config["url"]) documents = [] for table_name in conn_config["tables"]: query = f"SELECT * FROM {table_name}" df = pd.read_sql(query, engine) # 将每行转为一个文档 for _, row in df.iterrows(): content = self._row_to_text(row, table_name) documents.append(Document( content=content, metadata={ "source": f"db://{table_name}", "table": table_name, "primary_key": str(row.get('id', '')) } )) return documents class UnifiedParser: """统一解析入口""" def __init__(self): self.parsers = { ".pdf": PDFParser(), ".html": HTMLParser(), ".htm": HTMLParser(), ".docx": DocxParser(), ".md": MarkdownParser(), ".txt": TextParser(), } def parse(self, file_path: str) -> List[Document]: ext = Path(file_path).suffix.lower() parser = self.parsers.get(ext, TextParser()) return parser.parse(file_path) 2. 数据清洗 Pipeline class DataCleaner: def __init__(self): self.steps = [ self._remove_boilerplate, # 去除样板文字 self._normalize_whitespace, # 规范化空白 self._fix_encoding, # 修复编码 self._remove_duplicates, # 去重 self._filter_quality, # 质量过滤 self._add_metadata, # 补充元数据 ] def clean(self, documents: List[Document]) -> List[Document]: for step in self.steps: documents = step(documents) return documents def _remove_boilerplate(self, docs): """去除样板文字""" boilerplate_patterns = [ r"版权所有.*?保留所有权利", r"Cookie.*?设置", r"订阅我们的.*?通讯", r"© \d{4}.*?", ] cleaned = [] for doc in docs: text = doc.content for pattern in boilerplate_patterns: text = re.sub(pattern, "", text, flags=re.IGNORECASE) doc.content = text.strip() if len(doc.content) > 50: # 过滤过短的 cleaned.append(doc) return cleaned def _filter_quality(self, docs): """质量过滤""" filtered = [] for doc in docs: # 检查文本质量 quality_score = self._assess_quality(doc.content) if quality_score > 0.5: doc.metadata["quality_score"] = quality_score filtered.append(doc) return filtered def _assess_quality(self, text: str) -> float: score = 1.0 # 过短 if len(text) < 50: score -= 0.3 # 重复内容过多 words = text.split() if len(words) > 0: unique_ratio = len(set(words)) / len(words) if unique_ratio < 0.3: score -= 0.3 # 特殊字符过多 special_chars = sum(1 for c in text if not c.isalnum() and not c.isspace()) if len(text) > 0 and special_chars / len(text) > 0.2: score -= 0.2 # 乱码检测 if re.search(r'[\ufffd]{3,}', text): score -= 0.5 return max(0, score) 3. 数据增强 class DataAugmentor: """为原始文档增加结构化信息""" def augment(self, documents: List[Document]) -> List[Document]: for doc in documents: # 1. 生成摘要 doc.metadata["summary"] = self._generate_summary(doc.content) # 2. 提取关键词 doc.metadata["keywords"] = self._extract_keywords(doc.content) # 3. 生成问答对（用于 FAQ 式检索） doc.metadata["qa_pairs"] = self._generate_qa_pairs(doc.content) # 4. 实体标注 doc.metadata["entities"] = self._extract_entities(doc.content) # 5. 生成假设性问题（HyDE 优化） doc.metadata["hypothetical_questions"] = self._generate_hyde(doc.content) return documents def _generate_hyde(self, content: str) -> List[str]: """生成该内容可能回答的问题，用于提升检索召回""" prompt = f""" 基于以下内容，生成 3 个用户可能会问的问题，这些问题可以由这段内容回答： 内容：{content[:1000]} 输出 JSON 列表：["问题1", "问题2", "问题3"] """ return llm.generate(prompt, response_format="json") 4. 分块与索引 class ChunkingAndIndexing: def __init__(self): self.chunker = HybridChunker() self.embedder = EmbeddingModel("bge-m3") self.sparse_embedder = BM25Encoder() def process(self, documents: List[Document]): all_chunks = [] for doc in documents: # 1. 智能分块 chunks = self.chunker.chunk(doc.content) # 2. 为每个 chunk 添加上下文 for i, chunk in enumerate(chunks): chunk.metadata = { **doc.metadata, "chunk_index": i, "total_chunks": len(chunks), "context_before": chunks[i-1].text[-100:] if i > 0 else "", "context_after": chunks[i+1].text[:100] if i < len(chunks)-1 else "" } # 3. 生成 Embedding（稠密 + 稀疏） chunk.dense_embedding = self.embedder.encode(chunk.text) chunk.sparse_embedding = self.sparse_embedder.encode(chunk.text) # 4. HyDE embedding（用假设问题做额外 embedding） if "hypothetical_questions" in doc.metadata: hyde_embeddings = [ self.embedder.encode(q) for q in doc.metadata["hypothetical_questions"] ] chunk.hyde_embeddings = hyde_embeddings all_chunks.append(chunk) # 5. 写入向量数据库 self._write_to_index(all_chunks) return all_chunks 5. 质量监控 class PipelineQualityMonitor: def __init__(self): self.metrics = {} def monitor(self, pipeline_run): report = { "input": { "total_documents": len(pipeline_run.input_docs), "total_size_mb": pipeline_run.input_size, }, "parsing": { "success_rate": pipeline_run.parsed / pipeline_run.total, "failed_files": pipeline_run.failed, "avg_parse_time": pipeline_run.avg_parse_time, }, "cleaning": { "retention_rate": len(pipeline_run.cleaned) / len(pipeline_run.parsed), "avg_quality_score": np.mean([d.metadata.get("quality_score", 0) for d in pipeline_run.cleaned]), }, "chunking": { "total_chunks": len(pipeline_run.chunks), "avg_chunk_size": np.mean([len(c.text) for c in pipeline_run.chunks]), "chunks_per_doc": len(pipeline_run.chunks) / len(pipeline_run.cleaned), }, "indexing": { "index_time": pipeline_run.index_time, "index_size_mb": pipeline_run.index_size, "vectors_written": len(pipeline_run.chunks), } } # 告警 if report["parsing"]["success_rate"] < 0.95: alert("解析成功率低于 95%") if report["cleaning"]["avg_quality_score"] < 0.7: alert("平均质量分数低于 0.7") if report["chunking"]["avg_chunk_size"] > 2000: alert("平均块大小过大") return report 6. 增量更新与调度 class IncrementalPipeline: """支持增量更新的数据管道""" def run(self, data_source: str): # 1. 发现新增/修改/删除的文件 changes = self._detect_changes(data_source) if not changes["added"] and not changes["modified"] and not changes["deleted"]: return "No changes detected" # 2. 处理新增和修改 new_docs = [] for file_path in changes["added"] + changes["modified"]: docs = self.parser.parse(file_path) docs = self.cleaner.clean(docs) docs = self.augmentor.augment(docs) new_docs.extend(docs) # 3. 处理删除 for file_path in changes["deleted"]: self.index.delete_by_source(file_path) # 4. 更新索引 if new_docs: self.indexer.process(new_docs) # 5. 记录变更 self._update_file_registry(changes) # 6. 质量检查 quality_report = self.monitor.monitor(pipeline_run) return quality_report def _detect_changes(self, source: str) -> dict: """检测文件变更""" current_files = self._scan_files(source) registered_files = self._get_registered_files() added = set(current_files) - set(registered_files) deleted = set(registered_files) - set(current_files) modified = { f for f in current_files if f in registered_files and current_files[f] != registered_files[f] # 比较 hash } return {"added": list(added), "modified": list(modified), "deleted": list(deleted)} 总结 RAG 数据管道是系统工程，核心原则是： ...

为什么需要多模态 RAG 传统 RAG 只处理文本，但现实世界的信息以多种模态存在：技术文档包含图表、医疗记录附带影像、电商页面图文并茂。2026 年的多模态 RAG 已经成为企业级 RAG 系统的标配。 核心架构 多模态 RAG 的关键挑战是：如何让文本 Query 检索到相关图片，以及如何让图片 Query 检索到相关文本。 ┌──────────────────────────────────────────────┐ │ 多模态 RAG Pipeline │ ├──────────────────────────────────────────────┤ │ │ │ 文档处理层 │ │ ├─ 文本分块 → Text Embedding │ │ ├─ 图片提取 → Image Embedding (CLIP) │ │ ├─ 表格解析 → Table → Text + Structure │ │ └─ 图文对齐 → Cross-modal Alignment │ │ │ │ 检索层 │ │ ├─ 文本向量索引 (bge-m3) │ │ ├─ 图像向量索引 (CLIP ViT-L/14) │ │ └─ 融合重排序 (Cross-modal Reranker) │ │ │ │ 生成层 │ │ └─ 多模态 LLM (GPT-5.5 / Claude 4) │ │ │ └──────────────────────────────────────────────┘ 工程实现 1. 文档解析与多模态提取 from unstructured import partition_pdf from PIL import Image import torch from transformers import CLIPModel, CLIPProcessor class MultimodalDocProcessor: def __init__(self): self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14") self.clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14") self.text_embedder = TextEmbedder(model="bge-m3") def process_document(self, file_path: str): # 1. 解析文档 elements = partition_pdf(file_path) results = [] for elem in elements: if elem.category == "Text": results.append({ "type": "text", "content": elem.text, "embedding": self.text_embedder.encode(elem.text), "metadata": {"page": elem.metadata.page_number} }) elif elem.category == "Image": # 提取图片并生成 embedding img = elem.image img_embedding = self._encode_image(img) # 同时用 VLM 生成图片描述文本 img_description = vlm.describe(img) text_embedding = self.text_embedder.encode(img_description) results.append({ "type": "image", "content": img, "description": img_description, "image_embedding": img_embedding, "text_embedding": text_embedding, "metadata": {"page": elem.metadata.page_number} }) elif elem.category == "Table": # 表格转为结构化文本 table_text = self._table_to_text(elem) results.append({ "type": "table", "content": table_text, "embedding": self.text_embedder.encode(table_text), "metadata": {"page": elem.metadata.page_number} }) return results def _encode_image(self, image): inputs = self.clip_processor(images=image, return_tensors="pt") with torch.no_grad(): features = self.clip_model.get_image_features(**inputs) return features[0].cpu().numpy() 2. 双索引构建 class MultimodalIndex: def __init__(self): self.text_index = MilvusIndex( dim=1024, # bge-m3 metric="IP" ) self.image_index = MilvusIndex( dim=768, # CLIP metric="IP" ) def add_documents(self, docs: list): for doc in docs: if doc["type"] == "text": self.text_index.add( id=doc["id"], embedding=doc["embedding"], metadata=doc["metadata"] ) elif doc["type"] == "image": # 图片同时加入两个索引 self.image_index.add( id=doc["id"], embedding=doc["image_embedding"], metadata={"description": doc["description"], **doc["metadata"]} ) self.text_index.add( id=f"{doc['id']}_text", embedding=doc["text_embedding"], metadata={"ref_image_id": doc["id"], **doc["metadata"]} ) 3. 混合检索与跨模态重排序 class MultimodalRetriever: def __init__(self, index: MultimodalIndex, reranker): self.index = index self.reranker = reranker def retrieve(self, query: str, top_k: int = 10): # 1. 文本检索 text_hits = self.index.text_index.search( embedding=self.text_embedder.encode(query), top_k=top_k * 2 ) # 2. 用 CLIP 做跨模态检索（文本→图像） clip_query_emb = self.clip_encode_text(query) image_hits = self.index.image_index.search( embedding=clip_query_emb, top_k=top_k * 2 ) # 3. 合并候选 candidates = self._merge(text_hits, image_hits) # 4. 跨模态重排序 reranked = self.reranker.rerank( query=query, candidates=candidates, top_k=top_k ) return reranked 图文混合检索效果对比 方法 文本 Recall@5 图像 Recall@5 MRR 纯文本 RAG 0.82 0.00 0.71 纯图像 RAG 0.00 0.68 0.55 简单双路合并 0.80 0.65 0.73 双索引+重排序 0.88 0.79 0.82 双索引+VLM描述增强 0.91 0.85 0.88 实际案例：技术文档智能问答 以一份 200 页的技术白皮书为例，包含大量架构图和流程图： ...

引言 大模型的增量训练（Continued Pretraining，CPT）是指在已有预训练模型基础上，使用新数据继续训练，以注入新知识或适应新领域。与从零训练相比，增量训练大幅降低计算成本，同时保留模型的通用能力。但增量训练面临独特的技术挑战：知识注入与遗忘的平衡、训练稳定性、数据配比优化等。本文分享大模型增量训练的工程实践。 增量训练的定位 与其他训练方式的关系 训练方式 数据量 训练成本 知识注入深度 适用场景 从零预训练 万亿级token 极高 最深 构建全新基础模型 增量训练（CPT） 百亿级token 高 深 知识更新/领域注入 SFT微调 百万级 中 中 指令跟随能力 LoRA微调 万级 低 浅 轻量适配 何时需要增量训练 模型需要掌握大量新的领域知识（如从通用模型到医疗模型） 领域语言模式与通用文本差异大（如法律条文、代码） 需要更新时间敏感的知识（如新法规、新技术） 现有模型在目标领域的幻觉率过高 不推荐增量训练的场景：仅需调整输出格式（用SFT）、仅需少量领域知识（用RAG）、计算资源不足。 数据准备 数据来源与处理 class IncrementalDataPipeline: def __init__(self, tokenizer, max_length=4096): self.tokenizer = tokenizer self.max_length = max_length def process(self, raw_data_sources): """完整数据处理流水线""" all_data = [] for source in raw_data_sources: # 1. 格式统一 text = self.normalize_format(source) # 2. 质量过滤 if not self.quality_filter(text): continue # 3. 去重 text = self.deduplicate(text) # 4. 分词 tokens = self.tokenizer.encode(text) # 5. 长文档切分 chunks = self.split_long_text(tokens) all_data.extend(chunks) # 6. 数据配比 mixed_data = self.mix_data(all_data) return mixed_data def quality_filter(self, text): """质量过滤""" # 去除过短文本 if len(text) < 100: return False # 去除重复率高的文本 if self.repetition_ratio(text) > 0.3: return False # 去除乱码 if self.garbled_ratio(text) > 0.1: return False return True def split_long_text(self, tokens): """长文本切分，保留上下文""" chunks = [] overlap = 512 # 重叠token数 for i in range(0, len(tokens), self.max_length - overlap): chunk = tokens[i:i + self.max_length] chunks.append(chunk) if i + self.max_length >= len(tokens): break return chunks def mix_data(self, domain_data, general_ratio=0.3): """领域数据与通用数据混合""" domain_size = len(domain_data) general_size = int(domain_size * general_ratio / (1 - general_ratio)) general_data = self.load_general_data(general_size) mixed = domain_data + general_data random.shuffle(mixed) return mixed 数据配比策略 数据配比是增量训练最关键的超参数之一： ...