文昌市网站建设_网站建设公司_交互流畅度_seo优化

一、核心设计原则

1. 整页为单 Chunk：将单页保险文档作为 1 个检索单元（Chunk），保留内容逻辑关联性；

2. 元数据对齐：文档入库的元数据字段与提问提取的元数据字段完全一致，确保过滤检索精准；

3. 混合检索：元数据过滤（精准定位 Chunk）+ 向量 / 关键词检索（匹配 Chunk 内内容），兼顾精度与效率。

二、流程总览

原始保险文档（OCR文本）→ 提取文档元数据 → 整页Chunk+元数据上传至Dify知识库 ↑ ↓ 用户提问 → 提取提问元数据 → 元数据过滤检索Dify知识库 → 获取匹配Chunk → LLM生成回答

三、第一步：文档元数据提取 + 整页 Chunk 入库（Dify）

1. 定义通用元数据字段（保险类文档适配）

2. 文档元数据提取函数（LLM 驱动）

import os import json import requests from openai import OpenAI from dotenv import load_dotenv # 环境变量加载（通用配置） load_dotenv() LLM_API_KEY = os.getenv("LLM_API_KEY") DIFY_API_KEY = os.getenv("DIFY_API_KEY") DIFY_BASE_URL = os.getenv("DIFY_BASE_URL", "https://api.dify.ai/v1") DIFY_KNOWLEDGE_BASE_ID = os.getenv("DIFY_KNOWLEDGE_BASE_ID") # 初始化LLM客户端（通用，适配OpenAI/国产模型） llm_client = OpenAI(api_key=LLM_API_KEY) def extract_document_metadata(ocr_text): """ 通用函数：从保险文档OCR文本中提取结构化元数据 :param ocr_text: 单页保险文档的OCR文本（动态输入） :return: 通用化元数据字典 """ prompt = f""" # 任务：提取保险类文档的RAG检索专用元数据（整页为1个Chunk） # 输入文本： {ocr_text} # 提取规则： 1. 严格基于文本内容，未提及的字段填空字符串/空数组，不编造任何信息； 2. doc_type：提取文档类型（如保险产品介绍、保险条款）； 3. core_tags：提取所有可用于检索的核心关键词（如收益、权益、规则、缴费方式）； 4. data_modules：提取文档包含的逻辑模块（从["产品基础","收益案例","权益规则","条款约束","提取规则","退保规则"]中选择）； 5. key_numbers：提取所有带单位的核心数值（如年限、金额、百分比）； 6. 输出仅保留标准JSON，无解释性文字、无换行。 # 输出JSON格式： {{ "doc_type": "", "issuer": "", "update_time": "", "applicable_area": [], "supported_currencies": [], "core_tags": [], "data_modules": [], "key_numbers": [] }} """ try: response = llm_client.chat.completions.create( model="gpt-3.5-turbo", # 可替换为国产模型（如通义千问、文心一言） messages=[ {"role": "system", "content": "你是保险文档元数据提取专家，输出仅符合格式的JSON"}, {"role": "user", "content": prompt} ], temperature=0.0, # 无幻觉，严格基于文本提取 response_format={"type": "json_object"}, timeout=10 ) metadata = json.loads(response.choices[0].message.content) # 空值清洗（确保格式统一） for key in metadata: if isinstance(metadata[key], list) and len(metadata[key]) == 0: metadata[key] = [] elif isinstance(metadata[key], str) and metadata[key].strip() == "": metadata[key] = "" return metadata except Exception as e: print(f"文档元数据提取失败：{e}") # 返回空元数据兜底 return { "doc_type": "", "issuer": "", "update_time": "", "applicable_area": [], "supported_currencies": [], "core_tags": [], "data_modules": [], "key_numbers": [] }

3. 整页 Chunk 上传至 Dify 知识库

def upload_full_page_to_dify(full_page_text, metadata, doc_unique_id): """ 通用函数：将整页文档作为1个Chunk上传至Dify知识库 :param full_page_text: 整页OCR文本 :param metadata: 提取的文档元数据 :param doc_unique_id: 文档唯一标识（如"insurance_doc_001"） """ url = f"{DIFY_BASE_URL}/knowledge_bases/{DIFY_KNOWLEDGE_BASE_ID}/documents/batch" headers = { "Authorization": f"Bearer {DIFY_API_KEY}", "Content-Type": "application/json" } # 构造Dify上传请求体（单Chunk） documents = [ { "content": full_page_text, # 整页文本作为1个Chunk "metadata": metadata, # 绑定通用元数据 "document_id": doc_unique_id, # 自定义唯一ID（便于管理） "name": f"{metadata['doc_type']}_{doc_unique_id}" # 文档名称 } ] payload = { "documents": documents, "mode": "overwrite" # 可选：append（追加）/overwrite（覆盖） } try: response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() print(f"整页Chunk上传成功（ID：{doc_unique_id}）") except Exception as e: print(f"Chunk上传失败：{e}") if hasattr(e, 'response'): print(f"错误详情：{e.response.text}") # 文档入库主函数 def document_ingestion_pipeline(ocr_text, doc_unique_id): """ 文档入库流程：提取元数据 → 上传Chunk :param ocr_text: 整页OCR文本 :param doc_unique_id: 文档唯一ID """ # 步骤1：提取文档元数据 doc_metadata = extract_document_metadata(ocr_text) # 步骤2：上传整页Chunk+元数据 upload_full_page_to_dify(ocr_text, doc_metadata, doc_unique_id)

