告别无效检索：我用LangExtract + Milvus升级 RAG 管道的实战复盘

pip install langextract google-generativeai

import os# 在实际项目中，请使用 .env 文件或系统环境变量等更安全的方式os.environ["LANGEXTRACT_API_KEY"]="YOUR_GOOGLE_API_KEY"

import langextract as lximport textwrap
# 步骤1：定义提取任务prompt = textwrap.dedent("""从技术文档中精准提取以下字段:1. service_name:从标题中提取出的主要服务或API名称(例如："认证API","存储服务")2. version_number:仅提取版本号数字(例如："2.0","1.0")，不要包含"v"或"版本"等文字。3. document_category:文档类型，提取原文中的"参考"、"指南"或"排错指南"。4. rate_limits:任何关于速率限制的信息，提取完整描述。""")
# 步骤2：提供高质量示例examples =[    lx.data.ExampleData(        text="# 支付API v3.0 参考\n\n支付API用于处理所有交易请求。\n\n速率限制: 每分钟500次请求。",        extractions=[            lx.data.Extraction(extraction_class="service_name", extraction_text="支付API"),            lx.data.Extraction(extraction_class="version_number", extraction_text="3.0"),            lx.data.Extraction(extraction_class="document_category", extraction_text="参考"),            lx.data.Extraction(extraction_class="rate_limits", extraction_text="每分钟500次请求")        ]    ),    lx.data.ExampleData(        text="## 用户认证服务 - 排错指南\n\n本文档适用于 v1.5 版本。如果遇到认证失败，请检查您的API密钥。\n\n速率限制: 无。",        extractions=[            lx.data.Extraction(extraction_class="service_name", extraction_text="用户认证服务"),            lx.data.Extraction(extraction_class="version_number", extraction_text="1.5"),            lx.data.Extraction(extraction_class="document_category", extraction_text="排错指南"),            lx.data.Extraction(extraction_class="rate_limits", extraction_text="无")        ]    )]
# 步骤3：在新的输入文本上执行提取input_text = textwrap.dedent("""       # 图像处理服务 - 开发者指南       欢迎使用我们的图像处理服务。当前稳定版本为2.1。       本指南将引导您完成所有设置步骤。       ### API调用频率       为了保证服务稳定，我们对所有用户设置了统一的调用频率上限：每小时1000次调用。""")
result_doc = lx.extract(    text_or_documents=input_text,    prompt_description=prompt,    examples=examples,    model_id="gemini-1.5-flash-latest",)
# 步骤4：处理结果if result_doc.extractions:    extracted_metadata ={ext.extraction_class: ext.extraction_text for ext in result_doc.extractions}    print("提取出的结构化元数据:")    for key, value in extracted_metadata.items():       print(f"  - {key}: {value}")else:    print("未能提取出任何信息。")
# 步骤5 (可选): 可视化调试lx.io.save_annotated_documents([result_doc], output_name="extraction_results.jsonl", output_dir=".")html_content = lx.visualize("extraction_results.jsonl")with open("visualization.html","w", encoding="utf-8")as f:    f.write(html_content.data)

提取出的结构化元数据:service_name:图像处理服务version_number:2.1document_category:指南rate_limits:每小时1000次调用

import textwrap
prompt = textwrap.dedent("""从电影简介中，找到核心角色，并提取出它的主要类型、主角类型和核心主题。""")

