OpenAI文件搜索教程：用Responses API实现PDF RAG

本教程面向有编程基础的开发者，介绍如何使用OpenAI Responses API中的文件搜索工具，对PDF文档进行检索增强生成（RAG）操作。文章首先指导创建向量存储并上传PDF，然后演示独立向量搜索和集成到Responses API中的文件搜索。接着，通过生成评估数据集，计算Recall@k、Precision@k、MRR和MAP等指标，评估检索性能。最后，总结了简化RAG架构的优势。

技术背景

RAG（检索增强生成）是解决大语言模型知识时效性和专业领域知识不足的关键技术。传统RAG实现涉及PDF解析、文本分块、嵌入生成、向量数据库存储等多个步骤，流程复杂且维护成本高。OpenAI的Responses API提供了内置的文件搜索工具，将文件存储、嵌入和检索集成在一个API调用中，大大简化了RAG架构。该工具适用于需要基于私有文档回答问题的场景，如企业知识库、法律文档查询等。其主要限制包括依赖OpenAI的托管服务，数据隐私和成本需考虑。

核心原理

文件搜索工具的核心原理是：将PDF文件上传至OpenAI的向量存储，OpenAI自动解析PDF内容，将其分割成文本块，对每个文本块运行嵌入模型生成向量，并将向量和原始文本存储在向量数据库中。当用户查询时，系统使用混合搜索（结合关键词和向量相似度）检索最相关的文本块，并通过排序器计算相关性分数。在Responses API中，文件搜索工具作为工具被调用，检索结果作为上下文注入LLM，生成最终答案。

架构分析

系统架构包括三个主要组件：1) 向量存储（Vector Store）：用于存储PDF文件及其嵌入向量，由OpenAI托管；2) 文件搜索工具（file_search）：作为Responses API的工具，接收查询和向量存储ID，返回检索结果；3) Responses API：接收用户输入，调用文件搜索工具，并将检索结果与模型生成结合。数据流为：用户查询→Responses API→文件搜索工具→向量存储→检索结果→LLM生成答案。设计决策包括使用并行上传提高效率，以及通过tool_choice强制使用文件搜索以确保评估准确性。

完整实战教程

在Responses API中使用文件搜索工具

尽管RAG可能令人望而生畏，但搜索PDF文件不应复杂。目前最常用的方法之一是解析PDF、定义分块策略、将分块上传到存储提供商、对这些文本块运行嵌入，并将嵌入存储在向量数据库中。而这仅仅是设置——在我们的LLM工作流中检索内容也需要多个步骤。

这就是文件搜索——一个可以在Responses API中使用的托管工具——发挥作用的地方。它允许你搜索知识库，并根据检索到的内容生成答案。在本教程中，我们将把这些PDF上传到OpenAI上的向量存储，并使用文件搜索从该向量存储中获取额外上下文，以回答我们在第一步中生成的问题。然后，我们将根据从OpenAI博客（openai.com/news）提取的PDF创建一小部分问题。

_文件搜索以前在Assistants API中可用。现在它可在新的Responses API中使用，该API可以是有状态或无状态的，并具有元数据过滤等新功能。_

使用我们的PDF创建向量存储

!pip install PyPDF2 pandas tqdm openai -q

from openai import OpenAI
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
import concurrent
import PyPDF2
import os
import pandas as pd
import base64

client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
dir_pdfs = 'openai_blog_pdfs' # have those PDFs stored locally here
pdf_files = [os.path.join(dir_pdfs, f) for f in os.listdir(dir_pdfs)]

我们将在 OpenAI API 上创建一个向量存储，并将我们的 PDF 上传到该向量存储。OpenAI 将读取这些 PDF，将内容分割成多个文本块，对这些文本块运行嵌入，并将嵌入和文本存储在向量存储中。这将使我们能够查询此向量存储，根据查询返回相关内容。

def upload_single_pdf(file_path: str, vector_store_id: str):
    file_name = os.path.basename(file_path)
    try:
        file_response = client.files.create(file=open(file_path, 'rb'), purpose="assistants")
        attach_response = client.vector_stores.files.create(
            vector_store_id=vector_store_id,
            file_id=file_response.id
        )
        return {"file": file_name, "status": "success"}
    except Exception as e:
        print(f"Error with {file_name}: {str(e)}")
        return {"file": file_name, "status": "failed", "error": str(e)}

def upload_pdf_files_to_vector_store(vector_store_id: str):
    pdf_files = [os.path.join(dir_pdfs, f) for f in os.listdir(dir_pdfs)]
    stats = {"total_files": len(pdf_files), "successful_uploads": 0, "failed_uploads": 0, "errors": []}
    
    print(f"{len(pdf_files)} PDF files to process. Uploading in parallel...")

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(upload_single_pdf, file_path, vector_store_id): file_path for file_path in pdf_files}
        for future in tqdm(concurrent.futures.as_completed(futures), total=len(pdf_files)):
            result = future.result()
            if result["status"] == "success":
                stats["successful_uploads"] += 1
            else:
                stats["failed_uploads"] += 1
                stats["errors"].append(result)

    return stats