四、第二步：用户提问元数据提取（对齐文档元数据）

1. 提问元数据提取函数（字段与文档元数据完全对齐）

def extract_query_metadata(user_query): """ 通用函数：从用户提问中提取Dify检索用的元数据（字段与文档元数据对齐） :param user_query: 用户原始提问（口语化/精准化均可） :return: 结构化提问元数据（用于Dify过滤检索） """ prompt = f""" # 任务：从用户提问中提取保险类文档RAG检索的过滤元数据 # 核心规则： 1. 严格基于用户提问内容提取，未提及的字段填空字符串/空数组，不推测、不编造； 2. 字段值需与保险文档元数据格式对齐（如货币名称、模块名称统一）； 3. doc_type：提取用户提问指向的文档类型（如保险产品介绍）； 4. core_tags：提取提问中的核心检索关键词（如缴费方式、权益、金额、年限）； 5. data_modules：提取提问指向的逻辑模块（从["产品基础","收益案例","权益规则","条款约束"]中选择）； 6. key_numbers：提取提问中的核心数值（带单位）； 7. 输出仅保留标准JSON，无其他内容。 # 用户提问： {user_query} # 输出JSON格式： {{ "doc_type": "", "issuer": "", "update_time": "", "applicable_area": [], "supported_currencies": [], "core_tags": [], "data_modules": [], "key_numbers": [] }} """ try: response = llm_client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "你是保险提问元数据提取助手，输出仅符合格式的JSON"}, {"role": "user", "content": prompt} ], temperature=0.0, response_format={"type": "json_object"}, timeout=10 ) query_metadata = json.loads(response.choices[0].message.content) # 空值清洗 for key in query_metadata: if isinstance(query_metadata[key], list) and len(query_metadata[key]) == 0: query_metadata[key] = [] elif isinstance(query_metadata[key], str) and query_metadata[key].strip() == "": query_metadata[key] = "" return query_metadata except Exception as e: print(f"提问元数据提取失败：{e}") # 返回空元数据兜底 return { "doc_type": "", "issuer": "", "update_time": "", "applicable_area": [], "supported_currencies": [], "core_tags": [], "data_modules": [], "key_numbers": [] }

五、第三步：元数据过滤检索 + LLM 生成回答

1. Dify 知识库检索（元数据过滤 + 混合检索）

def retrieve_from_dify(query_metadata, user_query): """ 通用函数：调用Dify检索API，基于提问元数据过滤Chunk :param query_metadata: 提问元数据 :param user_query: 用户原始提问（用于向量/关键词检索） :return: Dify检索结果（匹配的Chunk列表） """ # 构造过滤条件（仅保留非空字段，减少无效过滤） filter_conditions = {} for key, value in query_metadata.items(): if value != "" and value != []: filter_conditions[key] = value # Dify检索API参数 url = f"{DIFY_BASE_URL}/knowledge_bases/{DIFY_KNOWLEDGE_BASE_ID}/retrieve" headers = { "Authorization": f"Bearer {DIFY_API_KEY}", "Content-Type": "application/json" } payload = { "query": user_query, # 用户提问（向量/关键词检索） "top_k": 3, # 返回Top3匹配的Chunk "filter": filter_conditions, # 元数据过滤条件（对齐文档元数据） "retrieval_mode": "hybrid", # 混合检索（关键词+向量，兼顾精度） "score_threshold": 0.3 # 相似度阈值（过滤低匹配结果） } try: response = requests.post(url, headers=headers, json=payload, timeout=20) response.raise_for_status() return response.json() except Exception as e: print(f"Dify检索失败：{e}") if hasattr(e, 'response'): print(f"错误详情：{e.response.text}") return None

2. LLM 生成回答（基于检索到的 Chunk）