examples =[    lx.data.ExampleData(        text="一个孤独的太空陆战队员，在一颗遥远的星球上与外星生物作战。",        extractions=[            lx.data.Extraction(                extraction_class="protagonist",# 我们给这个提取对象一个类名                extraction_text="太空陆战队员",# 我们希望提取出的关键文本                attributes={# 我们希望关联到这段文本上的元数据                   "genre":"科幻",                   "character_type":"军人",                   "theme":"战斗"                }            )        ]     )]

input ="一位年轻的巫师在霍格沃茨魔法学校发现了自己的魔法天赋，并与朋友们一起对抗黑魔王。"
result_doc = lx.extract(    text_or_documents=input,    prompt_description=prompt,    examples=examples,    model_id="gemini-1.5-flash-latest",)
if result_doc.extractions:    extraction = result_doc.extractions[0]    print(f"提取出的关键实体文本: {extraction.extraction_text}")    print(f"关联的结构化元数据: {extraction.attributes}")else:    print("未能提取出任何信息。")

提取出的关键实体文本:年轻的巫师关联的结构化元数据:{'genre':'奇幻','character_type':'魔法师','theme':'魔法与对抗'}

pip install pymilvus

from pymilvus importMilvusClient,DataType,FieldSchema,CollectionSchemaimport google.generativeai as genai
# 配置 genai，它同样会读取 GEMINI_API_KEYgenai.configure(api_key=os.environ["GEMINI_API_KEY"])
# --- Milvus 设置 (使用本地文件模式，对新手友好，无需服务) ---COLLECTION_NAME ="movie_db_prod"DB_FILE ="./milvus_movies.db"client =MilvusClient(uri=DB_FILE)
if client.has_collection(COLLECTION_NAME):    client.drop_collection(COLLECTION_NAME)
# --- 定义 Schema ---# 为主角(protagonist)创建一个专门的字段，其余元数据通过 dynamic field 自动映射fields =[   FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),   FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),   FieldSchema(name="protagonist", dtype=DataType.VARCHAR, max_length=1024),   FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)]schema =CollectionSchema(fields, enable_dynamic_field=True)client.create_collection(COLLECTION_NAME, schema=schema)
# --- 创建索引 ---index_params = client.prepare_index_params()index_params.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="L2")client.create_index(COLLECTION_NAME, index_params)
# --- 准备数据 ---input =["一个孤独的太空陆战队员，在一颗遥远的星球上与外星生物作战。","纽约警探约翰·麦克连在圣诞夜与恐怖分子在一栋洛杉矶摩天大楼里展开激战。","年轻巫师哈利·波特在霍格沃茨发现自己的魔法天赋，并对抗黑魔王。","一位天才发明家托尼·斯塔克打造了一套高科技战甲，成为钢铁侠。",]
# (复用第二部分中的 prompt 和 examples)all_data_entries =[]for plot in input:    # 1. 提取元数据    doc = lx.extract(plot, prompt_description=prompt, examples=examples, model_id="gemini-1.5-flash-latest")    metadata ={}    protagonist_text =""    if doc.extractions:        metadata = doc.extractions[0].attributes        protagonist_text = doc.extractions[0].extraction_text
    # 2. 生成向量    embedding = genai.embed_content(model="models/text-embedding-004", content=plot, task_type="RETRIEVAL_DOCUMENT")['embedding']
    # 3. 准备注入数据    data_entry ={        "text": plot,        "protagonist": protagonist_text,        "embedding": embedding,    }    data_entry.update(metadata)# 将 genre, theme 等动态字段合并进来    all_data_entries.append(data_entry)
# 4. 批量注入 Milvusclient.insert(collection_name=COLLECTION_NAME, data=all_data_entries)print(f"成功注入 {len(all_data_entries)} 条数据到 Milvus!")

results = client.query(    collection_name=COLLECTION_NAME,    filter='genre == "动作"',# 使用示例中存在的类型以确保有结果    output_fields=["text","genre","protagonist"])print("精确查询结果：\n", results)

query_text ="英雄的成长与冒险"query_vector = genai.embed_content(model="models/text-embedding-004", content=query_text, task_type="RETRIEVAL_QUERY")['embedding']
search_results = client.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    limit=2,    filter='genre == "奇幻"',# 先用元数据把范围缩小    output_fields=["text","genre","protagonist"]# 再在这个小范围里做向量搜索)print("\n混合搜索结果：\n", search_results)