multimodal-agent¶
第3章 · 用户记忆和知识库 · 配套项目
chapter3/multimodal-agent
项目说明¶
Multimodal Agent with Multiple Extraction Techniques¶
An educational agent framework that compares different multimodal content extraction techniques across multiple AI providers (Gemini, OpenAI, Doubao).
Features¶
Three Extraction Modes¶
- Native Multimodality: Direct processing using model's built-in multimodal capabilities
- Gemini 2.5 Pro: Documents (PDF), Images, Audio
- GPT-5/GPT-4o: Images (OpenAI multimodal format)
-
Doubao 1.6: Images (OpenAI multimodal format)
-
Extract to Text: Convert multimodal content to text first, then process
- PDF: OCR using Gemini or GPT-5
- Image: Generate descriptions using GPT-5 or Doubao 1.6
-
Audio: Transcription using Whisper API or Gemini
-
Multimodal Analysis Tools: Additive functionality for follow-up questions
- Image analysis tool (GPT-5/Doubao 1.6)
- Audio analysis tool (Gemini 2.5 Pro)
- PDF analysis tool (Gemini 2.5 Pro)
Architecture¶
The agent follows a modular architecture with clear separation of concerns:
MultimodalAgent
├── Configuration (config.py)
│ ├── Model configurations
│ ├── API key management
│ └── Extraction mode settings
├── Agent Core (agent.py)
│ ├── Message handling (OpenAI format)
│ ├── Conversation history
│ ├── Mode-specific processing
│ └── Streaming responses
└── Multimodal Tools
├── Image analysis
├── Audio analysis
└── PDF analysis
Installation¶
-
Clone the repository and navigate to the project directory:
-
Install dependencies:
-
Configure API keys:
-
Load environment variables:
Quick Offline Start (No API Key)¶
Generate a bundled multimodal sample — a chart-bearing report — so you can run Experiment 3-7 end to end. The exact quarterly figures live only in the chart's bars, not in the surrounding text, which is what makes the three-paradigm trade-off measurable.
# Offline: creates test_files/sample_chart.png and test_files/sample_report.pdf
python create_sample.py # or: python demo.py --generate-sample
Then compare the three extraction paradigms on the same file + question (requires a vision API key such as OpenAI or Gemini):
python demo.py \
--file test_files/sample_chart.png \
--query "Which quarter had the highest revenue, and what was the exact value?" \
--model gpt-5.6-luna
All CLIs expose a Chinese --help (python demo.py --help,
python main.py --help, python create_sample.py --help).
Usage¶
Interactive Chat Mode¶
Start an interactive session with the agent:
Available commands in interactive mode:
- /file <path> - Load a multimodal file
- /mode <native|extract_to_text> - Switch extraction mode
- /model <model_name> - Switch model
- /tools <on|off> - Enable/disable multimodal tools
- /history - Show conversation history
- /clear - Clear conversation history
- /quit - Exit
Process Single File¶
Process a file with a specific query:
# Native mode (default)
python main.py --file document.pdf --query "What is the main topic?"
# Extract to text mode
python main.py --mode extract_to_text --file image.jpg --query "Describe this image"
# With multimodal tools enabled
python main.py --tools --mode extract_to_text --file audio.mp3 --query "What's the content?"
Programmatic Usage¶
import asyncio
from agent import MultimodalAgent, MultimodalContent
from config import ExtractionMode
async def example():
# Initialize agent
agent = MultimodalAgent(
model="gemini-3.5-flash",
mode=ExtractionMode.NATIVE,
enable_tools=True
)
# Process a PDF
content = MultimodalContent(
type="pdf",
path="document.pdf"
)
result = await agent.process_multimodal_content(
content,
"Summarize this document"
)
print(result)
# Chat with streaming
async for chunk in agent.chat("Tell me more about the key points", stream=True):
print(chunk, end="", flush=True)
asyncio.run(example())
Comparing Extraction Techniques¶
Demo Script¶
Run the comprehensive comparison demo:
# Compare extraction modes for a specific file (flags form)
python demo.py --file document.pdf --query "What are the key findings?" --model gpt-5.6-luna
# Backward-compatible positional form still works
python demo.py document.pdf "What are the key findings?"
# Save the full transcript, and skip the cross-model pass
python demo.py --file test_files/sample_chart.png \
--query "Which quarter had the highest revenue?" \
--model gpt-5.6-luna --skip-model-comparison --output result.txt
# This will run:
# 1. Native multimodal mode
# 2. Extract to text mode
# 3. Extract to text with tools
# 4. Comparison across different models (unless --skip-model-comparison)
Demo CLI flags:
| Flag | Description |
|---|---|
--file / positional file |
Multimodal file to process (image / PDF / audio) |
--query / positional query |
Question to ask about the file |
--model |
Model for native/extract modes (default: gemini-3.5-flash) |
--skip-model-comparison |
Only run the three-paradigm comparison |
--generate-sample |
Offline: create the bundled chart sample, then exit |
--output, -o |
Also write the full transcript to a file |
Mode Comparison¶
| Mode | Advantages | Disadvantages | Best For |
|---|---|---|---|
| Native | - Preserves full context - Better visual understanding - Direct processing |
- Limited model support - Higher token usage |
Complex documents with mixed content |
| Extract to Text | - Works with all models - Lower token usage - Can cache extractions |
- Loses visual context - Two-step process |
Text-heavy documents, cost optimization |
| With Tools | - Best of both worlds - Follow-up questions - Selective deep analysis |
- More complex setup - Multiple API calls |
Interactive sessions, detailed Q&A |
Supported File Types¶
Documents¶
- PDF files (up to 1000 pages)
- Best with Gemini 2.5 Pro native mode
Images¶
- JPEG, PNG, GIF, BMP, WebP
- Supported by all models
Audio¶
- MP3, WAV, M4A, FLAC, AAC, OGG
- Transcription via Whisper or Gemini
Model Capabilities¶
| Model | Native PDF | Native Image | Native Audio | Extract to Text | Tools Support |
|---|---|---|---|---|---|
| Gemini 2.5 Pro | ✅ | ✅ | ✅ | ✅ | ✅ |
| GPT-5/GPT-4o | ❌ | ✅ | ❌ | ✅ | ✅ |
| Doubao 1.6 | ❌ | ✅ | ❌ | ✅ | ✅ |
API Configuration¶
Required API Keys¶
- Google Gemini: Required for PDF/Audio native processing
- Get key at: https://makersuite.google.com/app/apikey
-
Set:
GOOGLE_API_KEY(orGEMINI_API_KEY— both are read) -
OpenAI: Required for GPT models and Whisper
- Get key at: https://platform.openai.com/api-keys
-
Set:
OPENAI_API_KEY -
Doubao: Required for Doubao model
- Get key at: https://console.volcengine.com/
- Set:
DOUBAO_API_KEY(orARK_API_KEY— both are read)
File Size Limits¶
- PDF: 20MB
- Images: 20MB
- Audio: 25MB
Testing¶
Run the test suite:
Examples¶
Example 1: Analyzing a Research Paper¶
# Native mode for best understanding
agent = MultimodalAgent(
model="gemini-3.5-flash",
mode=ExtractionMode.NATIVE
)
content = MultimodalContent(type="pdf", path="research_paper.pdf")
summary = await agent.process_multimodal_content(
content,
"What are the main contributions and findings?"
)
Example 2: Processing Images with Follow-up¶
# Extract to text with tools for follow-up questions
agent = MultimodalAgent(
model="gpt-5.6-luna",
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=True
)
# Initial processing
await agent.chat("I have an image at photo.jpg", MultimodalContent(type="image", path="photo.jpg"))
# Follow-up using tools
await agent.chat("What objects are in the background?")
await agent.chat("What's the color scheme?")
Example 3: Audio Transcription and Analysis¶
# Using Whisper for transcription
agent = MultimodalAgent(
model="gpt-5.6-luna",
mode=ExtractionMode.EXTRACT_TO_TEXT
)
content = MultimodalContent(type="audio", path="interview.mp3")
transcript = await agent._extract_audio_to_text(content)
print(f"Transcript: {transcript}")
# Analyze the transcript
analysis = await agent._answer_with_context(
transcript,
"What are the key points discussed?"
)
Best Practices¶
- Mode Selection:
- Use Native mode when visual/audio understanding is crucial
- Use Extract to Text for cost optimization and caching
-
Enable tools for interactive sessions with follow-ups
-
Model Selection:
- Gemini 2.5 Pro: Best for PDFs and audio
- GPT-4o/GPT-5: Best for complex image understanding
-
Doubao 1.6: Alternative for image processing
-
Performance Optimization:
- Cache extracted text for repeated queries
- Use streaming for better user experience
-
Process files under size limits
-
Error Handling:
- Always validate file existence and type
- Check API key configuration before processing
- Handle rate limits and API errors gracefully
Troubleshooting¶
Common Issues¶
- API Key Errors:
- Ensure all required API keys are set in .env
-
Check API key validity and quota
-
File Processing Errors:
- Verify file format is supported
- Check file size is within limits
-
Ensure file path is correct
-
Model Compatibility:
- Not all models support all content types natively
- Use Extract to Text mode for unsupported combinations
Architecture Details¶
Message Format¶
The agent uses OpenAI-compatible message format:
{
"role": "user" | "assistant" | "system" | "tool",
"content": "message text" | [{"type": "text", "text": "..."}, ...],
"tool_calls": [...], # Optional
"tool_call_id": "...", # For tool responses
}
Tool Calling Format¶
Tools follow OpenAI function calling specification:
{
"type": "function",
"function": {
"name": "analyze_image",
"description": "...",
"parameters": {
"type": "object",
"properties": {...},
"required": [...]
}
}
}
Streaming Implementation¶
The agent supports streaming responses for better UX: - Gemini: Native streaming API - OpenAI/Doubao: Stream via chat completions API - Tool results are streamed as they complete
Contributing¶
This is an educational project demonstrating multimodal AI capabilities. Contributions should focus on: - Adding new extraction techniques - Improving model comparisons - Enhancing documentation - Adding more test cases
License¶
MIT License - See LICENSE file for details
OpenRouter 通用回退 / Universal OpenRouter fallback¶
This experiment now supports a universal OpenRouter fallback for its chat LLM.
- If the primary provider key (e.g.
MOONSHOT_API_KEY/KIMI_API_KEY/OPENAI_API_KEY/DOUBAO_API_KEY…) is present, behavior is unchanged. - Else if
OPENROUTER_API_KEYis set, the chat LLM is automatically routed through OpenRouter (https://openrouter.ai/api/v1). Model names are mapped automatically:gpt-*/o1-*→openai/…,claude-*→anthropic/claude-opus-4.8, ids already containing/are kept as-is, and other provider-native ids (e.g.kimi-k3,doubao-*) fall back toopenai/gpt-5.6-luna. SetOPENROUTER_MODELto force a specific OpenRouter model id. - Else a clear error lists the accepted keys.
Add OPENROUTER_API_KEY=... to your .env (see env.example) to enable it.
Note: image analysis and text chat route through OpenRouter (vision-capable default
openai/gpt-5.6-luna). Audio transcription (Whisper) and native-PDF extraction still require a direct OpenAI/Gemini key.
源代码¶
agent.py¶
"""
Multimodal Agent with Multiple Extraction Techniques
Supports native multimodality, extract to text, and multimodal tools
"""
import os
import sys
import json
import base64
import httpx
import asyncio
from typing import Dict, Any, List, Optional, Union, Generator, AsyncGenerator
from dataclasses import dataclass, field
from pathlib import Path
import mimetypes
from datetime import datetime
# Google Gemini imports
from google import genai
from google.genai import types
# OpenAI imports
from openai import OpenAI, AsyncOpenAI
from config import Config, ExtractionMode, Provider, ModelConfig, _openrouter_model_id
@dataclass
class Message:
"""Unified message format"""
role: str # "system", "user", "assistant", "tool"
content: Union[str, List[Dict[str, Any]]]
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary format"""
result = {"role": self.role, "content": self.content}
if self.tool_calls:
result["tool_calls"] = self.tool_calls
if self.tool_call_id:
result["tool_call_id"] = self.tool_call_id
if self.name:
result["name"] = self.name
return result
@dataclass
class MultimodalContent:
"""Container for multimodal content"""
type: str # "pdf", "image", "audio"
data: Optional[bytes] = None
path: Optional[str] = None
url: Optional[str] = None
mime_type: Optional[str] = None
extracted_text: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
# 自动补全 MIME 类型：优先按文件名推断，再按声明的模态兜底。
# 否则原生 OpenAI / Doubao 图像请求会拼出 "data:None;base64,..." 导致 400 错误。
if not self.mime_type and self.path:
guessed = mimetypes.guess_type(self.path)[0]
if guessed:
self.mime_type = guessed
if not self.mime_type:
self.mime_type = {
"pdf": "application/pdf",
"image": "image/jpeg",
"audio": "audio/mpeg",
}.get(self.type)
def get_bytes(self) -> bytes:
"""Get content as bytes"""
if self.data:
return self.data
elif self.path:
return Path(self.path).read_bytes()
elif self.url:
response = httpx.get(self.url)
return response.content
else:
raise ValueError("No content source available")
def get_base64(self) -> str:
"""Get content as base64 encoded string"""
return base64.b64encode(self.get_bytes()).decode('utf-8')
class MultimodalTools:
"""Tools for multimodal content analysis"""
def __init__(self, agent: 'MultimodalAgent'):
self.agent = agent
async def analyze_image(self, image_path: str, query: str) -> str:
"""Analyze an image with a specific query"""
content = MultimodalContent(
type="image",
path=image_path,
mime_type=mimetypes.guess_type(image_path)[0] or "image/jpeg"
)
# Use GPT-5 or Doubao for image analysis
if self.agent.config.get_model_config(self.agent.current_model).provider == Provider.DOUBAO:
return await self._analyze_with_doubao(content, query)
else:
return await self._analyze_with_openai(content, query)
async def analyze_audio(self, audio_path: str, query: str) -> str:
"""Analyze audio with a specific query"""
content = MultimodalContent(
type="audio",
path=audio_path,
mime_type=mimetypes.guess_type(audio_path)[0] or "audio/mpeg"
)
# Use Gemini for audio analysis
return await self._analyze_with_gemini_audio(content, query)
async def analyze_pdf(self, pdf_path: str, query: str) -> str:
"""Analyze a PDF document with a specific query"""
content = MultimodalContent(
type="pdf",
path=pdf_path,
mime_type="application/pdf"
)
# Use Gemini for PDF analysis
return await self._analyze_with_gemini_pdf(content, query)
async def _analyze_with_openai(self, content: MultimodalContent, query: str) -> str:
"""Use OpenAI (or OpenRouter fallback) for content analysis"""
cfg = self.agent.config
# 视觉默认 gpt-5.6-luna（视觉可用）；直连 gpt-5.6 需组织实名，故有 OpenRouter key 时优先走 OpenRouter
if cfg.openrouter_api_key:
client = AsyncOpenAI(api_key=cfg.openrouter_api_key, base_url=cfg.openrouter_base_url)
model = _openrouter_model_id("gpt-5.6-luna")
elif cfg.openai_api_key:
client = AsyncOpenAI(api_key=cfg.openai_api_key)
model = "gpt-5.6-luna"
else:
raise RuntimeError("需要 OPENAI_API_KEY 或 OPENROUTER_API_KEY 才能进行视觉分析")
messages = [{
"role": "user",
"content": [
{"type": "text", "text": query},
{
"type": "image_url",
"image_url": {
"url": f"data:{content.mime_type};base64,{content.get_base64()}"
}
}
]
}]
response = await client.chat.completions.create(
model=model,
messages=messages,
temperature=self.agent.config.temperature
)
return response.choices[0].message.content
async def _analyze_with_doubao(self, content: MultimodalContent, query: str) -> str:
"""Use Doubao for content analysis"""
client = AsyncOpenAI(
api_key=self.agent.config.doubao_api_key,
base_url=self.agent.config.models["doubao-1.6"].base_url
)
messages = [{
"role": "user",
"content": [
{"type": "text", "text": query},
{
"type": "image_url",
"image_url": {
"url": f"data:{content.mime_type};base64,{content.get_base64()}"
}
}
]
}]
response = await client.chat.completions.create(
model="Doubao-1.6",
messages=messages,
temperature=self.agent.config.temperature
)
return response.choices[0].message.content
async def _analyze_with_gemini_audio(self, content: MultimodalContent, query: str) -> str:
"""Use Gemini for audio analysis with thinking mode"""
client = genai.Client(api_key=self.agent.config.gemini_api_key)
audio_data = content.get_bytes()
# Always enable thinking mode
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
response = client.models.generate_content(
model='gemini-3.5-flash',
contents=[
query,
types.Part.from_bytes(
data=audio_data,
mime_type=content.mime_type
)
],
config=config
)
# Extract and print thinking content, return only the answer
result = ""
first_thinking = True
first_response = True
if hasattr(response, 'candidates') and response.candidates:
for part in response.candidates[0].content.parts:
if not part.text:
continue
if part.thought:
if first_thinking:
print("\n[Gemini Thinking] ", end="", flush=True)
first_thinking = False
print(part.text, end="", flush=True)
else:
if first_response:
if not first_thinking: # We had thinking output
print() # End the thinking line
print("[Gemini Response]", flush=True)
first_response = False
# Don't print response text, just collect it
result += part.text
else:
result = response.text
print("\n", flush=True) # End the output line
return result
async def _analyze_with_gemini_pdf(self, content: MultimodalContent, query: str) -> str:
"""Use Gemini for PDF analysis with thinking mode"""
client = genai.Client(api_key=self.agent.config.gemini_api_key)
# Get PDF bytes directly - no base64 encoding needed with new SDK
pdf_data = content.get_bytes()
# Always enable thinking mode
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
response = client.models.generate_content(
model='gemini-3.5-flash',
contents=[
types.Part.from_bytes(
data=pdf_data,
mime_type='application/pdf'
),
query
],
config=config
)
# Extract and print thinking content, return only the answer
result = ""
first_thinking = True
first_response = True
if hasattr(response, 'candidates') and response.candidates:
for part in response.candidates[0].content.parts:
if not part.text:
continue
if part.thought:
if first_thinking:
print("\n[Gemini Thinking] ", end="", flush=True)
first_thinking = False
print(part.text, end="", flush=True)
else:
if first_response:
if not first_thinking: # We had thinking output
print() # End the thinking line
print("[Gemini Response]", flush=True)
first_response = False
# Don't print response text, just collect it
result += part.text
else:
result = response.text
print("\n", flush=True) # End the output line
return result
class MultimodalAgent:
"""Main agent class supporting multiple extraction modes"""
def __init__(
self,
model: Optional[str] = None,
mode: Optional[ExtractionMode] = None,
enable_tools: bool = False
):
self.config = Config()
self.current_model = model or self.config.default_model
self.extraction_mode = mode or self.config.default_mode
self.enable_multimodal_tools = enable_tools
# Conversation history
self.conversation_history: List[Message] = []
# Store current content path for reference
self.current_content_path: Optional[str] = None
# Multimodal tools
self.tools = MultimodalTools(self) if enable_tools else None
# Tool definitions for OpenAI-style function calling
self.tool_definitions = []
if enable_tools:
self.tool_definitions = [
{
"type": "function",
"function": {
"name": "analyze_image",
"description": "Analyze an image with a specific query",
"parameters": {
"type": "object",
"properties": {
"image_path": {
"type": "string",
"description": "Path to the image file"
},
"query": {
"type": "string",
"description": "Question or analysis request about the image"
}
},
"required": ["image_path", "query"]
}
}
},
{
"type": "function",
"function": {
"name": "analyze_audio",
"description": "Analyze audio content with a specific query",
"parameters": {
"type": "object",
"properties": {
"audio_path": {
"type": "string",
"description": "Path to the audio file"
},
"query": {
"type": "string",
"description": "Question or analysis request about the audio"
}
},
"required": ["audio_path", "query"]
}
}
},
{
"type": "function",
"function": {
"name": "analyze_pdf",
"description": "Analyze a PDF document with a specific query",
"parameters": {
"type": "object",
"properties": {
"pdf_path": {
"type": "string",
"description": "Path to the PDF file"
},
"query": {
"type": "string",
"description": "Question or analysis request about the PDF"
}
},
"required": ["pdf_path", "query"]
}
}
}
]
def add_message(self, message: Message):
"""Add a message to conversation history"""
self.conversation_history.append(message)
async def process_multimodal_content(
self,
content: MultimodalContent,
query: Optional[str] = None
) -> str:
"""Process multimodal content based on extraction mode"""
if self.extraction_mode == ExtractionMode.NATIVE:
return await self._process_native(content, query)
elif self.extraction_mode == ExtractionMode.EXTRACT_TO_TEXT:
return await self._extract_to_text(content, query)
else:
raise ValueError(f"Unknown extraction mode: {self.extraction_mode}")
async def _process_native(self, content: MultimodalContent, query: Optional[str]) -> str:
"""Process using native multimodal capabilities"""
model_config = self.config.get_model_config(self.current_model)
if not model_config.supports_native_multimodal:
raise ValueError(f"Model {self.current_model} doesn't support native multimodality")
# Universal OpenRouter fallback: the model's own provider key is missing
# but OPENROUTER_API_KEY is present -> route via OpenRouter (OpenAI-compat).
if self.config.use_openrouter(model_config.provider):
return await self._process_native_openrouter(model_config, content, query)
if model_config.provider == Provider.GEMINI:
return await self._process_native_gemini(content, query)
elif model_config.provider == Provider.OPENAI:
return await self._process_native_openai(content, query)
elif model_config.provider == Provider.DOUBAO:
return await self._process_native_doubao(content, query)
else:
raise ValueError(f"Unknown provider: {model_config.provider}")
async def _process_native_openrouter(self, model_config: ModelConfig, content: MultimodalContent, query: Optional[str]) -> str:
"""Process content via OpenRouter's OpenAI-compatible endpoint.
Images are sent as native vision input; other types are extracted to
text first (OpenRouter has no audio-transcription / native-PDF path)."""
client_kwargs, model_name = self.config.openai_client_args(model_config)
client = AsyncOpenAI(**client_kwargs)
message_content = []
if query:
message_content.append({"type": "text", "text": query})
if content.type == "image":
message_content.append({
"type": "image_url",
"image_url": {
"url": f"data:{content.mime_type};base64,{content.get_base64()}"
}
})
else:
extracted = await self._extract_single_content(content)
message_content.append({"type": "text", "text": extracted})
response = await client.chat.completions.create(
model=model_name,
messages=[{"role": "user", "content": message_content}],
temperature=self.config.temperature
)
return response.choices[0].message.content
async def _process_native_gemini(self, content: MultimodalContent, query: Optional[str]) -> str:
"""Process using Gemini's native multimodal API with thinking mode"""
client = genai.Client(api_key=self.config.gemini_api_key)
# Build content parts
contents = []
# Add the multimodal content using types.Part.from_bytes
content_bytes = content.get_bytes()
if content.type == "pdf":
mime_type = "application/pdf"
elif content.type == "image":
mime_type = content.mime_type or "image/jpeg"
elif content.type == "audio":
mime_type = content.mime_type or "audio/mpeg"
else:
mime_type = content.mime_type
contents.append(types.Part.from_bytes(
data=content_bytes,
mime_type=mime_type
))
# Add the query
if query:
contents.append(query)
else:
contents.append(f"Please analyze this {content.type} content.")
# Always enable thinking mode
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
response = client.models.generate_content(
model='gemini-3.5-flash',
contents=contents,
config=config
)
# Extract and print thinking content, return only the answer
result = ""
if hasattr(response, 'candidates') and response.candidates:
for part in response.candidates[0].content.parts:
if hasattr(part, 'thought') and part.thought and part.text:
print(f"\n💭 [Gemini Thinking]: {part.text}\n", flush=True)
elif part.text:
result += part.text
else:
result = response.text
return result
async def _process_native_openai(self, content: MultimodalContent, query: Optional[str]) -> str:
"""Process using OpenAI's native multimodal API"""
client = AsyncOpenAI(api_key=self.config.openai_api_key)
messages = []
message_content = []
if query:
message_content.append({"type": "text", "text": query})
# OpenAI primarily supports images natively
if content.type == "image":
message_content.append({
"type": "image_url",
"image_url": {
"url": f"data:{content.mime_type};base64,{content.get_base64()}"
}
})
else:
# For other types, we'll need to extract to text first
extracted = await self._extract_single_content(content)
message_content.append({"type": "text", "text": extracted})
messages.append({"role": "user", "content": message_content})
response = await client.chat.completions.create(
model=self.current_model,
messages=messages,
temperature=self.config.temperature
)
return response.choices[0].message.content
async def _process_native_doubao(self, content: MultimodalContent, query: Optional[str]) -> str:
"""Process using Doubao's native multimodal API"""
client = AsyncOpenAI(
api_key=self.config.doubao_api_key,
base_url=self.config.models["doubao-1.6"].base_url
)
messages = []
message_content = []
if query:
message_content.append({"type": "text", "text": query})
# Doubao supports images natively
if content.type == "image":
message_content.append({
"type": "image_url",
"image_url": {
"url": f"data:{content.mime_type};base64,{content.get_base64()}"
}
})
else:
# For other types, extract to text first
extracted = await self._extract_single_content(content)
message_content.append({"type": "text", "text": extracted})
messages.append({"role": "user", "content": message_content})
response = await client.chat.completions.create(
model="Doubao-1.6",
messages=messages,
temperature=self.config.temperature
)
return response.choices[0].message.content
async def _extract_to_text(self, content: MultimodalContent, query: Optional[str]) -> str:
"""Extract multimodal content to text first"""
extracted_text = await self._extract_single_content(content)
content.extracted_text = extracted_text
# Now process the query with extracted text
if query:
return await self._answer_with_context(extracted_text, query)
else:
return extracted_text
async def _extract_single_content(self, content: MultimodalContent) -> str:
"""Extract a single piece of content to text"""
if content.type == "pdf":
return await self._extract_pdf_to_text(content)
elif content.type == "image":
return await self._extract_image_to_text(content)
elif content.type == "audio":
return await self._extract_audio_to_text(content)
else:
raise ValueError(f"Unknown content type: {content.type}")
async def _extract_pdf_to_text(self, content: MultimodalContent) -> str:
"""Extract PDF to text using OCR with thinking mode"""
# Use Gemini for PDF extraction with new SDK
client = genai.Client(api_key=self.config.gemini_api_key)
pdf_data = content.get_bytes()
# Always enable thinking mode
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
response = client.models.generate_content(
model='gemini-3.5-flash',
contents=[
types.Part.from_bytes(
data=pdf_data,
mime_type='application/pdf'
),
"Extract all text content from this PDF document, preserving structure and formatting."
],
config=config
)
# Extract and print thinking content, return only the answer
result = ""
first_thinking = True
first_response = True
if hasattr(response, 'candidates') and response.candidates:
for part in response.candidates[0].content.parts:
if not part.text:
continue
if part.thought:
if first_thinking:
print("\n[Gemini Thinking] ", end="", flush=True)
first_thinking = False
print(part.text, end="", flush=True)
elif part.text:
if first_response:
if not first_thinking: # We had thinking output
print() # End the thinking line
print("[Gemini Response] ", end="", flush=True)
first_response = False
print(part.text, end="", flush=True)
result += part.text
else:
result = response.text
print("\n", flush=True) # End the output line
return result
async def _extract_image_to_text(self, content: MultimodalContent) -> str:
"""Extract image to text description"""
# 图像转文本：gpt-5.6-luna（优先 OpenRouter，直连 5.6 需组织实名）/ Doubao / OpenRouter 兜底
if self.config.openrouter_api_key:
client = AsyncOpenAI(
api_key=self.config.openrouter_api_key,
base_url=self.config.openrouter_base_url
)
model = _openrouter_model_id("gpt-5.6-luna")
elif self.config.openai_api_key:
client = AsyncOpenAI(api_key=self.config.openai_api_key)
model = "gpt-5.6-luna"
elif self.config.doubao_api_key:
client = AsyncOpenAI(
api_key=self.config.doubao_api_key,
base_url=self.config.models["doubao-1.6"].base_url
)
model = "Doubao-1.6"
else:
raise RuntimeError("需要 OPENAI_API_KEY / OPENROUTER_API_KEY / DOUBAO_API_KEY 才能进行图像转文本")
messages = [{
"role": "user",
"content": [
{
"type": "text",
"text": "Describe this image in detail, including all text, objects, and contextual information."
},
{
"type": "image_url",
"image_url": {
"url": f"data:{content.mime_type};base64,{content.get_base64()}"
}
}
]
}]
response = await client.chat.completions.create(
model=model,
messages=messages,
temperature=0.3
)
return response.choices[0].message.content
async def _extract_audio_to_text(self, content: MultimodalContent) -> str:
"""Extract audio to text transcript"""
# Option 1: Use Whisper API
if self.config.openai_api_key:
client = AsyncOpenAI(api_key=self.config.openai_api_key)
# Save audio temporarily for Whisper
import tempfile
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp:
tmp.write(content.get_bytes())
tmp_path = tmp.name
try:
with open(tmp_path, "rb") as audio_file:
transcript = await client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
)
return transcript.text
finally:
os.unlink(tmp_path)
else:
# Option 2: Use Gemini for audio understanding
client = genai.Client(api_key=self.config.gemini_api_key)
audio_data = content.get_bytes()
# Always enable thinking mode
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
response = client.models.generate_content(
model='gemini-3.5-flash',
contents=[
"Transcribe this audio content completely and accurately.",
types.Part.from_bytes(
data=audio_data,
mime_type=content.mime_type or "audio/mpeg"
)
],
config=config
)
# Extract and print thinking content, return only the answer
result = ""
if hasattr(response, 'candidates') and response.candidates:
for part in response.candidates[0].content.parts:
if hasattr(part, 'thought') and part.thought and part.text:
print(f"\n💭 [Gemini Thinking]: {part.text}\n", flush=True)
elif part.text:
result += part.text
else:
result = response.text
return result
async def _answer_with_context(self, context: str, query: str) -> str:
"""Answer a query given extracted text context"""
model_config = self.config.get_model_config(self.current_model)
prompt = f"Context:\n{context}\n\nQuestion: {query}\n\nAnswer:"
# Universal OpenRouter fallback (primary provider key absent).
if self.config.use_openrouter(model_config.provider):
client_kwargs, model_name = self.config.openai_client_args(model_config)
client = AsyncOpenAI(**client_kwargs)
response = await client.chat.completions.create(
model=model_name,
messages=[{"role": "user", "content": prompt}],
temperature=self.config.temperature
)
return response.choices[0].message.content
if model_config.provider == Provider.GEMINI:
client = genai.Client(api_key=self.config.gemini_api_key)
# Always enable thinking mode
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
response = client.models.generate_content(
model=model_config.model_name,
contents=[prompt],
config=config
)
# Extract and print thinking content, return only the answer
result = ""
if hasattr(response, 'candidates') and response.candidates:
for part in response.candidates[0].content.parts:
if hasattr(part, 'thought') and part.thought and part.text:
print(f"\n💭 [Gemini Thinking]: {part.text}\n", flush=True)
elif part.text:
result += part.text
else:
result = response.text
return result
else:
# Use OpenAI-compatible API
if model_config.provider == Provider.OPENAI:
client = AsyncOpenAI(api_key=self.config.openai_api_key)
else: # Doubao
client = AsyncOpenAI(
api_key=self.config.doubao_api_key,
base_url=model_config.base_url
)
response = await client.chat.completions.create(
model=model_config.model_name,
messages=[{"role": "user", "content": prompt}],
temperature=self.config.temperature
)
return response.choices[0].message.content
async def chat(
self,
message: str,
multimodal_content: Optional[MultimodalContent] = None,
stream: bool = True
) -> AsyncGenerator[str, None]:
"""Main chat interface with streaming support"""
# Add user message to history
self.add_message(Message(role="user", content=message))
# Process new multimodal content if provided (for native mode)
# In extract mode, content should already be in conversation history via load_and_extract_content
if multimodal_content and self.extraction_mode == ExtractionMode.EXTRACT_TO_TEXT:
# If new content is provided inline during chat, extract and add to the current message
extracted = await self._extract_single_content(multimodal_content)
# Update the last message with extracted context
enhanced_message = f"[Context from {multimodal_content.type}]:\n{extracted}\n\n{message}"
self.conversation_history[-1].content = enhanced_message
# Get response based on model
model_config = self.config.get_model_config(self.current_model)
if stream:
async for chunk in self._stream_response(model_config, multimodal_content):
yield chunk
else:
response = await self._get_response(model_config)
yield response
async def _stream_response(self, model_config: ModelConfig, multimodal_content: Optional[MultimodalContent] = None) -> AsyncGenerator[str, None]:
"""Stream response from the model"""
# Universal OpenRouter fallback -> use the OpenAI-compatible stream path.
if self.config.use_openrouter(model_config.provider):
async for chunk in self._stream_openai_response(model_config):
yield chunk
elif model_config.provider == Provider.GEMINI:
async for chunk in self._stream_gemini_response(multimodal_content):
yield chunk
else:
async for chunk in self._stream_openai_response(model_config):
yield chunk
async def _stream_gemini_response(self, multimodal_content: Optional[MultimodalContent] = None) -> AsyncGenerator[str, None]:
"""Stream response from Gemini with thinking mode for debugging"""
client = genai.Client(api_key=self.config.gemini_api_key)
# Build full conversation history for Gemini
# Format: alternating user/assistant messages as a single string
conversation_parts = []
for msg in self.conversation_history:
if msg.role == "user":
conversation_parts.append(f"User: {msg.content}")
elif msg.role == "assistant":
conversation_parts.append(f"Assistant: {msg.content}")
# System messages can be included as context
elif msg.role == "system":
conversation_parts.append(f"System: {msg.content}")
# Join all conversation parts
full_conversation = "\n\n".join(conversation_parts)
if not full_conversation:
return
# Build content list
contents = []
# Add multimodal content if present and in native mode
if multimodal_content and self.extraction_mode == ExtractionMode.NATIVE:
# Add the multimodal data first
content_bytes = multimodal_content.get_bytes()
if multimodal_content.type == "pdf":
mime_type = "application/pdf"
elif multimodal_content.type == "image":
mime_type = multimodal_content.mime_type or "image/jpeg"
elif multimodal_content.type == "audio":
mime_type = multimodal_content.mime_type or "audio/mpeg"
else:
mime_type = multimodal_content.mime_type
contents.append(types.Part.from_bytes(
data=content_bytes,
mime_type=mime_type
))
# Add the full conversation as context
contents.append(full_conversation)
# Always enable thinking mode for transparency and debugging
config = types.GenerateContentConfig(
thinking_config=types.ThinkingConfig(
include_thoughts=True
)
)
# Stream the response - note: generate_content_stream returns a regular generator
response = client.models.generate_content_stream(
model=self.config.get_model_config(self.current_model).model_name,
contents=contents,
config=config
)
full_response = ""
first_thinking = True
first_response = True
# Use regular for loop since the SDK returns a regular generator
for chunk in response:
# Handle thinking mode output
if hasattr(chunk, 'candidates') and chunk.candidates:
for part in chunk.candidates[0].content.parts:
if not part.text:
continue
if part.thought:
# Print thinking header once, then content without newlines
if first_thinking:
print("\n[Gemini Thinking] ", end="", flush=True)
first_thinking = False
print(part.text, end="", flush=True)
else:
# Print response header once for debugging
if first_response:
if not first_thinking: # We had thinking output
print() # End the thinking line
print("[Gemini Response]", flush=True)
first_response = False
# Yield regular response text to the user (don't print, it will be printed in main.py)
yield part.text
full_response += part.text
elif chunk.text:
# Fallback for standard streaming
if first_response:
if not first_thinking: # We had thinking output
print() # End the thinking line
print("[Gemini Response]", flush=True)
first_response = False
# Yield text without printing (will be printed in main.py)
yield chunk.text
full_response += chunk.text
# End the console output line
print("\n", flush=True)
# Add assistant response to history (excluding thinking parts)
self.add_message(Message(role="assistant", content=full_response))
async def _stream_openai_response(self, model_config: ModelConfig) -> AsyncGenerator[str, None]:
"""Stream response from OpenAI-compatible API (direct or via OpenRouter)"""
client_kwargs, model_name = self.config.openai_client_args(model_config)
client = AsyncOpenAI(**client_kwargs)
# Convert conversation history to OpenAI format
messages = [msg.to_dict() for msg in self.conversation_history]
# Add tools if enabled
kwargs = {
"model": model_name,
"messages": messages,
"temperature": self.config.temperature,
"stream": True
}
if self.enable_multimodal_tools and self.tool_definitions:
kwargs["tools"] = self.tool_definitions
kwargs["tool_choice"] = "auto"
response = await client.chat.completions.create(**kwargs)
full_response = ""
tool_calls = []
async for chunk in response:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
yield content
full_response += content
# Handle tool calls
if chunk.choices[0].delta.tool_calls:
for tool_call in chunk.choices[0].delta.tool_calls:
# Accumulate tool call information
if tool_call.index >= len(tool_calls):
tool_calls.append({
"id": tool_call.id,
"type": "function",
"function": {"name": "", "arguments": ""}
})
if tool_call.function.name:
tool_calls[tool_call.index]["function"]["name"] = tool_call.function.name
if tool_call.function.arguments:
tool_calls[tool_call.index]["function"]["arguments"] += tool_call.function.arguments
# Process tool calls if any
if tool_calls:
# Add assistant message with tool calls
self.add_message(Message(
role="assistant",
content=full_response or "",
tool_calls=tool_calls
))
# Execute tools
for tool_call in tool_calls:
tool_result = await self._execute_tool(tool_call)
# Add tool result to history
self.add_message(Message(
role="tool",
content=tool_result,
tool_call_id=tool_call["id"],
name=tool_call["function"]["name"]
))
# Stream tool result
yield f"\n[Tool: {tool_call['function']['name']}]\n{tool_result}\n"
# Get final response after tool execution
async for chunk in self._stream_openai_response(model_config):
yield chunk
else:
# Add assistant response to history
self.add_message(Message(role="assistant", content=full_response))
async def _execute_tool(self, tool_call: Dict[str, Any]) -> str:
"""Execute a tool call"""
function_name = tool_call["function"]["name"]
try:
arguments = json.loads(tool_call["function"]["arguments"])
except json.JSONDecodeError:
return f"Error: invalid JSON arguments for tool '{function_name}'"
if function_name == "analyze_image":
image_path = arguments.get("image_path")
query = arguments.get("query")
if not image_path or not query:
return "Error: analyze_image requires 'image_path' and 'query' arguments"
return await self.tools.analyze_image(image_path, query)
elif function_name == "analyze_audio":
audio_path = arguments.get("audio_path")
query = arguments.get("query")
if not audio_path or not query:
return "Error: analyze_audio requires 'audio_path' and 'query' arguments"
return await self.tools.analyze_audio(audio_path, query)
elif function_name == "analyze_pdf":
pdf_path = arguments.get("pdf_path")
query = arguments.get("query")
if not pdf_path or not query:
return "Error: analyze_pdf requires 'pdf_path' and 'query' arguments"
return await self.tools.analyze_pdf(pdf_path, query)
else:
return f"Unknown tool: {function_name}"
async def _get_response(self, model_config: ModelConfig) -> str:
"""Get non-streaming response"""
full_response = ""
async for chunk in self._stream_response(model_config):
full_response += chunk
return full_response
def reset_conversation(self):
"""Clear conversation history and current content"""
self.conversation_history = []
self.current_content_path = None
async def load_and_extract_content(self, content: MultimodalContent) -> str:
"""Load and extract content if in extract mode"""
# Store the current content path
self.current_content_path = content.path
if self.extraction_mode == ExtractionMode.EXTRACT_TO_TEXT:
# Extract content immediately
print("Extracting content to text...", flush=True)
extracted_text = await self._extract_single_content(content)
# Add the extracted content directly to conversation history as a user message
# This provides context for all subsequent questions
context_msg = f"[Document: {content.path}]\n\n{extracted_text}"
self.add_message(Message(role="user", content=context_msg))
return f"Extracted {content.type} content and added to conversation context. Ready for questions."
else:
# In native mode, just note that content is loaded
return f"Loaded {content.type}: {content.path}"
def get_conversation_history(self) -> List[Dict[str, Any]]:
"""Get conversation history in OpenAI format"""
return [msg.to_dict() for msg in self.conversation_history]
config.py¶
"""
Configuration for Multimodal Agent
Supports multiple providers and extraction modes
"""
import os
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
def _openrouter_model_id(model) -> str:
"""Map a provider-native model name to an OpenRouter model id, used by the
universal OpenRouter fallback. An explicit OPENROUTER_MODEL env var wins.
Vision-capable default (gpt-5.6-luna) so image analysis still works."""
override = os.getenv("OPENROUTER_MODEL")
if override:
return override
m = (model or "").strip()
if not m:
return "openai/gpt-5.6-luna"
if "/" in m:
return m
ml = m.lower()
if ml.startswith(("gpt-", "o1", "o3", "o4", "chatgpt")):
return "openai/" + m
if ml.startswith("claude-"):
return "anthropic/claude-opus-4.8"
if ml.startswith("gemini"):
return "google/" + m # e.g. gemini-3.5-flash -> google/gemini-3.5-flash
# Provider-native ids (doubao-*/qwen/...) -> a widely-available vision model.
return "openai/gpt-5.6-luna"
class ExtractionMode(Enum):
"""Modes for multimodal content extraction"""
NATIVE = "native" # Use model's native multimodal capabilities
EXTRACT_TO_TEXT = "extract_to_text" # Convert multimodal to text first
class Provider(Enum):
"""Supported model providers"""
GEMINI = "gemini"
OPENAI = "openai"
DOUBAO = "doubao"
@dataclass
class ModelConfig:
"""Configuration for a specific model"""
provider: Provider
model_name: str
api_key: str
base_url: Optional[str] = None
supports_native_multimodal: bool = True
class Config:
"""Main configuration class for multimodal agent"""
def __init__(self):
# Load API keys from environment.
# 兼容常见别名：Gemini 官方 SDK 用 GEMINI_API_KEY，旧文档用 GOOGLE_API_KEY，两者都接受；
# 豆包/方舟(Ark)的 Key 环境变量常见为 DOUBAO_API_KEY 或 ARK_API_KEY。
self.gemini_api_key = os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY", "")
self.openai_api_key = os.getenv("OPENAI_API_KEY", "")
self.doubao_api_key = os.getenv("DOUBAO_API_KEY") or os.getenv("ARK_API_KEY", "")
# Universal OpenRouter fallback: when a model's own provider key is
# missing but OPENROUTER_API_KEY is present, route that model through
# OpenRouter's OpenAI-compatible endpoint.
self.openrouter_api_key = os.getenv("OPENROUTER_API_KEY", "")
self.openrouter_base_url = "https://openrouter.ai/api/v1"
# Model configurations
self.models = {
"gemini-3.5-flash": ModelConfig(
provider=Provider.GEMINI,
model_name="gemini-3.5-flash",
api_key=self.gemini_api_key,
supports_native_multimodal=True
),
"gpt-5": ModelConfig(
provider=Provider.OPENAI,
model_name="gpt-5",
api_key=self.openai_api_key,
supports_native_multimodal=True
),
"gpt-5.6-luna": ModelConfig(
provider=Provider.OPENAI,
model_name="gpt-5.6-luna",
api_key=self.openai_api_key,
supports_native_multimodal=True
),
"doubao-1.6": ModelConfig(
provider=Provider.DOUBAO,
model_name="Doubao-1.6",
api_key=self.doubao_api_key,
base_url="https://ark.cn-beijing.volces.com/api/v3",
supports_native_multimodal=True
)
}
# Default settings
self.default_model = "gemini-3.5-flash"
self.default_mode = ExtractionMode.NATIVE
self.enable_multimodal_tools = False
# File size limits (in MB)
self.max_pdf_size_mb = 20
self.max_image_size_mb = 20
self.max_audio_size_mb = 25
# Whisper settings for audio transcription
self.whisper_model = "whisper-1"
# Temperature settings
self.temperature = 0.7
self.max_tokens = 4096
def get_model_config(self, model_name: str) -> ModelConfig:
"""Get configuration for a specific model"""
if model_name not in self.models:
raise ValueError(f"Unknown model: {model_name}")
return self.models[model_name]
def validate_api_keys(self) -> Dict[str, bool]:
"""Check which API keys are configured"""
return {
"gemini": bool(self.gemini_api_key),
"openai": bool(self.openai_api_key),
"doubao": bool(self.doubao_api_key),
"openrouter": bool(self.openrouter_api_key)
}
def has_provider_key(self, provider: 'Provider') -> bool:
"""Whether the direct API key for a provider is configured."""
if provider == Provider.OPENAI:
return bool(self.openai_api_key)
if provider == Provider.DOUBAO:
return bool(self.doubao_api_key)
if provider == Provider.GEMINI:
return bool(self.gemini_api_key)
return False
def use_openrouter(self, provider: 'Provider') -> bool:
"""True when a model's own provider key is missing but OpenRouter is
available -> the call should be routed through OpenRouter."""
return (not self.has_provider_key(provider)) and bool(self.openrouter_api_key)
def openai_client_args(self, model_config: 'ModelConfig'):
"""Return (client_kwargs, model_name) for an OpenAI-compatible call,
applying the universal OpenRouter fallback when needed."""
provider = model_config.provider
_m = (model_config.model_name or "").lower()
_prefer_or = bool(self.openrouter_api_key) and _m.startswith("gpt-5") # 直连 gpt-5.6 需组织实名，优先 OpenRouter
if _prefer_or or self.use_openrouter(provider):
return (
{"api_key": self.openrouter_api_key, "base_url": self.openrouter_base_url},
_openrouter_model_id(model_config.model_name),
)
if provider == Provider.DOUBAO:
return {"api_key": self.doubao_api_key, "base_url": model_config.base_url}, model_config.model_name
# OPENAI (and any GEMINI forced through the OpenAI-compatible path)
return {"api_key": self.openai_api_key}, model_config.model_name
create_sample.py¶
"""
离线样例生成器 (Offline sample generator)
生成一个"含图表的报告"作为多模态样例，用于实验 3-7 对比三种提取范式。
产物同时包含：
- test_files/sample_chart.png 仅图表（图像模态）
- test_files/sample_report.pdf 图表 + 文字说明（文档模态，书中的"含图表的 PDF 报告"）
关键设计：图表里的精确数值（如各季度营收）只出现在柱状图上，正文并未逐一写出。
这样在实验中：
- 原生多模态模式可以直接"看懂"柱子读出数值；
- 提取为文本模式若用通用描述器转写图像，往往丢失精确数值与空间关系；
从而让三种范式的取舍可被直接测量，而不是靠猜。
本脚本完全离线，不需要任何 API Key。
"""
import argparse
from pathlib import Path
import matplotlib
matplotlib.use("Agg") # 无界面后端，纯离线出图
import matplotlib.pyplot as plt
# 图表数据：只在柱状图上标注，正文不重复这些精确数字
QUARTERS = ["Q1", "Q2", "Q3", "Q4"]
REVENUE = [120, 150, 95, 180] # 单位：百万美元 ($M)
def create_chart(output_path: Path) -> Path:
"""用 matplotlib 生成一张柱状图（图像模态样例）。"""
output_path.parent.mkdir(parents=True, exist_ok=True)
fig, ax = plt.subplots(figsize=(6, 4), dpi=150)
bars = ax.bar(QUARTERS, REVENUE, color=["#4C72B0", "#55A868", "#C44E52", "#8172B3"])
# 把精确数值标注在柱子顶端——这些信息只存在于图像里
for bar, value in zip(bars, REVENUE):
ax.text(
bar.get_x() + bar.get_width() / 2,
bar.get_height() + 3,
f"${value}M",
ha="center",
va="bottom",
fontsize=11,
fontweight="bold",
)
ax.set_title("Acme Corp Quarterly Revenue 2024", fontsize=13, fontweight="bold")
ax.set_ylabel("Revenue (in $M)")
ax.set_ylim(0, 210)
ax.grid(axis="y", linestyle="--", alpha=0.4)
fig.tight_layout()
fig.savefig(output_path)
plt.close(fig)
return output_path
def create_report_pdf(chart_path: Path, output_path: Path) -> Path:
"""把图表和一段文字说明组合成一份 PDF 报告（文档模态样例）。"""
try:
from reportlab.lib.pagesizes import A4
from reportlab.lib.units import cm
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import (
SimpleDocTemplate,
Paragraph,
Spacer,
Image as RLImage,
)
except ImportError:
print("提示：未安装 reportlab，跳过 PDF 生成（pip install reportlab）。")
return None
output_path.parent.mkdir(parents=True, exist_ok=True)
styles = getSampleStyleSheet()
# 正文刻意只给出定性描述，不逐一写出各季度精确数值——数值只在图里
body_text = (
"This internal report summarizes Acme Corp's revenue performance in 2024. "
"Overall the year showed healthy growth, with a mid-year dip followed by a "
"strong recovery in the final quarter. The chart below breaks down revenue "
"by quarter; management attributes the fourth-quarter surge to the launch of "
"the new enterprise product line."
)
doc = SimpleDocTemplate(str(output_path), pagesize=A4)
story = [
Paragraph("Acme Corp 2024 Revenue Report", styles["Title"]),
Spacer(1, 0.4 * cm),
Paragraph(body_text, styles["BodyText"]),
Spacer(1, 0.6 * cm),
RLImage(str(chart_path), width=14 * cm, height=9.3 * cm),
]
doc.build(story)
return output_path
def main():
parser = argparse.ArgumentParser(
description="离线生成含图表的多模态样例（图像 + PDF 报告），供实验 3-7 使用。无需 API Key。"
)
parser.add_argument(
"--output-dir",
default="test_files",
help="样例输出目录（默认：test_files）",
)
parser.add_argument(
"--no-pdf",
action="store_true",
help="只生成 PNG 图表，不生成 PDF 报告",
)
args = parser.parse_args()
out_dir = Path(args.output_dir)
chart_path = create_chart(out_dir / "sample_chart.png")
print(f"已生成图表: {chart_path}")
if not args.no_pdf:
pdf_path = create_report_pdf(chart_path, out_dir / "sample_report.pdf")
if pdf_path:
print(f"已生成报告: {pdf_path}")
print(
"\n提示：图表上的精确季度营收只存在于图像中，正文并未逐一写出。\n"
"可用如下问题对比三种范式（原生 / 提取为文本 / 带工具）：\n"
' "Which quarter had the highest revenue, and what was the exact value?"'
)
if __name__ == "__main__":
main()
demo.py¶
"""
Demo script showcasing different extraction techniques
"""
import argparse
import asyncio
import sys
from pathlib import Path
from typing import List, Optional, Tuple
from agent import MultimodalAgent, MultimodalContent
from config import ExtractionMode
class _Tee:
"""Duplicate stdout writes to a file so --output can save the transcript."""
def __init__(self, stream, file_handle):
self._stream = stream
self._file = file_handle
def write(self, data):
self._stream.write(data)
self._file.write(data)
def flush(self):
self._stream.flush()
self._file.flush()
async def compare_extraction_modes(file_path: str, query: str, model: str = "gemini-3.5-flash"):
"""Compare different extraction modes for the same content"""
print(f"\n{'='*80}")
print(f"COMPARING EXTRACTION MODES")
print(f"File: {file_path}")
print(f"Query: {query}")
print(f"{'='*80}\n")
# Determine content type
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == '.pdf':
content_type = "pdf"
elif suffix in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']:
content_type = "image"
elif suffix in ['.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg']:
content_type = "audio"
else:
print(f"Unsupported file type: {suffix}")
return
# Test with native mode (Gemini)
print("\n" + "-"*60)
print(f"1. NATIVE MULTIMODAL MODE ({model})")
print("-"*60)
agent_native = MultimodalAgent(
model=model,
mode=ExtractionMode.NATIVE,
enable_tools=False
)
content = MultimodalContent(type=content_type, path=file_path)
try:
result = await agent_native.process_multimodal_content(content, query)
print(result)
except Exception as e:
print(f"Error: {e}")
# Test with extract-to-text mode
print("\n" + "-"*60)
print("2. EXTRACT TO TEXT MODE")
print("-"*60)
agent_extract = MultimodalAgent(
model=model,
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=False
)
try:
# First extract the content
print("Extracting content to text...")
extracted = await agent_extract._extract_single_content(content)
print("\nExtracted text:")
print(extracted)
# Then answer the query
print(f"\nAnswering query with extracted text...")
result = await agent_extract._answer_with_context(extracted, query)
print(result)
except Exception as e:
print(f"Error: {e}")
# Test with extract-to-text + tools mode
print("\n" + "-"*60)
print("3. EXTRACT TO TEXT + MULTIMODAL TOOLS")
print("-"*60)
agent_tools = MultimodalAgent(
model=model,
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=True
)
try:
print("Using extract-to-text with tools enabled for follow-up questions...")
# Initial processing
extracted = await agent_tools._extract_single_content(content)
print(f"Extracted {len(extracted)} characters")
# Simulate a conversation with follow-up
async for chunk in agent_tools.chat(query, content, stream=True):
print(chunk, end="", flush=True)
print()
# Follow-up question that might use tools
if content_type == "image":
follow_up = f"What colors are dominant in the image at {file_path}?"
elif content_type == "pdf":
follow_up = f"What specific data or figures are mentioned in the PDF at {file_path}?"
else: # audio
follow_up = f"What is the tone or mood of the audio at {file_path}?"
print(f"\nFollow-up question: {follow_up}")
async for chunk in agent_tools.chat(follow_up, None, stream=True):
print(chunk, end="", flush=True)
print()
except Exception as e:
print(f"Error: {e}")
async def compare_models(file_path: str, query: str):
"""Compare different models for the same task"""
print(f"\n{'='*80}")
print(f"COMPARING MODELS")
print(f"File: {file_path}")
print(f"Query: {query}")
print(f"{'='*80}\n")
# Determine content type
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == '.pdf':
content_type = "pdf"
elif suffix in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']:
content_type = "image"
elif suffix in ['.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg']:
content_type = "audio"
else:
print(f"Unsupported file type: {suffix}")
return
content = MultimodalContent(type=content_type, path=file_path)
# Test with different models
models = ["gemini-3.5-flash", "gpt-5.6-luna", "doubao-1.6"]
for model in models:
print("\n" + "-"*60)
print(f"Model: {model}")
print("-"*60)
try:
# Skip if API key not configured
from config import Config
config = Config()
if model == "gemini-3.5-flash" and not config.gemini_api_key:
print("Skipping: Gemini API key not configured")
continue
elif model in ["gpt-5.6-luna", "gpt-5"] and not (config.openai_api_key or config.openrouter_api_key):
print("Skipping: OpenAI API key not configured")
continue
elif model == "doubao-1.6" and not config.doubao_api_key:
print("Skipping: Doubao API key not configured")
continue
agent = MultimodalAgent(
model=model,
mode=ExtractionMode.NATIVE if content_type != "audio" or model == "gemini-3.5-flash" else ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=False
)
result = await agent.process_multimodal_content(content, query)
print(result)
except Exception as e:
print(f"Error: {e}")
async def demo_conversation_with_tools():
"""Demonstrate a conversation with multimodal tools"""
print(f"\n{'='*80}")
print("DEMO: CONVERSATION WITH MULTIMODAL TOOLS")
print(f"{'='*80}\n")
agent = MultimodalAgent(
model="gemini-3.5-flash",
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=True
)
# Simulate a conversation
conversations = [
("I need help analyzing some documents. I have PDFs, images, and audio files.", None),
("Can you analyze the image at test_files/sample.jpg and tell me what you see?", None),
("Now analyze the PDF at test_files/document.pdf and summarize its main points.", None),
("What's in the audio file at test_files/recording.mp3?", None),
("Based on all these files, what's the common theme?", None)
]
for message, content in conversations:
print(f"\nUser: {message}")
print("Assistant: ", end="", flush=True)
try:
async for chunk in agent.chat(message, content, stream=True):
print(chunk, end="", flush=True)
print()
except Exception as e:
print(f"\nError: {e}")
print("(File might not exist - this is a demo)")
def build_parser() -> argparse.ArgumentParser:
"""构建实验 3-7 的命令行接口。"""
parser = argparse.ArgumentParser(
description=(
"实验 3-7：多模态信息提取的三种技术范式对比（原生多模态 / 提取为文本 / 带工具）。\n"
"将同一多模态文件和同一问题分别交给三种模式处理，观察表现差异。"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例：\n"
" # 先离线生成含图表的样例（无需 API Key）\n"
" python demo.py --generate-sample\n"
" # 用生成的图表跑三种范式对比（需要 API Key）\n"
" python demo.py --file test_files/sample_chart.png \\\n"
' --query \"Which quarter had the highest revenue, and what was the exact value?\"\n'
" # 兼容旧写法（位置参数）\n"
" python demo.py document.pdf \"总结这份文档的要点\""
),
)
parser.add_argument(
"file", nargs="?", default=None,
help="要处理的多模态文件（图像 / PDF 文档 / 音频）。也可用 --file 指定",
)
parser.add_argument(
"query", nargs="?", default=None,
help="向该文件提出的问题。也可用 --query 指定",
)
parser.add_argument(
"--file", dest="file_opt", default=None,
help="要处理的多模态文件（等价于位置参数 file）",
)
parser.add_argument(
"--query", dest="query_opt", default=None,
help="向该文件提出的问题（等价于位置参数 query）",
)
parser.add_argument(
"--model", default="gemini-3.5-flash",
help="原生 / 提取模式使用的模型（默认：gemini-3.5-flash）",
)
parser.add_argument(
"--skip-model-comparison", action="store_true",
help="只跑三种范式对比，跳过跨模型对比",
)
parser.add_argument(
"--generate-sample", action="store_true",
help="离线生成含图表的样例文件到 test_files/ 后退出（无需 API Key）",
)
parser.add_argument(
"--output", "-o", default=None,
help="将完整对比结果同时写入指定文件（如 result.txt）",
)
return parser
async def run_comparison(file_path: str, query: str, model: str, skip_model_comparison: bool):
"""运行三种范式对比，可选跨模型对比。"""
print("="*80)
print("MULTIMODAL AGENT DEMO")
print("="*80)
await compare_extraction_modes(file_path, query, model=model)
if not skip_model_comparison:
await compare_models(file_path, query)
async def main():
"""实验入口：解析参数并运行对比。"""
parser = build_parser()
args = parser.parse_args()
# 离线样例生成：不需要 API Key，直接产出图表 + PDF 报告
if args.generate_sample:
import create_sample
sys.argv = ["create_sample.py"] # 用默认输出目录 test_files/
create_sample.main()
return
file_path = args.file_opt or args.file
query = args.query_opt or args.query
# 缺少文件或问题时，回退到无需真实文件的对话演示
if not file_path or not query:
print("="*80)
print("MULTIMODAL AGENT DEMO")
print("="*80)
print("\n未提供 <file> 与 <query>，改为运行对话演示。")
print("用法：python demo.py --file <文件> --query <问题>")
print("先生成样例：python demo.py --generate-sample\n")
await demo_conversation_with_tools()
return
# 支持 --output：把整段对比结果同时落盘
if args.output:
with open(args.output, "w", encoding="utf-8") as fh:
original_stdout = sys.stdout
sys.stdout = _Tee(original_stdout, fh)
try:
await run_comparison(file_path, query, args.model, args.skip_model_comparison)
finally:
sys.stdout = original_stdout
print(f"\n完整对比结果已写入：{args.output}")
else:
await run_comparison(file_path, query, args.model, args.skip_model_comparison)
if __name__ == "__main__":
asyncio.run(main())
main.py¶
"""
Main entry point for Multimodal Agent
Demonstrates different extraction modes and model capabilities
"""
import asyncio
import sys
import argparse
from pathlib import Path
from typing import Optional
from agent import MultimodalAgent, MultimodalContent
from config import ExtractionMode, Config
class _Tee:
"""将 stdout 同时写入终端与文件，用于 --output。"""
def __init__(self, stream, file_handle):
self._stream = stream
self._file = file_handle
def write(self, data):
self._stream.write(data)
self._file.write(data)
def flush(self):
self._stream.flush()
self._file.flush()
async def process_file(
agent: MultimodalAgent,
file_path: str,
query: Optional[str] = None
) -> None:
"""Process a single file with the agent"""
path = Path(file_path)
if not path.exists():
print(f"Error: File '{file_path}' not found")
return
# Determine content type
suffix = path.suffix.lower()
if suffix == '.pdf':
content_type = "pdf"
elif suffix in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']:
content_type = "image"
elif suffix in ['.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg']:
content_type = "audio"
else:
print(f"Error: Unsupported file type '{suffix}'")
return
# Create multimodal content
content = MultimodalContent(
type=content_type,
path=file_path
)
print(f"\n{'='*60}")
print(f"Processing {content_type.upper()}: {path.name}")
print(f"Mode: {agent.extraction_mode.value}")
print(f"Model: {agent.current_model}")
print(f"Multimodal Tools: {'Enabled' if agent.enable_multimodal_tools else 'Disabled'}")
print(f"{'='*60}\n")
try:
# Process content
if agent.extraction_mode == ExtractionMode.NATIVE:
# Use native multimodal processing
result = await agent.process_multimodal_content(content, query)
print("Native Processing Result:")
print("-" * 40)
print(result)
else:
# Extract to text mode
print("Extracting content to text...")
extracted = await agent._extract_single_content(content)
print("Extracted Text:")
print("-" * 40)
print(extracted[:1000] + "..." if len(extracted) > 1000 else extracted)
if query:
print(f"\nAnswering query: {query}")
print("-" * 40)
answer = await agent._answer_with_context(extracted, query)
print(answer)
except Exception as e:
print(f"Error processing file: {e}")
async def interactive_chat(agent: MultimodalAgent) -> None:
"""Interactive chat session with the agent"""
print("\n" + "="*60)
print("Interactive Multimodal Chat")
print(f"Model: {agent.current_model}")
print(f"Mode: {agent.extraction_mode.value}")
print(f"Multimodal Tools: {'Enabled' if agent.enable_multimodal_tools else 'Disabled'}")
print("="*60)
print("\nCommands:")
print(" /file <path> - Load a multimodal file")
print(" /mode <native|extract_to_text> - Switch extraction mode")
print(" /model <model_name> - Switch model")
print(" /tools <on|off> - Enable/disable multimodal tools")
print(" /history - Show conversation history")
print(" /clear - Clear conversation history")
print(" /quit - Exit")
print("\n")
current_content = None
while True:
try:
user_input = input("You: ").strip()
if not user_input:
continue
# Handle commands
if user_input.startswith("/"):
parts = user_input.split(maxsplit=1)
command = parts[0].lower()
args = parts[1] if len(parts) > 1 else ""
if command == "/quit":
print("Goodbye!")
break
elif command == "/file":
if not args:
print("Usage: /file <path>")
continue
path = Path(args)
if not path.exists():
print(f"File not found: {args}")
continue
# Determine content type
suffix = path.suffix.lower()
if suffix == '.pdf':
content_type = "pdf"
elif suffix in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']:
content_type = "image"
elif suffix in ['.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg']:
content_type = "audio"
else:
print(f"Unsupported file type: {suffix}")
continue
current_content = MultimodalContent(
type=content_type,
path=args
)
# Extract content immediately if in extract mode
result = await agent.load_and_extract_content(current_content)
print(result)
# In extract mode, content is already extracted, no need to keep it
if agent.extraction_mode == ExtractionMode.EXTRACT_TO_TEXT:
current_content = None
elif command == "/mode":
if args == "native":
agent.extraction_mode = ExtractionMode.NATIVE
print("Switched to native multimodal mode")
elif args == "extract_to_text":
agent.extraction_mode = ExtractionMode.EXTRACT_TO_TEXT
print("Switched to extract-to-text mode")
else:
print("Usage: /mode <native|extract_to_text>")
elif command == "/model":
if args in agent.config.models:
agent.current_model = args
print(f"Switched to model: {args}")
else:
print(f"Available models: {', '.join(agent.config.models.keys())}")
elif command == "/tools":
if args == "on":
agent.enable_multimodal_tools = True
agent.tools = MultimodalTools(agent) if not agent.tools else agent.tools
print("Multimodal tools enabled")
elif args == "off":
agent.enable_multimodal_tools = False
agent.tools = None
print("Multimodal tools disabled")
else:
print("Usage: /tools <on|off>")
elif command == "/history":
history = agent.get_conversation_history()
print("\nConversation History:")
print("-" * 40)
for msg in history:
role = msg["role"].upper()
content = msg["content"]
if isinstance(content, str):
preview = content[:200] + "..." if len(content) > 200 else content
print(f"{role}: {preview}")
print("-" * 40)
elif command == "/clear":
agent.reset_conversation()
current_content = None
print("Conversation history cleared")
else:
print(f"Unknown command: {command}")
continue
# Regular chat message
print("\nAssistant: ", end="", flush=True)
try:
async for chunk in agent.chat(user_input, current_content, stream=True):
print(chunk, end="", flush=True)
print("\n")
# Clear current content after first use
current_content = None
except Exception as e:
print(f"\nError: {e}")
except KeyboardInterrupt:
print("\n\nInterrupted. Type /quit to exit.")
continue
except Exception as e:
print(f"Error: {e}")
continue
async def main():
"""Main entry point"""
parser = argparse.ArgumentParser(
description="多模态 Agent：对比原生多模态、提取为文本、带工具三种信息提取范式。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例：\n"
" # 处理图像并提问\n"
" python main.py --file test_files/sample_chart.png --query \"图中哪个季度营收最高？\"\n"
" # 处理 PDF 文档（提取为文本模式）\n"
" python main.py --mode extract_to_text --file report.pdf --query \"总结要点\"\n"
" # 进入交互式对话\n"
" python main.py --interactive"
),
)
parser.add_argument("--mode", choices=["native", "extract_to_text"], default="native",
help="提取模式：native（原生多模态）或 extract_to_text（提取为文本），默认 native")
parser.add_argument("--model", default="gemini-3.5-flash",
help="使用的模型（默认：gemini-3.5-flash）")
parser.add_argument("--tools", action="store_true",
help="启用多模态分析工具（analyze_image / analyze_audio / analyze_pdf）")
parser.add_argument("--file", help="要处理的单个文件（图像 / PDF 文档 / 音频）")
parser.add_argument("--query", help="向该文件提出的问题")
parser.add_argument("--output", "-o", help="将处理结果同时写入指定文件")
parser.add_argument("--interactive", action="store_true",
help="进入交互式对话会话")
args = parser.parse_args()
# Validate API keys
config = Config()
api_keys = config.validate_api_keys()
print("API Key Status:")
for provider, has_key in api_keys.items():
status = "✓ Configured" if has_key else "✗ Not configured"
print(f" {provider.capitalize()}: {status}")
# Create agent
mode = ExtractionMode.NATIVE if args.mode == "native" else ExtractionMode.EXTRACT_TO_TEXT
agent = MultimodalAgent(
model=args.model,
mode=mode,
enable_tools=args.tools
)
# Process based on arguments
if args.file:
if args.output:
# 将结果同时写入文件
with open(args.output, "w", encoding="utf-8") as fh:
original_stdout = sys.stdout
sys.stdout = _Tee(original_stdout, fh)
try:
await process_file(agent, args.file, args.query)
finally:
sys.stdout = original_stdout
print(f"\n处理结果已写入：{args.output}")
else:
await process_file(agent, args.file, args.query)
elif args.interactive:
await interactive_chat(agent)
else:
# Default to interactive mode
await interactive_chat(agent)
if __name__ == "__main__":
asyncio.run(main())
quickstart.py¶
"""
Quickstart script for testing multimodal agent
Creates sample files and demonstrates capabilities
"""
import asyncio
import base64
from pathlib import Path
import os
from agent import MultimodalAgent, MultimodalContent
from config import ExtractionMode, Config
def create_sample_files():
"""Create sample files for testing"""
# Create test_files directory
test_dir = Path("test_files")
test_dir.mkdir(exist_ok=True)
# Create a simple text-based "image" (SVG)
svg_content = """<?xml version="1.0" encoding="UTF-8"?>
<svg width="200" height="200" xmlns="http://www.w3.org/2000/svg">
<rect x="10" y="10" width="180" height="180" fill="lightblue" stroke="black" stroke-width="2"/>
<circle cx="100" cy="100" r="50" fill="yellow" stroke="orange" stroke-width="3"/>
<text x="100" y="105" text-anchor="middle" font-size="20" fill="black">Hello AI!</text>
</svg>
"""
svg_path = test_dir / "sample.svg"
svg_path.write_text(svg_content)
print(f"Created: {svg_path}")
# Create a simple text file that we'll treat as a "document"
doc_content = """
# Sample Document for Multimodal Agent Testing
## Introduction
This is a test document created for demonstrating the multimodal agent's capabilities.
The agent can process this document in different modes:
1. **Native Mode**: Direct processing using the model's built-in capabilities
2. **Extract to Text**: Convert to text first, then analyze
3. **With Tools**: Use specialized tools for detailed analysis
## Key Features
- Support for multiple file formats (PDF, images, audio)
- Multiple AI model providers (Gemini, OpenAI, Doubao)
- Streaming responses for better user experience
- Tool calling for advanced analysis
## Technical Details
The system uses a unified message format compatible with OpenAI's API structure,
making it easy to switch between different providers while maintaining consistency.
## Conclusion
This multimodal agent demonstrates state-of-the-art AI capabilities for
content understanding and analysis across different modalities.
"""
doc_path = test_dir / "sample_document.txt"
doc_path.write_text(doc_content)
print(f"Created: {doc_path}")
return test_dir
async def test_basic_functionality():
"""Test basic agent functionality"""
print("\n" + "="*60)
print("QUICKSTART: Testing Multimodal Agent")
print("="*60)
# Check API keys
config = Config()
api_keys = config.validate_api_keys()
print("\n1. API Key Status:")
print("-" * 40)
for provider, has_key in api_keys.items():
status = "✅ Configured" if has_key else "❌ Not configured"
print(f" {provider.capitalize()}: {status}")
if not any(api_keys.values()):
print("\n⚠️ Warning: No API keys configured!")
print("Please copy env.example to .env and add your API keys.")
return
# Create sample files
print("\n2. Creating Sample Files:")
print("-" * 40)
test_dir = create_sample_files()
# Test with available model
if api_keys["gemini"]:
model = "gemini-3.5-flash"
print(f"\n3. Testing with {model}:")
print("-" * 40)
agent = MultimodalAgent(
model=model,
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=False
)
# Process the text document
doc_path = test_dir / "sample_document.txt"
content = MultimodalContent(
type="text",
path=str(doc_path),
data=doc_path.read_bytes()
)
print("Processing sample document...")
try:
# Simulate as if it's a PDF for demonstration
content.type = "pdf"
result = await agent._extract_pdf_to_text(content)
print("Extracted content preview:")
print(result[:300] + "..." if len(result) > 300 else result)
# Answer a question
print("\nAsking a question about the document...")
answer = await agent._answer_with_context(
result,
"What are the three modes mentioned in the document?"
)
print("Answer:", answer)
except Exception as e:
print(f"Error: {e}")
elif api_keys["openai"]:
model = "gpt-5.6-luna"
print(f"\n3. Testing with {model}:")
print("-" * 40)
agent = MultimodalAgent(
model=model,
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=False
)
print("Note: OpenAI models work best with images.")
print("For document processing, Gemini is recommended.")
else:
print("\n3. Skipping tests - no API keys configured")
async def test_conversation_mode():
"""Test conversation mode with streaming"""
config = Config()
if not config.gemini_api_key and not config.openai_api_key:
print("\nSkipping conversation test - no API keys configured")
return
print("\n" + "="*60)
print("4. Testing Conversation Mode")
print("="*60)
# Use available model
if config.gemini_api_key:
model = "gemini-3.5-flash"
else:
model = "gpt-5.6-luna"
agent = MultimodalAgent(
model=model,
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=True
)
print(f"Using model: {model}")
print("Tools: Enabled")
print("\nStarting conversation...")
print("-" * 40)
# Simulate a conversation
messages = [
"Hello! I'm testing the multimodal agent. Can you explain what you can do?",
"What types of files can you process?",
"How do the different extraction modes work?"
]
for message in messages:
print(f"\nUser: {message}")
print("Assistant: ", end="", flush=True)
try:
response_text = ""
async for chunk in agent.chat(message, stream=True):
print(chunk, end="", flush=True)
response_text += chunk
print()
# Small delay for readability
await asyncio.sleep(0.5)
except Exception as e:
print(f"\nError: {e}")
break
async def main():
"""Run all quickstart tests"""
print("🚀 Multimodal Agent Quickstart")
print("=" * 60)
# Run basic tests
await test_basic_functionality()
# Run conversation test
await test_conversation_mode()
print("\n" + "="*60)
print("✅ Quickstart Complete!")
print("="*60)
print("\nNext steps:")
print("1. Add your API keys to .env file")
print("2. Try with your own files: python main.py --file <path> --query <question>")
print("3. Start interactive mode: python main.py --interactive")
print("4. Run comparisons: python demo.py <file> <query>")
if __name__ == "__main__":
asyncio.run(main())
test_execute_tool_robust.py¶
"""Regression tests: _execute_tool must return an error string to the model
instead of raising KeyError/JSONDecodeError on malformed LLM tool-call
arguments (missing required fields or truncated streamed JSON)."""
import asyncio
import json
import os
import sys
import types
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from agent import MultimodalAgent
def _make_agent():
"""Build a MultimodalAgent without __init__ (which needs API keys)."""
agent = MultimodalAgent.__new__(MultimodalAgent)
calls = []
async def fake_analyze(path, query):
calls.append((path, query))
return "analysis ok"
agent.tools = types.SimpleNamespace(
analyze_image=fake_analyze,
analyze_audio=fake_analyze,
analyze_pdf=fake_analyze,
)
return agent, calls
def _run(agent, name, arguments):
tool_call = {"id": "call_1", "function": {"name": name, "arguments": arguments}}
return asyncio.run(agent._execute_tool(tool_call))
def test_missing_query_returns_error_not_keyerror():
agent, calls = _make_agent()
result = _run(agent, "analyze_image", json.dumps({"image_path": "cat.png"}))
assert result.startswith("Error:")
assert calls == []
def test_missing_path_returns_error_not_keyerror():
agent, calls = _make_agent()
result = _run(agent, "analyze_pdf", json.dumps({"query": "what is this?"}))
assert result.startswith("Error:")
assert calls == []
def test_malformed_json_returns_error_not_exception():
agent, calls = _make_agent()
result = _run(agent, "analyze_audio", '{"audio_path": "a.mp3", "que')
assert result.startswith("Error:")
assert calls == []
def test_valid_arguments_still_call_tool():
agent, calls = _make_agent()
result = _run(agent, "analyze_image", json.dumps({"image_path": "cat.png", "query": "describe"}))
assert result == "analysis ok"
assert calls == [("cat.png", "describe")]
def test_unknown_tool_still_reported():
agent, _ = _make_agent()
result = _run(agent, "analyze_video", json.dumps({}))
assert "Unknown tool" in result
test_multimodal.py¶
"""
Test script for multimodal agent functionality
"""
import asyncio
import unittest
from unittest.mock import Mock, patch, AsyncMock
from pathlib import Path
from agent import MultimodalAgent, MultimodalContent, MultimodalTools, Message
from config import ExtractionMode, Provider
class TestMultimodalContent(unittest.TestCase):
"""Test MultimodalContent class"""
def test_content_creation(self):
"""Test creating multimodal content"""
content = MultimodalContent(
type="image",
path="test.jpg",
mime_type="image/jpeg"
)
self.assertEqual(content.type, "image")
self.assertEqual(content.path, "test.jpg")
self.assertEqual(content.mime_type, "image/jpeg")
def test_get_base64(self):
"""Test base64 encoding"""
content = MultimodalContent(
type="text",
data=b"Hello World"
)
base64_str = content.get_base64()
self.assertEqual(base64_str, "SGVsbG8gV29ybGQ=")
class TestMessage(unittest.TestCase):
"""Test Message class"""
def test_message_creation(self):
"""Test creating messages"""
msg = Message(
role="user",
content="Hello"
)
self.assertEqual(msg.role, "user")
self.assertEqual(msg.content, "Hello")
def test_message_to_dict(self):
"""Test converting message to dictionary"""
msg = Message(
role="assistant",
content="Hi there",
tool_calls=[{"id": "1", "function": {"name": "test"}}]
)
msg_dict = msg.to_dict()
self.assertEqual(msg_dict["role"], "assistant")
self.assertEqual(msg_dict["content"], "Hi there")
self.assertIn("tool_calls", msg_dict)
class TestMultimodalAgent(unittest.IsolatedAsyncioTestCase):
"""Test MultimodalAgent class"""
async def test_agent_initialization(self):
"""Test agent initialization"""
agent = MultimodalAgent(
model="gemini-3.5-flash",
mode=ExtractionMode.NATIVE,
enable_tools=False
)
self.assertEqual(agent.current_model, "gemini-3.5-flash")
self.assertEqual(agent.extraction_mode, ExtractionMode.NATIVE)
self.assertFalse(agent.enable_multimodal_tools)
self.assertIsNone(agent.tools)
async def test_agent_with_tools(self):
"""Test agent initialization with tools"""
agent = MultimodalAgent(
model="gemini-3.5-flash",
mode=ExtractionMode.EXTRACT_TO_TEXT,
enable_tools=True
)
self.assertTrue(agent.enable_multimodal_tools)
self.assertIsNotNone(agent.tools)
self.assertEqual(len(agent.tool_definitions), 3)
async def test_conversation_history(self):
"""Test conversation history management"""
agent = MultimodalAgent()
# Add messages
agent.add_message(Message(role="user", content="Hello"))
agent.add_message(Message(role="assistant", content="Hi"))
history = agent.get_conversation_history()
self.assertEqual(len(history), 2)
self.assertEqual(history[0]["role"], "user")
self.assertEqual(history[1]["role"], "assistant")
# Reset conversation
agent.reset_conversation()
history = agent.get_conversation_history()
self.assertEqual(len(history), 0)
@patch('agent.genai.configure')
@patch('agent.genai.GenerativeModel')
async def test_extract_pdf_to_text(self, mock_model_class, mock_configure):
"""Test PDF extraction to text"""
agent = MultimodalAgent(mode=ExtractionMode.EXTRACT_TO_TEXT)
# Mock Gemini response
mock_model = Mock()
mock_response = Mock()
mock_response.text = "Extracted PDF text"
mock_model.generate_content.return_value = mock_response
mock_model_class.return_value = mock_model
content = MultimodalContent(
type="pdf",
data=b"PDF content"
)
result = await agent._extract_pdf_to_text(content)
self.assertEqual(result, "Extracted PDF text")
@patch('agent.AsyncOpenAI')
async def test_extract_image_to_text(self, mock_openai_class):
"""Test image extraction to text"""
agent = MultimodalAgent(mode=ExtractionMode.EXTRACT_TO_TEXT)
# Mock OpenAI response
mock_client = AsyncMock()
mock_response = AsyncMock()
mock_choice = Mock()
mock_message = Mock()
mock_message.content = "Image description"
mock_choice.message = mock_message
mock_response.choices = [mock_choice]
mock_client.chat.completions.create.return_value = mock_response
mock_openai_class.return_value = mock_client
content = MultimodalContent(
type="image",
data=b"Image data",
mime_type="image/jpeg"
)
result = await agent._extract_image_to_text(content)
self.assertEqual(result, "Image description")
@patch('agent.genai.configure')
@patch('agent.genai.GenerativeModel')
async def test_process_native_gemini(self, mock_model_class, mock_configure):
"""Test native Gemini processing"""
agent = MultimodalAgent(
model="gemini-3.5-flash",
mode=ExtractionMode.NATIVE
)
# Mock Gemini response
mock_model = Mock()
mock_response = Mock()
mock_response.text = "Gemini analysis result"
mock_model.generate_content.return_value = mock_response
mock_model_class.return_value = mock_model
content = MultimodalContent(
type="pdf",
data=b"PDF content"
)
result = await agent._process_native_gemini(content, "Analyze this")
self.assertEqual(result, "Gemini analysis result")
class TestMultimodalTools(unittest.IsolatedAsyncioTestCase):
"""Test MultimodalTools class"""
async def test_tools_initialization(self):
"""Test tools initialization"""
agent = MultimodalAgent(enable_tools=True)
tools = MultimodalTools(agent)
self.assertEqual(tools.agent, agent)
@patch('agent.AsyncOpenAI')
async def test_analyze_image_tool(self, mock_openai_class):
"""Test image analysis tool"""
agent = MultimodalAgent(enable_tools=True)
# Mock OpenAI response
mock_client = AsyncMock()
mock_response = AsyncMock()
mock_choice = Mock()
mock_message = Mock()
mock_message.content = "Image analysis"
mock_choice.message = mock_message
mock_response.choices = [mock_choice]
mock_client.chat.completions.create.return_value = mock_response
mock_openai_class.return_value = mock_client
# Create temporary test image
import tempfile
with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as tmp:
tmp.write(b"test image data")
tmp_path = tmp.name
try:
result = await agent.tools.analyze_image(tmp_path, "What's in this image?")
self.assertEqual(result, "Image analysis")
finally:
Path(tmp_path).unlink()
def run_tests():
"""Run all tests"""
unittest.main(argv=[''], exit=False, verbosity=2)
if __name__ == "__main__":
run_tests()