def create_vector_store(store_name: str) -> dict:
    try:
        vector_store = client.vector_stores.create(name=store_name)
        details = {
            "id": vector_store.id,
            "name": vector_store.name,
            "created_at": vector_store.created_at,
            "file_count": vector_store.file_counts.completed
        }
        print("Vector store created:", details)
        return details
    except Exception as e:
        print(f"Error creating vector store: {e}")
        return {}

store_name = "openai_blog_store"
vector_store_details = create_vector_store(store_name)
upload_pdf_files_to_vector_store(vector_store_details["id"])

独立向量搜索

现在我们的向量存储已经准备就绪，我们可以直接查询向量存储，并为特定查询检索相关内容。使用新的向量搜索 API，我们无需将其集成到 LLM 查询中，即可从知识库中找到相关条目。

query = "What's Deep Research?"
search_results = client.vector_stores.search(
    vector_store_id=vector_store_details['id'],
    query=query
)

for result in search_results.data:
    print(str(len(result.content[0].text)) + ' of character of content from ' + result.filename + ' with a relevant score of ' + str(result.score))

我们可以看到，搜索查询返回了不同大小（以及底层不同的文本）的结果。它们都有不同的相关性分数，这些分数是由我们的排序器使用混合搜索计算得出的。

在单个 API 调用中将搜索结果与 LLM 集成

然而，与其先查询向量存储，然后将数据传递给 Responses 或 Chat Completion API 调用，更便捷的方式是将 file_search 工具作为 OpenAI Responses API 的一部分来使用这些搜索结果。

query = "What's Deep Research?"
response = client.responses.create(
    input= query,
    model="gpt-4o-mini",
    tools=[{
        "type": "file_search",
        "vector_store_ids": [vector_store_details['id']],
    }]
)

# Extract annotations from the response
annotations = response.output[1].content[0].annotations
    
# Get top-k retrieved filenames
retrieved_files = set([result.filename for result in annotations])

print(f'Files used: {retrieved_files}')
print('Response:')
print(response.output[1].content[0].text) # 0 being the filesearch call

我们可以看到 gpt-4o-mini 能够回答一个需要关于 OpenAI Deep Research 的更新、更专业知识的查询。它使用了文件 Introducing deep research _ OpenAI.pdf 中的内容，其中包含最相关的文本块。如果我们想更深入地分析检索到的文本块，我们还可以通过向查询中添加 include=["output[*].file_search_call.search_results"] 来分析搜索引擎返回的不同文本。

评估性能

对于这些信息检索系统来说，关键还在于衡量检索到的文件与这些答案的相关性和质量。本手册的以下步骤将包括生成一个评估数据集，并在此生成的数据集上计算不同的指标。这是一种不完美的方法，我们始终建议您针对自己的用例使用人工验证的评估数据集，但这将向您展示评估这些系统的方法论。它可能不完美，因为生成的一些问题可能是通用的（例如：本文档中主要利益相关者说了什么），而我们的检索测试将很难确定该问题是为哪个文档生成的。

生成评估数据

我们将创建一些函数，这些函数会读取我们本地已有的 PDF 文件，并生成一个只能由该文档回答的问题。因此，这将创建我们的评估数据集，供我们之后使用。

def extract_text_from_pdf(pdf_path):
    text = ""
    try:
        with open(pdf_path, "rb") as f:
            reader = PyPDF2.PdfReader(f)
            for page in reader.pages:
                page_text = page.extract_text()
                if page_text:
                    text += page_text
    except Exception as e:
        print(f"Error reading {pdf_path}: {e}")
    return text

def generate_questions(pdf_path):
    text = extract_text_from_pdf(pdf_path)

    prompt = (
        "Can you generate a question that can only be answered from this document?:\n"
        f"{text}\n\n"
    )

    response = client.responses.create(
        input=prompt,
        model="gpt-4o",
    )

    question = response.output[0].content[0].text

    return question

如果我们对第一个 PDF 文件运行 generate_question 函数，我们将能够看到它生成的问题类型。

generate_questions(pdf_files[0])

现在，我们可以为本地存储的所有PDF生成所有问题了。

# Generate questions for each PDF and store in a dictionary
questions_dict = {}
for pdf_path in pdf_files:
    questions = generate_questions(pdf_path)
    questions_dict[os.path.basename(pdf_path)] = questions

questions_dict

现在我们有了一个 filename:question 的字典，可以循环遍历并向 gpt-4o(-mini) 提问，而无需提供文档，gpt-4o 应该能够在向量存储中找到相关文档。

评估

我们将把字典转换为数据框，并使用 gpt-4o-mini 进行处理。我们将留意预期的文件。

rows = []
for filename, query in questions_dict.items():
    rows.append({"query": query, "_id": filename.replace(".pdf", "")})

# Metrics evaluation parameters
k = 5
total_queries = len(rows)
correct_retrievals_at_k = 0
reciprocal_ranks = []
average_precisions = []

