欢迎！本笔记本演示了如何利用OpenAI的视觉与响应API构建一个检索增强生成(RAG)系统。该系统专注于多模态数据，结合图像与文本输入来分析客户体验。系统采用GPT-4.1，将图像理解与文件搜索功能相集成，从而提供具备上下文感知能力的响应。
多模态数据集正变得越来越普遍，特别是在医疗保健等领域，这些记录通常同时包含视觉数据（如放射扫描）和伴随文本（如临床笔记）。现实世界的数据集也往往存在噪声，信息不完整或缺失，这使得同时分析多种模态变得至关重要。
本指南聚焦于一个客户服务用例：评估可能包含照片和书面评价的客户反馈。您将学习如何合成生成图像和文本输入，使用文件搜索进行上下文检索，并应用Evals API来评估整合图像理解如何影响整体性能。
%pip install openai evals pandas numpy matplotlib tqdm ipython --upgrade --quietimport 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.pngdef 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")借助文件搜索功能，我们可以通过自然语言查询来分析数据集。对于纯文本数据集，我们发现缺少可能影响分析的信息。
7月份唯一一条关于意大利面的正面评价附有视觉反馈，我们可以看到仅基于文本上下文的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标准，我们检查输出是否为三个可能值之一：positive、negative或unclear。
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.ideval_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)我们可以检索这些评估运行的结果并进行一些本地分析。在本例中，我们将比较纯文本和文本+图像运行的性能，并评估增加总token数（通过添加图像上下文）如何影响模型的准确性。我们还可以通过分析失败示例的模型输入来进行一些基本的错误分析。
# 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)