欢迎！本笔记本演示了如何使用 OpenAI 的 Vision 和 Responses API 构建检索增强生成（RAG）系统。它侧重于多模态数据，结合图像和文本输入来分析客户体验。该系统利用 GPT-4.1 并将图像理解与文件搜索相结合，以提供上下文感知的响应。
多模态数据集越来越普遍，尤其是在医疗保健等领域，那里的记录通常包含视觉数据（例如放射学扫描）和伴随的文本（例如临床笔记）。真实世界的数据集也往往是嘈杂的，信息不完整或缺失，因此对多种模态进行协同分析至关重要。
本指南侧重于客户服务用例：评估可能包含照片和书面评论的客户反馈。您将学习如何合成生成图像和文本输入，使用文件搜索进行上下文检索，并应用 Evals API 来评估包含图像理解如何影响整体性能。
概述
目录
- 设置和依赖项
- 示例生成
- 数据处理 - 加载合成数据集 - 合并数据
- 填充向量存储 - 上传数据以进行文件搜索 - 设置属性过滤器
- 检索和过滤 - 测试检索性能 - 应用基于属性的过滤器
- 评估和分析 - 将预测与真实情况进行比较 - 分析性能指标
设置和依赖项
%pip install openai evals pandas numpy matplotlib tqdm ipython --upgrade --quiet
import base64
from io import BytesIO
import os
from pathlib import Path
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from openai import OpenAI
from IPython.display import display, Image
from tqdm.notebook import tqdm
cache_dir = Path('.local_cache')
cache_dir.mkdir(parents=True, exist_ok=True)
client = OpenAI()
示例生成
为机器学习任务生成高质量的训练和评估数据可能成本高昂且耗时。合成数据提供了一种实用且可扩展的替代方案。在此笔记本中，OpenAI Image API 用于生成合成图像，而 Responses API 用于创建合成文本，从而能够跨多模态任务进行高效的原型设计和实验。
prompt = ("Gourmet pasta neatly plated with garnish and sides on a white ceramic plate, "
"photographed from above on a restaurant table. Soft shadows and vibrant colors.")
cache_path = f".local_cache/{hash(prompt)}.png"
if not os.path.exists(cache_path):
response = client.images.generate(
model="gpt-image-1",
prompt=prompt,
size="1024x1024"
)
with open(cache_path, "wb") as f:
f.write(base64.b64decode(response.data[0].b64_json))
print(f"Generated and cached: {cache_path}")
else:
print(f"Loading from cache: {cache_path}")
display(Image(filename=cache_path))
def generate_food_delivery_review(sentiment: str = 'positive') -> str:
"""
Generate a synthetic food delivery review with the specified sentiment.
Args:
sentiment: An adjective such as 'positive' or 'negative'.
Returns:
Generated review text
"""
prompt = "Write a very concise, realistic customer review for a recent food delivery."
prompt += f" The review should reflect a {sentiment} experience."
response = client.responses.create(
model="gpt-4.1",
input=[{"role": "user", "content": prompt}]
)
return response.output_text
review = generate_food_delivery_review()
print(review)
数据处理
在此示例中，我们将处理一个预先生成的合成客户反馈数据集，其中包含简短的文本片段、来自客户评论的图像，以及偶尔组合的多模态条目。您还可以使用上面提供的示例生成自己的合成数据集，以根据您的具体用例定制数据。
# Download the dataset
! mkdir -p .local_cache/images
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/df.csv -O .local_cache/df.csv
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/1.png -O .local_cache/images/1.png
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/2.png -O .local_cache/images/2.png
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/3.png -O .local_cache/images/3.png
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/4.png -O .local_cache/images/4.png
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/5.png -O .local_cache/images/5.png
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/6.png -O .local_cache/images/6.png
! wget https://raw.githubusercontent.com/robtinn/image_understanding_rag_dataset/main/data/images/7.png -O .local_cache/images/7.png
def encode_image(image_path: str) -> str:
"""Encode image file to base64 string."""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def analyze_image_sentiment(image_path: str) -> str:
"""Analyze food delivery image and return sentiment analysis."""
base64_image = encode_image(image_path)
response = client.responses.create(
model="gpt-4.1",
input=[{
"role": "user",
"content": [
{
"type": "input_text",
"text": "Analyze this food delivery image. Respond with a brief description and sentiment (positive/negative) in one line."
},
{
"type": "input_image",
"image_url": f"data:image/jpeg;base64,{base64_image}",
},
],
}],
max_output_tokens=50,
temperature=0.2
)
return response.output_text.strip()
df = pd.read_csv(".local_cache/df.csv")
cache_dir = Path(".local_cache")
for idx, row in df[~df['image_path'].isna()].iterrows():
image_path = cache_dir / 'images' / row['image_path']
sentiment = analyze_image_sentiment(str(image_path))
df.at[idx, 'full_sentiment'] = f"{row['text']} {sentiment}" if pd.notna(row['text']) else sentiment
print(f"Processed {row['image_path']}")
df['full_sentiment'] = df['full_sentiment'].fillna(df['text'])
output_path = cache_dir / "df_full_sentiment.csv"
df.to_csv(output_path, index=False)
print(f"\nSaved results to {output_path}")
pd.set_option('display.max_colwidth', 100) # Increase from default (50) to view full sentiment
display(df.head())
填充向量存储
本示例使用 OpenAI 的内置向量存储和文件搜索功能来构建一个 RAG 系统，该系统可以分析客户反馈（可以是视觉和文本形式）中的客户体验。我们创建了两个向量存储进行比较，一个具有图像理解功能，一个没有。
text_vector_store = client.vector_stores.create(
name="food_delivery_reviews_text",
metadata={
"purpose": "text_understanding",
"created_by": "notebook",
"version": "1.0"
}
)
text_vector_store_id = text_vector_store.id
text_image_vector_store = client.vector_stores.create(
name="food_delivery_reviews_text_image",
metadata={
"purpose": "text_image_understanding",
"created_by": "notebook",
"version": "1.0"
}
)
text_image_vector_store_id = text_image_vector_store.id
print("Vector Store IDs:")
print(f" Text: {text_vector_store_id}")
print(f" Text+Image: {text_image_vector_store_id}")
# upload files to vector database and set metadata
def upload_files_to_vector_store(vector_store_id, df, column_name="full_sentiment"):
file_ids = []
for i, row in tqdm(df.iterrows(), total=len(df), desc="Uploading context files"):
if pd.isna(row[column_name]):
file_stream = BytesIO('No information available.'.encode('utf-8'))
else:
file_stream = BytesIO(row[column_name].encode('utf-8'))
file_stream.name = f"context_{row.get('id', i)}_{row.get('month', '')}.txt"
file = client.vector_stores.files.upload(
vector_store_id=vector_store_id,
file=file_stream
)
file_ids.append(file.id)
for i, row in tqdm(df.iterrows(), total=len(df), desc="Updating file attributes"):
client.vector_stores.files.update(
vector_store_id=vector_store_id,
file_id=file_ids[i],
attributes={"month": row["month"]}
)
upload_files_to_vector_store(text_image_vector_store_id, df)
upload_files_to_vector_store(text_vector_store_id, df, column_name="text")
检索和过滤
借助文件搜索，我们可以使用自然语言查询来分析我们的数据集。对于仅文本数据集，我们看到信息缺失，这可能会影响我们的分析。
七月份关于意大利面的唯一正面评论包含视觉反馈，我们可以看到只有文本上下文的 RAG 系统不确定正面细节。然而，在提供图像上下文后，第二个 RAG 系统能够提供更准确的响应。
# Query the vector store for spaghetti reviews in July
query = "Where there any comments about the 'spaghetti'?"
print(f"🔍 Query: {query}\n")
# Execute the search with filtering
response = client.responses.create(
model="gpt-4.1",
input=query,
tools=[{
"type": "file_search",
"vector_store_ids": [text_vector_store_id],
"filters": {
"type": "eq",
"key": "month",
"value": "july"
}
}]
)
# Display the results
print("📝 Response:")
print("-" * 40)
print(response.output_text)
query = "Where there any comments about the 'spaghetti'?"
print(f"🔍 Query: {query}\n")
response = client.responses.create(
model="gpt-4.1",
input=query,
tools=[{
"type": "file_search",
"vector_store_ids": [text_image_vector_store_id],
"filters": {
"type": "eq",
"key": "month",
"value": "july"
}
}]
)
print("📝 Response:")
print("-" * 40)
print(response.output_text)
我们可以通过检查检索到的图像来确认这一点是否正确。
IMAGE_ID_MAPPING = {
f"context_{row['id']}_{row['month']}.txt": row["image_path"]
for _, row in df[~df['image_path'].isna()].iterrows()
}
def display_retrieved_images(
response,
cache_dir: str = ".local_cache"
):
"""
Display images from the retrieved search results.
Args:
response: The response object from the search query
cache_dir: Directory where images are stored
Returns:
Dict mapping filenames to image paths for the displayed images
"""
# Get the annotations from the response
try:
annotations = response.output[1].content[0].annotations
retrieved_files = {result.filename for result in annotations}
except (AttributeError, IndexError):
print("No search results found in the response.")
return {}
# Display matching images
displayed_images = {}
for file in retrieved_files:
if file in IMAGE_ID_MAPPING and IMAGE_ID_MAPPING[file]:
image_path = Path(cache_dir) / 'images' / IMAGE_ID_MAPPING[file]
print(f"Displaying image for {file}:")
display(Image(str(image_path)))
displayed_images[file] = str(image_path)
return displayed_images
displayed = display_retrieved_images(response)
print(f"Displayed {len(displayed)} images")
同样，我们可以测试六月份关于烧焦披萨的负面评论。
query = "Were there any negative reviews for pizza, and if so, was the pizza burnt?"
print(f"🔍 Query: {query}\n")
response = client.responses.create(
model="gpt-4.1",
input=query,
tools=[{
"type": "file_search",
"vector_store_ids": [text_image_vector_store_id],
"filters": {
"type": "eq",
"key": "month",
"value": "june"
}
}]
)
print("📝 Response:")
print("-" * 40)
print(response.output_text)
我们可以通过检查检索到的图像来确认这一点是否正确。
displayed = display_retrieved_images(response)
print(f"Displayed {len(displayed)} images")
评估和分析
随着我们的数据集可能随时间推移而演变，并且我们希望评估新模型，我们可以使用 OpenAI Evaluation API 来评估我们系统在情感分析方面的性能。在这个简单的示例中，使用 string_check 标准，我们检查输出是否是三个可能值之一：正面、负面或不确定。
def prepare_evaluation_data(df, text_col="full_sentiment", label_col="label"):
"""Prepare data items for evaluation from DataFrame."""
return [{"item": {"input": str(row[text_col]), "ground_truth": row[label_col]}}
for _, row in df.iterrows()]
def prepare_evaluation_data(
df: pd.DataFrame,
text_col: str = "full_sentiment",
label_col: str = "label"
) -> list:
"""
Prepare evaluation data items from a DataFrame.
Args:
df: Input pandas DataFrame.
text_col: Column containing the input text.
label_col: Column containing the ground truth label.
Returns:
List of dicts formatted for evaluation.
"""
return [
{"item": {"input": str(row[text_col]), "ground_truth": row[label_col]}}
for _, row in df.iterrows()
]
def create_eval_run(evaluation_data: list, eval_id: str) -> str:
"""
Create and launch an evaluation run.
Args:
evaluation_data: List of evaluation items.
eval_id: The evaluation object ID.
Returns:
The run ID as a string.
"""
eval_config = {
"type": "completions",
"model": "gpt-4.1",
"input_messages": {
"type": "template",
"template": [
{
"type": "message",
"role": "user",
"content": {
"type": "input_text",
"text": (
"Classify the sentiment of this food delivery review: {{ item.input }}. "
"Categorize the request into one of \"positive\", \"negative\" or \"unclear\". "
"Respond with only one of those words."
)
}
}
]
},
"source": {
"type": "file_content",
"content": evaluation_data
}
}
run = client.evals.runs.create(
eval_id=eval_id,
data_source=eval_config
)
print("✅ Evaluation run created successfully")
print(f"Run ID: {run.id}")
return run.id
eval_obj = client.evals.create(
name="food-categorization-eval",
data_source_config={
"type": "custom",
"item_schema": {
"type": "object",
"properties": {
"input": {"type": "string"},
"ground_truth": {"type": "string"}
},
"required": ["input", "ground_truth"]
},
"include_sample_schema": True
},
testing_criteria=[
{
"type": "string_check",
"name": "Match output to human label",
"input": "{{sample.output_text}}",
"reference": "{{item.ground_truth}}",
"operation": "eq"
}
]
)
eval_id = eval_obj.id
eval_id
# create evaluation runs
evaluation_data = prepare_evaluation_data(df, text_col="text")
text_only_run_id = create_eval_run(evaluation_data, eval_id)
evaluation_data = prepare_evaluation_data(df)
text_image_run_id = create_eval_run(evaluation_data, eval_id)
# retrieve both run urls
text_only_run = client.evals.runs.retrieve(eval_id=eval_id, run_id=text_only_run_id)
print(text_only_run.to_dict()['report_url'])
text_image_run = client.evals.runs.retrieve(eval_id=eval_obj.id, run_id=text_image_run_id)
print(text_image_run.to_dict()['report_url'])
# you may need to wait a few seconds before running this cell for the eval runs to finish up
text_only_run_output_items = client.evals.runs.output_items.list(eval_id=eval_id, run_id=text_only_run_id)
text_image_run_output_items = client.evals.runs.output_items.list(eval_id=eval_id, run_id=text_image_run_id)
我们可以检索这些评估运行的结果并执行一些本地分析。在这种情况下，我们将比较仅文本和文本+图像运行的性能，并评估总令牌数（通过添加图像上下文）如何影响模型的准确性。我们还可以通过分析失败示例的模型输入来进行一些基本的错误分析。
# Calculate passed and total for text_only_run
text_only_data = text_only_run_output_items.to_dict()['data']
text_only_passed = sum(1 for output_item in text_only_data if output_item['results'][0]['passed'])
text_only_total = len(text_only_data)
# Calculate passed and total for text_image_run
text_image_data = text_image_run_output_items.to_dict()['data']
text_image_passed = sum(1 for output_item in text_image_data if output_item['results'][0]['passed'])
text_image_total = len(text_image_data)
# Calculate average total_tokens for each run
def avg_total_tokens(data):
tokens = [item['sample']['usage']['total_tokens'] for item in data if 'usage' in item['sample']]
return sum(tokens) / len(tokens) if tokens else 0
text_only_avg_tokens = avg_total_tokens(text_only_data)
text_image_avg_tokens = avg_total_tokens(text_image_data)
# Plotting
labels = ['Text Only', 'Text + Image']
passed = [text_only_passed, text_image_passed]
avg_tokens = [text_only_avg_tokens, text_image_avg_tokens]
x = np.arange(len(labels))
width = 0.35
fig, ax1 = plt.subplots()
# Bar for passed only
bars1 = ax1.bar(x - width/2, passed, width, label='Passed', color='green')
ax1.set_ylabel('Accuracy')
ax1.set_xticks(x)
ax1.set_xticklabels(labels)
ax1.set_title('Accuracy and Avg Total Tokens')
ax1.legend(loc='upper left')
# Second y-axis for avg total tokens
ax2 = ax1.twinx()
bars2 = ax2.bar(x + width/2, avg_tokens, width, label='Avg Total Tokens', color='blue', alpha=0.5)
ax2.set_ylabel('Avg Total Tokens')
ax2.legend(loc='upper right')
plt.show()
failed_samples = [
{
"Input": sample['sample']['input'],
"Model Output": sample['sample']['output']
}
for sample in text_only_run_output_items.to_dict()['data']
if not sample['results'][0]['passed']
]
pd.set_option('display.max_colwidth', 150) # Adjust as needed
failed_df = pd.DataFrame(failed_samples)
display(failed_df.style.set_properties(**{'text-align': 'left'}))
最后，让我们清理一下我们创建的一些资源。
# delete vector stores
deleted_vector_store = client.vector_stores.delete(
vector_store_id=text_vector_store_id
)
print(deleted_vector_store)
deleted_vector_store = client.vector_stores.delete(
vector_store_id=text_image_vector_store_id
)
print(deleted_vector_store)