def generate_answer(retrieve_result, user_query): """ 通用函数：基于检索到的Chunk生成精准回答 :param retrieve_result: Dify检索结果 :param user_query: 用户原始提问 :return: 结构化回答 """ # 无匹配结果兜底 if not retrieve_result or len(retrieve_result["documents"]) == 0: return "未检索到与您的问题匹配的保险文档信息，请调整提问关键词。" # 提取检索到的Chunk内容（整页文本） retrieved_content = "\n\n".join([doc["content"] for doc in retrieve_result["documents"]]) # 生成回答的Prompt（通用化，无具体产品） answer_prompt = f""" # 任务：基于保险文档信息回答用户问题 # 文档信息： {retrieved_content} # 回答规则： 1. 仅使用提供的文档信息回答，不编造任何内容； 2. 回答简洁准确，聚焦用户问题核心，忽略无关信息； 3. 若文档中无明确答案，明确说明“文档中未提及相关信息”； 4. 涉及数值/规则的，需标注“非保证”等文档中的约束条件（如有）。 # 用户问题： {user_query} """ try: response = llm_client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "你是专业的保险文档解答助手，回答严格基于提供的信息"}, {"role": "user", "content": answer_prompt} ], temperature=0.1 # 低随机性，确保回答精准 ) return response.choices[0].message.content except Exception as e: print(f"回答生成失败：{e}") return "回答生成失败，请重试。"

六、第四步：全流程串联（通用 RAG 问答管道）

def insurance_rag_qa_pipeline(user_query, ocr_text=None, doc_unique_id=None): """ 保险类文档RAG全流程： 1. 若传入OCR文本+文档ID，先执行文档入库； 2. 提取提问元数据 → 检索 → 生成回答 """ # 可选：文档入库（首次上传时执行） if ocr_text and doc_unique_id: document_ingestion_pipeline(ocr_text, doc_unique_id) # 核心流程：提问处理 → 检索 → 回答 # 步骤1：提取提问元数据 query_metadata = extract_query_metadata(user_query) # 步骤2：Dify元数据过滤检索 retrieve_result = retrieve_from_dify(query_metadata, user_query) # 步骤3：生成回答 final_answer = generate_answer(retrieve_result, user_query) return final_answer # 全流程测试（示例） if __name__ == "__main__": # 示例1：文档入库（首次上传） sample_ocr_text = """【保险产品介绍】 发行机构：XX保险公司 更新时间：2024年8月 支持货币：美元、港元、欧元 核心收益：长期总内部回报率预期超7%，回本周期短至8年 权益规则：支持货币转换、红利锁/解锁、受保人变更 条款约束：实际收益非保证，提取金额需符合保单规则 """ # 执行入库（仅首次执行） insurance_rag_qa_pipeline( user_query="", # 提问为空，仅执行入库 ocr_text=sample_ocr_text, doc_unique_id="insurance_doc_001" ) # 示例2：用户提问检索+回答 user_query = "美元保单的长期IRR是多少？是否有保证？" answer = insurance_rag_qa_pipeline(user_query=user_query) print("=== 最终回答 ===") print(answer)

七、通用化优化建议（适配所有保险类文档）

1. 元数据扩展

可根据实际需求新增通用字段，如：

• payment_methods：缴费方式（数组，如 [“5 年缴”,“10 年缴”,“整付”]）；

• right_types：权益类型（数组，如 [“货币转换”,“红利解锁”,“受保人变更”]）；

• constraint_tags：约束标签（数组，如 [“非保证收益”,“退保条件限制”]）。

2. 适配国产 LLM

若不用 OpenAI，替换llm_client为国产模型调用逻辑（如通义千问、文心一言），Prompt 模板完全通用：

# 通义千问适配示例 import dashscope dashscope.api_key = os.getenv("DASHSCOPE_API_KEY") def extract_document_metadata(ocr_text): prompt = "..." # 保留原Prompt response = dashscope.Generation.call( model="qwen-plus", messages=[{"role": "user", "content": prompt}], result_format="json", temperature=0.0 ) metadata = json.loads(response.output.choices[0].message.content) return metadata

3. 批量处理优化

def batch_ingestion(folder_path): """批量入库文件夹中的保险文档""" import glob for idx, file_path in enumerate(glob.glob(f"{folder_path}/\*.txt")): with open(file_path, "r", encoding="utf-8") as f: ocr_text = f.read() doc_unique_id = f"insurance_doc_{idx:03d}" document_ingestion_pipeline(ocr_text, doc_unique_id)

4. Dify 检索配置

• 检索模式：选择「混合检索」，兼顾元数据关键词和向量相似度；

• 向量模型：选择支持长文本的模型（如text-embedding-3-large、m3e-large）；

• 过滤逻辑：Dify 支持「数组包含匹配」（如supported_currencies包含 “美元” 即匹配），无需完全一致。

总结

该方案实现了完全通用化的保险类文档 RAG 全流程：

1. 文档侧：整页为 Chunk + 提取通用元数据，无需拆分，适配任意保险文档；

2. 提问侧：提取与文档元数据对齐的检索标签，精准过滤 Chunk；

3. 检索侧：元数据过滤 + 混合检索，兼顾精度与效率；

4. 回答侧：基于检索结果生成精准回答，无编造、无冗余。

全流程无具体产品名称依赖，可直接复用至各类保险产品文档的 RAG 系统开发。