def process_query(row):
    query = row['query']
    expected_filename = row['_id'] + '.pdf'
    # Call file_search via Responses API
    response = client.responses.create(
        input=query,
        model="gpt-4o-mini",
        tools=[{
            "type": "file_search",
            "vector_store_ids": [vector_store_details['id']],
            "max_num_results": k,
        }],
        tool_choice="required" # it will force the file_search, while not necessary, it's better to enforce it as this is what we're testing
    )
    # Extract annotations from the response
    annotations = None
    if hasattr(response.output[1], 'content') and response.output[1].content:
        annotations = response.output[1].content[0].annotations
    elif hasattr(response.output[1], 'annotations'):
        annotations = response.output[1].annotations

    if annotations is None:
        print(f"No annotations for query: {query}")
        return False, 0, 0

    # Get top-k retrieved filenames
    retrieved_files = [result.filename for result in annotations[:k]]
    if expected_filename in retrieved_files:
        rank = retrieved_files.index(expected_filename) + 1
        rr = 1 / rank
        correct = True
    else:
        rr = 0
        correct = False

    # Calculate Average Precision
    precisions = []
    num_relevant = 0
    for i, fname in enumerate(retrieved_files):
        if fname == expected_filename:
            num_relevant += 1
            precisions.append(num_relevant / (i + 1))
    avg_precision = sum(precisions) / len(precisions) if precisions else 0
    
    if expected_filename not in retrieved_files:
        print("Expected file NOT found in the retrieved files!")
        
    if retrieved_files and retrieved_files[0] != expected_filename:
        print(f"Query: {query}")
        print(f"Expected file: {expected_filename}")
        print(f"First retrieved file: {retrieved_files[0]}")
        print(f"Retrieved files: {retrieved_files}")
        print("-" * 50)
    
    
    return correct, rr, avg_precision

process_query(rows[0])

在这个例子中，召回率和精确率都为1，我们的文件排名第一，因此在这个例子中MRR和MAP都为1。

我们现在可以在我们的问题集上执行这个处理。

with ThreadPoolExecutor() as executor:
    results = list(tqdm(executor.map(process_query, rows), total=total_queries))

correct_retrievals_at_k = 0
reciprocal_ranks = []
average_precisions = []

for correct, rr, avg_precision in results:
    if correct:
        correct_retrievals_at_k += 1
    reciprocal_ranks.append(rr)
    average_precisions.append(avg_precision)

recall_at_k = correct_retrievals_at_k / total_queries
precision_at_k = recall_at_k  # In this context, same as recall
mrr = sum(reciprocal_ranks) / total_queries
map_score = sum(average_precisions) / total_queries

上面记录的输出要么显示某个文件在我们的评估数据集期望它排名第一时没有排名第一，要么根本没有被找到。正如我们从并不完美的评估数据集中看到的，有些问题比较泛化，期望的是另一篇文档，而我们的检索系统并没有专门针对该问题检索到它。

# Print the metrics with k
print(f"Metrics at k={k}:")
print(f"Recall@{k}: {recall_at_k:.4f}")
print(f"Precision@{k}: {precision_at_k:.4f}")
print(f"Mean Reciprocal Rank (MRR): {mrr:.4f}")
print(f"Mean Average Precision (MAP): {map_score:.4f}")

通过本教程，我们了解了如何：

使用PDF上下文填充（利用4o的视觉模态）和传统PDF阅读器生成评估数据集
创建向量存储并用PDF填充
利用OpenAI Response API中开箱即用的file_search工具调用，通过RAG系统获取LLM对查询的答案
理解文本块如何被检索、排序并作为Response API的一部分使用
在之前生成的评估数据集上衡量准确率、精确率、召回率、MRR和MAP

通过将文件搜索与Responses结合使用，您可以简化RAG架构，并利用新的Responses API在单次API调用中实现这一功能。文件存储、嵌入和检索全部集成在一个工具中！

常见问题

常见错误：1) 文件上传失败，可能因文件格式不支持或网络问题，需检查文件路径和API密钥；2) 向量存储创建失败，需确认API权限和配额；3) 检索结果为空，可能因查询与文档内容不匹配，或向量存储中文件未处理完成；4) 评估时预期文件未检索到，可能因问题过于泛化或检索参数不当。排查方法：检查API响应错误信息，确认文件状态，调整max_num_results参数，使用include参数查看详细检索结果。生产环境优化：使用批量上传和并行处理，设置合理的max_num_results，定期更新向量存储，监控API调用成本。

企业应用扩展

本教程展示了文件搜索的基本用法，可进一步扩展：1) 结合元数据过滤，实现更精细的检索；2) 使用有状态会话保持上下文；3) 集成到更复杂的RAG流程中，如多轮对话；4) 评估数据集可人工标注以提高准确性；5) 探索其他模型（如gpt-4o）对检索质量的影响；6) 考虑使用Azure OpenAI服务以满足合规要求。

总结

内容来源与验证

内容等级：PREMIUM_A
质量等级：A
难度：中级
验证等级：V1
原文作者：pap-openai
原文日期：2025-03-11
来源许可：MIT
来源版本：e8e77f12b774eae51f8fb95a23befe334ca41b8e
查看原文

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员