Sanaz
06/28/2025, 4:43 PMMarvin
06/28/2025, 4:43 PMMarvin
06/28/2025, 4:44 PMpython
from prefect import flow, task
from marvin import ai_fn, ai_model
from marvin.tools.search import Search
from marvin.tools.file_reader import FileReader
import json
# Configure specialized agents for different tasks
@ai_model
class WebResearchAgent:
"""Agent specialized in web research and information synthesis."""
llm_model = "gpt-4"
temperature = 0.3
@ai_fn
def search_and_synthesize(self, query: str) -> str:
"""Search the web and synthesize findings."""
pass
@ai_model
class FileProcessingAgent:
"""Agent specialized in processing and analyzing different file types."""
llm_model = "gpt-4"
temperature = 0.2
@ai_fn
def analyze_file(self, file_content: str, file_type: str) -> dict:
"""Analyze file content and extract key information."""
pass
@task
def perform_web_research(query: str) -> str:
"""Task to perform web research using the WebResearchAgent."""
agent = WebResearchAgent()
search_tool = Search()
# Perform search and get results
search_results = search_tool.search(query)
# Have the agent synthesize the results
synthesis = agent.search_and_synthesize(search_results)
return synthesis
@task
def process_file(file_path: str) -> dict:
"""Task to process different file types using the FileProcessingAgent."""
agent = FileProcessingAgent()
file_reader = FileReader()
# Read the file content
file_content = file_reader.read(file_path)
file_type = file_path.split('.')[-1]
# Have the agent analyze the content
analysis = agent.analyze_file(file_content, file_type)
return analysis
@flow
def research_and_analyze_flow(
search_query: str,
file_paths: list[str]
) -> dict:
"""
Main workflow that coordinates web research and file processing.
Args:
search_query: The query to research online
file_paths: List of paths to files that need to be processed
"""
# Perform web research
web_results = perform_web_research(search_query)
# Process all files
file_results = []
for file_path in file_paths:
result = process_file(file_path)
file_results.append(result)
# Combine results
combined_results = {
"web_research": web_results,
"file_analysis": file_results
}
return combined_results
# Example usage
if __name__ == "__main__":
query = "Latest developments in AI and data engineering"
files_to_process = [
"data/report.pdf",
"data/analysis.xlsx",
"data/notes.txt"
]
results = research_and_analyze_flow(query, files_to_process)
print(json.dumps(results, indent=2))
This example demonstrates several key concepts:
1. Specialized Agents: We create two distinct AI agents:
- `WebResearchAgent`: Focused on web research and synthesis
- `FileProcessingAgent`: Specialized in processing different file types
2. Task Separation: Each major operation is separated into its own task:
- `perform_web_research`: Handles web searching and result synthesis
- `process_file`: Manages file reading and analysis
3. Flow Orchestration: The main flow research_and_analyze_flow
coordinates the entire process, managing both web research and file processing tasks.
4. Tool Integration: The workflow uses Marvin's built-in tools:
- `Search`: For web searching capabilities
- `FileReader`: For handling different file types
To use this workflow, you'll need to:
1. Install the required dependencies:
```bash
pipSanaz
06/30/2025, 12:44 PMMarvin
06/30/2025, 12:44 PMMarvin
06/30/2025, 12:45 PMbash
# Core requirements
pip install "prefect>=3.0.0"
pip install "marvin>=1.0.0"
pip install "openai>=1.0.0" # Required for GPT model access
# For web search capabilities
pip install "duckduckgo-search>=3.0.0" # For web searching
pip install "beautifulsoup4>=4.9.0" # For parsing web content
# For file processing
pip install "python-magic>=0.4.27" # For file type detection
pip install "pypdf>=3.0.0" # For PDF processing
pip install "openpyxl>=3.0.0" # For Excel files
pip install "pandas>=1.0.0" # For structured data processing
Additionally, you'll need to:
1. Set up your OpenAI API key:
bash
export OPENAI_API_KEY="your-api-key-here"
Or in your Python code:
python
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
2. Optional but recommended - create a virtual environment:
bash
# Create a virtual environment
python -m venv venv
# Activate it
# On Windows:
venv\Scripts\activate
# On Unix or MacOS:
source venv/bin/activate
Important Notes:
1. Version Compatibility:
- The versions specified are minimum versions. You can use newer versions, but be aware of potential breaking changes.
- If you encounter version conflicts, you might need to adjust the versions to find a compatible combination.
2. File Type Support:
- The dependencies listed above cover common file types (PDF, Excel, text files)
- If you need to process additional file types, you might need extra libraries:
- docx2txt
for Word documents
- pillow
for image processing
- python-pptx
for PowerPoint files
3. System Dependencies:
- python-magic
might require additional system libraries on Windows
- On Linux, you might need to install `libmagic`:
bash
# Ubuntu/Debian
sudo apt-get install libmagic1
# CentOS/RHEL
sudo yum install file-devel
4. Memory Requirements:
- When processing large files or running multiple agents simultaneously, ensure your system has adequate RAM
- GPT-4 models typically require more memory than GPT-3.5
To verify your installation, you can run this simple test:
python
import prefect
import marvin
import openai
import duckduckgo_search
import magic
import pypdf
import pandas
print("All core dependencies successfully imported!")
If you plan to deploy this workflow in a production environment, it's recommended to:
1. Create a requirements.txt
file:
bash
pip freeze > requirements.txt
2. Or use a pyproject.toml
or setup.py
for more structured dependency management.
Would you like me to provide more specific information about any of these dependencies or help with troubleshooting the installation process?Sanaz
06/30/2025, 12:57 PMMarvin
06/30/2025, 12:57 PMMarvin
06/30/2025, 12:58 PMpython
from marvin import ai_fn, ai_model
from prefect import flow, task
@ai_model
class ResearchAgent:
llm_model = "gpt-4"
@ai_fn
def research_topic(self, topic: str) -> str:
"""Research a topic and return key findings."""
pass
@ai_model
class OutlineAgent:
llm_model = "gpt-4"
@ai_fn
def create_outline(self, research: str) -> list[str]:
"""Create an article outline based on research."""
pass
@ai_model
class WriterAgent:
llm_model = "gpt-4"
@ai_fn
def write_section(self, outline: str, research: str) -> str:
"""Write a section of the article."""
pass
@flow
def article_creation_flow(topic: str):
# Research phase
researcher = ResearchAgent()
research = researcher.research_topic(topic)
# Outline phase
outliner = OutlineAgent()
outline = outliner.create_outline(research)
# Writing phase
writer = WriterAgent()
sections = []
for section in outline:
content = writer.write_section(section, research)
sections.append(content)
return "\n\n".join(sections)
2. Advanced Web Search Example
Here's a more sophisticated web research workflow:
python
from marvin import ai_fn, ai_model
from prefect import flow, task
from duckduckgo_search import ddg
from bs4 import BeautifulSoup
import requests
import asyncio
@ai_model
class WebSearchAgent:
llm_model = "gpt-4"
temperature = 0.3
@ai_fn
def generate_search_queries(self, topic: str) -> list[str]:
"""Generate multiple search queries to cover different aspects of the topic."""
pass
@ai_fn
def analyze_webpage(self, content: str, url: str) -> dict:
"""Analyze webpage content and extract relevant information."""
pass
@ai_fn
def synthesize_findings(self, results: list[dict]) -> str:
"""Synthesize findings from multiple sources into a coherent summary."""
pass
@task
async def fetch_webpage(url: str) -> str:
"""Fetch and parse webpage content."""
headers = {'User-Agent': 'Mozilla/5.0'}
response = requests.get(url, headers=headers)
soup = BeautifulSoup(response.text, 'html.parser')
return soup.get_text()
@flow
async def advanced_research_flow(topic: str, max_sources: int = 5):
agent = WebSearchAgent()
# Generate multiple search queries
queries = agent.generate_search_queries(topic)
# Perform searches
all_results = []
for query in queries:
search_results = ddg(query, max_results=max_sources)
# Process each search result
for result in search_results:
content = await fetch_webpage(result['link'])
analysis = agent.analyze_webpage(content, result['link'])
all_results.append(analysis)
# Synthesize findings
final_summary = agent.synthesize_findings(all_results)
return final_summary
3. Multi-Format Document Processing Example
```python
from marvin import ai_fn, ai_model
from prefect import flow, task
import pypdf
import pandas as pd
from typing import Union, Dict
@ai_model
class DocumentAnalysisAgent:
llm_model = "gpt-4"
@ai_fn
def analyze_Sanaz
06/30/2025, 1:03 PMMarvin
06/30/2025, 1:03 PMMarvin
06/30/2025, 1:04 PM@ai_model
decorator in Marvin is a class decorator that transforms a regular Python class into an AI-powered class with the following key features:
1. Model Configuration
python
@ai_model
class ResearchAgent:
# Configure which LLM to use
llm_model = "gpt-4" # or "gpt-3.5-turbo", etc.
# Optional: Configure model parameters
temperature = 0.7 # Controls randomness (0.0 - 1.0)
max_tokens = 1000 # Maximum response length
2. State Management: The decorator allows the class to maintain state and context across multiple AI function calls.
3. AI Function Integration: When combined with @ai_fn
, it enables methods to be executed using the configured LLM:
python
@ai_model
class ContentAnalyzer:
llm_model = "gpt-4"
@ai_fn
def analyze_text(self, text: str) -> dict:
"""
The docstring becomes the instruction for the AI.
The method signature defines input/output types.
"""
pass # The actual implementation is handled by the LLM
4. Inheritance Support: AI models can inherit from other AI models:
python
@ai_model
class BaseAnalyzer:
llm_model = "gpt-4"
@ai_fn
def basic_analysis(self, text: str) -> str:
"""Perform basic text analysis."""
pass
@ai_model
class SpecializedAnalyzer(BaseAnalyzer):
@ai_fn
def detailed_analysis(self, text: str) -> dict:
"""Perform detailed domain-specific analysis."""
pass
5. Memory and Context:
python
@ai_model
class ConversationalAgent:
llm_model = "gpt-4"
def __init__(self):
self.conversation_history = []
@ai_fn
def respond(self, user_input: str) -> str:
"""Respond based on the conversation history."""
self.conversation_history.append(user_input)
return # AI generates response
Key Benefits:
1. Encapsulation: Groups related AI functions and their configuration in a single class
2. Reusability: AI models can be instantiated multiple times with different configurations
3. Type Safety: Enforces input/output types for AI functions
4. State Management: Maintains context across multiple function calls
5. Configuration Management: Centralizes LLM settings for related functions
Example of Different Configurations:
python
@ai_model
class CreativeWriter:
llm_model = "gpt-4"
temperature = 0.9 # Higher temperature for more creative outputs
@ai_fn
def write_story(self, prompt: str) -> str:
"""Write a creative story based on the prompt."""
pass
@ai_model
class TechnicalWriter:
llm_model = "gpt-4"
temperature = 0.2 # Lower temperature for more focused, precise outputs
@ai_fn
def write_documentation(self, code: str) -> str:
"""Write technical documentation for the given code."""
pass
Common Patterns:
1. Task-Specific Agents:
python
@ai_model
class ResearchAgent:
llm_model = "gpt-4"
@ai_fn
def search(self, query: str) -> list[str]:
"""Generate search queries."""
pass
@ai_fn
def synthesize(self, results: list[str]) -> str:
"""Synthesize search results."""
pass
2. Chained Processing:
```python
@ai_model
class DocumentProcessor:
llm_model = "gpt-4"
@ai_fn
def extract_topics(self, text: str) -> list[str]:
"""Extract main topics from text."""
pass
@ai_fn
def summarize_topic(self, topicSanaz
06/30/2025, 1:09 PMMarvin
06/30/2025, 1:10 PMMarvin
06/30/2025, 1:11 PMpython
# Instead of
from marvin import ai_fn, ai_model # This won't work
# Use this instead
from marvin import fn as ai_fn # The decorator is now called 'fn'
from marvin import Agent as ai_model # The base class is now called 'Agent'
Here's a corrected example:
python
from marvin import fn as ai_fn
from marvin import Agent
from prefect import flow, task
@Agent
class ResearchAgent:
model = "openai:gpt-4" # Specify the model using provider:model_name format
@ai_fn
def research_topic(self, topic: str) -> str:
"""Research a topic and return key findings."""
pass
# Example usage
@flow
def research_flow(topic: str):
agent = ResearchAgent()
results = agent.research_topic(topic)
return results
If you're still having issues, make sure you:
1. Have the latest version of Marvin installed:
bash
pip install --upgrade marvin
2. Have set up your OpenAI API key:
python
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
3. Have all required dependencies:
bash
pip install "marvin[all]" # Installs all optional dependencies
Alternative Import Patterns:
python
# Option 1: Import everything
import marvin
@marvin.Agent
class MyAgent:
model = "openai:gpt-4"
@marvin.fn
def my_function(self, input: str) -> str:
"""Process the input."""
pass
# Option 2: Direct imports
from marvin import Agent, fn
@Agent
class MyAgent:
model = "openai:gpt-4"
@fn
def my_function(self, input: str) -> str:
"""Process the input."""
pass
The main changes to note are:
- ai_fn
is now fn
- ai_model
is now Agent
- Model specification uses the format provider:model_name
- Configuration is more streamlined with environment variables
Would you like me to provide a complete working example with these updated imports?Sanaz
06/30/2025, 1:25 PMMarvin
06/30/2025, 1:25 PMMarvin
06/30/2025, 1:26 PMbash
# Create and navigate to project directory
mkdir marvin-workflow
cd marvin-workflow
# Create virtual environment
python -m venv venv
# Activate virtual environment
# On Windows:
venv\Scripts\activate
# On Unix or MacOS:
source venv/bin/activate
2. Install required packages:
bash
pip install "marvin[all]"
pip install "prefect>=3.0.0"
3. Create a .env
file in your project directory:
env
OPENAI_API_KEY=your-api-key-here
MARVIN_AGENT_MODEL=openai:gpt-4
4. Create a new file called workflow.py
with this example code:
python
import os
from dotenv import load_dotenv
from marvin import fn
from marvin import Agent
from prefect import flow, task
# Load environment variables
load_dotenv()
@Agent
class ResearchAgent:
model = "openai:gpt-4"
@fn
def research_topic(self, topic: str) -> str:
"""Research the given topic and provide detailed findings."""
pass
@Agent
class SummaryAgent:
model = "openai:gpt-4"
@fn
def create_summary(self, research: str) -> str:
"""Create a concise summary of the research findings."""
pass
@task
def perform_research(topic: str) -> str:
agent = ResearchAgent()
return agent.research_topic(topic)
@task
def create_summary(research: str) -> str:
agent = SummaryAgent()
return agent.create_summary(research)
@flow(name="Research Flow")
def research_flow(topic: str):
# Perform the research
research_results = perform_research(topic)
# Create a summary
summary = create_summary(research_results)
return {
"research": research_results,
"summary": summary
}
if __name__ == "__main__":
# Example usage
topic = "The impact of artificial intelligence on modern workflow automation"
results = research_flow(topic)
print("\n=== Research Results ===")
print(results["research"])
print("\n=== Summary ===")
print(results["summary"])
5. Run the workflow:
There are several ways to run the workflow:
A. Direct Python execution:
bash
python workflow.py
B. Using Prefect CLI (recommended for production):
bash
# Start the Prefect server (in a separate terminal)
prefect server start
# In another terminal, start a worker
prefect worker start -p process
# Deploy your flow
prefect deploy workflow.py:research_flow -n "research-deployment" -p process
# Run the deployment
prefect deployment run research-flow/research-deployment
C. Interactive Python shell:
python
>>> from workflow import research_flow
>>> results = research_flow("The future of quantum computing")
>>> print(results["summary"])
6. Monitor your flow:
- Open the Prefect UI at http://127.0.0.1:4200 (if using Prefect server)
- View flow runs, logs, and results
Important Tips:
1. Error Handling:
python
from prefect import flow, task
from prefect.utilities.annotations import retry
@task(retries=3, retry_delay_seconds=60)
def perform_research(topic: str) -> str:
try:
agent = ResearchAgent()
return agent.research_topic(topic)
except Exception as e:
print(f"Error during research: {e}")
raise
2. Logging:
```python
from prefect import get_run_logger
@task
def perform_research(topic: str) -> str:
logger = get_run_logger()
logger.info(f"Starting research on topic: {topic}")
agent = ResearchAgent()Sanaz
06/30/2025, 1:53 PMMarvin
06/30/2025, 1:53 PMMarvin
06/30/2025, 1:54 PMbash
# Start the Prefect server
prefect server start
This will:
- Start a local PostgreSQL database
- Launch the API server
- Start the UI server (typically at http://127.0.0.1:4200)
2. Configure Client Connection
bash
# Point your client to the local server
prefect config set PREFECT_API_URL="<http://127.0.0.1:4200/api>"
3. Create a Work Pool
bash
# Create a process work pool
prefect work-pool create my-process-pool --type process
# Create a kubernetes work pool (if using k8s)
prefect work-pool create my-k8s-pool --type kubernetes
4. Example Workflow Setup
Create a file `example_flow.py`:
python
from prefect import flow, task
from datetime import timedelta
@task(retries=3, retry_delay_seconds=60)
def process_data(data: int) -> int:
return data * 2
@flow(name="Example Flow", description="A simple example flow")
def my_flow(number: int = 1):
result = process_data(number)
return result
if __name__ == "__main__":
my_flow(5)
5. Deploy the Flow
python
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
from example_flow import my_flow
# Create deployment
deployment = my_flow.to_deployment(
name="example-deployment",
work_pool_name="my-process-pool",
schedule=CronSchedule(cron="0 0 * * *"), # Run daily at midnight
tags=["example"]
)
# Apply the deployment
deployment.apply()
6. Start a Worker
bash
# Start a worker for your work pool
prefect worker start -p "my-process-pool"
7. Monitor and Manage
Access the UI at http://127.0.0.1:4200 to:
- Monitor flow runs
- View logs
- Manage deployments
- Configure work pools
- Set up notifications
8. Common Operations
python
from prefect import get_client
from prefect.deployments import run_deployment
async def manage_flows():
# Create API client
async with get_client() as client:
# Get flow runs
flow_runs = await client.read_flow_runs()
# Start a deployment
deployment_run = await run_deployment(
name="example-flow/example-deployment",
parameters={"number": 10}
)
# Get deployment details
deployment = await client.read_deployment("example-flow/example-deployment")
9. Environment Variables
Create a .env
file:
env
PREFECT_API_URL="<http://127.0.0.1:4200/api>"
PREFECT_SERVER_API_HOST="0.0.0.0"
PREFECT_SERVER_API_PORT="4200"
10. Production Configuration
For production, consider:
bash
# Set database URL
prefect config set PREFECT_API_DATABASE_CONNECTION_URL="<postgresql+asyncpg://user:pass@hostname:5432/db>"
# Set API URL
prefect config set PREFECT_API_URL="<http://your-server:4200/api>"
11. Server Management Commands
bash
# Check server status
prefect server status
# View server settings
prefect config view
# Clear server database (be careful!)
prefect server database reset
# Start server in development mode
prefect server start --dev
12. Troubleshooting
If you encounter issues:
bash
# Check server logs
prefect server logs
# Verify configuration
prefect config view
# Reset configuration
prefect config unset PREFECT_API_URL
# Check worker status
prefect worker status
Common Issues and Solutions:
- If the UI isn't accessible, check if port 4200 is available
- If workers can't connectSanaz
06/30/2025, 3:41 PMMarvin
06/30/2025, 3:41 PMMarvin
06/30/2025, 3:48 PMbash
pip install "marvin[all]"
pip install prefect
pip install duckduckgo-search
pip install beautifulsoup4
pip install requests
2. Create a file named `web_search_agent.py`:
```python
from typing import List, Dict
from marvin import Agent, fn
from prefect import flow, task
from duckduckgo_search import ddg
import requests
from bs4 import BeautifulSoup
from urllib.parse import urlparse
import time
@Agent
class WebSearchAgent:
model = "openai:gpt-4"
temperature = 0.3 # Lower temperature for more focused results
@fn
def generate_search_queries(self, topic: str) -> List[str]:
"""
Generate a list of specific search queries based on the main topic.
The queries should cover different aspects of the topic.
Return a list of 3-5 specific search queries.
"""
pass
@fn
def analyze_search_results(self,
search_results: List[Dict],
original_query: str) -> List[Dict]:
"""
Analyze search results and identify the most relevant ones.
Filter out any irrelevant or low-quality results.
Args:
search_results: List of search results with titles and descriptions
original_query: The original search query
Returns:
List of filtered and ranked results with relevance scores
"""
pass
@fn
def synthesize_information(self,
webpage_contents: List[Dict],
query: str) -> Dict:
"""
Synthesize information from multiple webpage contents into a coherent summary.
Args:
webpage_contents: List of dictionaries containing webpage content and metadata
query: Original search query
Returns:
Dictionary containing:
- summary: Overall summary of findings
- key_points: List of key points
- sources: List of sources used
"""
pass
@task(retries=2, retry_delay_seconds=30)
def fetch_webpage_content(url: str) -> Dict:
"""Fetch and extract content from a webpage."""
try:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
# Remove unwanted elements
for element in soup(['script', 'style', 'nav', 'footer', 'iframe']):
element.decompose()
# Extract main content
content = ' '.join(soup.stripped_strings)
return {
'url': url,
'domain': urlparse(url).netloc,
'content': content[:8000], # Limit content length
'status': 'success'
}
except Exception as e:
return {
'url': url,
'domain': urlparse(url).netloc,
'content': '',
'status': f'error: {str(e)}'
}
@task
def perform_web_search(query: str, max_results: int = 5) -> List[Dict]:
"""Perform web search using DuckDuckGo."""
try:
results = ddg(query, max_results=max_results)
return list(results)
except Exception as e:
print(f"Search error: {e}")
return []
@flow(name="Web Research Flow")
def web_research_flow(
topic: str,
max_results_per_query: int = 3,
max_queries: int = 3
) -> Dict:
"""
Orchestrates the web research process.
Args:
topic: The main topic to research
max_results_per_query: Maximum number of results toSanaz
06/30/2025, 3:57 PMMarvin
06/30/2025, 3:57 PMMarvin
06/30/2025, 3:58 PMpython
from marvin import Agent, fn
from marvin.tools.image import ImageTool
from PIL import Image
import io
@Agent
class ImageAnalysisAgent:
model = "openai:gpt-4-vision-preview" # Note: Use vision model for image analysis
@fn
def analyze_image(self, image_path: str) -> dict:
"""
Analyze the content of an image and provide detailed description.
"""
# Load and process image using ImageTool
image_tool = ImageTool()
image_data = image_tool.load(image_path)
return image_data
@fn
def generate_image_caption(self, image_path: str) -> str:
"""
Generate a descriptive caption for the image.
"""
pass
# Example usage
@flow
def process_image_flow(image_path: str):
agent = ImageAnalysisAgent()
analysis = agent.analyze_image(image_path)
caption = agent.generate_image_caption(image_path)
return {"analysis": analysis, "caption": caption}
2. Web Search
python
from marvin import Agent, fn
from marvin.tools.web import WebSearchTool, WebScrapeTool
from typing import List, Dict
@Agent
class WebResearchAgent:
model = "openai:gpt-4"
def __init__(self):
self.search_tool = WebSearchTool()
self.scrape_tool = WebScrapeTool()
@fn
def search_web(self, query: str, max_results: int = 5) -> List[Dict]:
"""
Perform web search and return results.
"""
results = self.search_tool.search(query, max_results=max_results)
return results
@fn
def scrape_webpage(self, url: str) -> Dict:
"""
Scrape content from a webpage.
"""
content = self.scrape_tool.scrape(url)
return content
# Example usage
@flow
def web_research_flow(query: str):
agent = WebResearchAgent()
search_results = agent.search_web(query)
detailed_results = []
for result in search_results:
content = agent.scrape_webpage(result['url'])
detailed_results.append(content)
return detailed_results
3. Audio Files
python
from marvin import Agent, fn
from marvin.tools.audio import AudioTool
import numpy as np
from typing import Dict
@Agent
class AudioProcessingAgent:
model = "openai:gpt-4"
def __init__(self):
self.audio_tool = AudioTool()
@fn
def transcribe_audio(self, audio_path: str) -> str:
"""
Transcribe speech from audio file to text.
"""
return self.audio_tool.transcribe(audio_path)
@fn
def analyze_audio(self, audio_path: str) -> Dict:
"""
Analyze audio characteristics and content.
"""
audio_data = self.audio_tool.load(audio_path)
return {
"duration": audio_data.duration,
"sample_rate": audio_data.sample_rate,
"channels": audio_data.channels,
"analysis": "Detailed analysis of audio content"
}
# Example usage
@flow
def process_audio_flow(audio_path: str):
agent = AudioProcessingAgent()
# Transcribe audio to text
transcript = agent.transcribe_audio(audio_path)
# Analyze audio characteristics
analysis = agent.analyze_audio(audio_path)
return {
"transcript": transcript,
"analysis": analysis
}
4. Combined Multimedia Processing
```python
from marvin import Agent, fnSanaz
06/30/2025, 4:28 PMMarvin
06/30/2025, 4:28 PMMarvin
06/30/2025, 4:30 PMpython
# Core Marvin imports
from marvin import Agent
from marvin.tools import search, web, images, audio
from prefect import flow, task
# If you need specific tools
from marvin.tools.search import DuckDuckGoSearchTool
from marvin.tools.web import WebScraper
from marvin.tools.images import ImageAnalyzer
from marvin.tools.audio import AudioProcessor
# Example implementation
@Agent
class MultiMediaAgent:
model = "openai:gpt-4"
def __init__(self):
self.search_tool = DuckDuckGoSearchTool()
self.web_scraper = WebScraper()
self.image_analyzer = ImageAnalyzer()
self.audio_processor = AudioProcessor()
def search_and_analyze(self, query: str):
# Implement your search and analysis logic here
results = self.search_tool.search(query)
return results
@flow
def multimedia_flow(query: str):
agent = MultiMediaAgent()
results = agent.search_and_analyze(query)
return results
Alternatively, for a simpler approach using just the basic tools:
python
from marvin import Agent
from prefect import flow, task
import requests
from bs4 import BeautifulSoup
from duckduckgo_search import ddg
@Agent
class SimpleWebSearchAgent:
model = "openai:gpt-4"
def search_web(self, query: str, max_results: int = 5):
try:
results = ddg(query, max_results=max_results)
return list(results)
except Exception as e:
print(f"Search error: {e}")
return []
def scrape_webpage(self, url: str):
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
return soup.get_text()
except Exception as e:
print(f"Scraping error: {e}")
return ""
@flow
def search_flow(query: str):
agent = SimpleWebSearchAgent()
search_results = agent.search_web(query)
all_content = []
for result in search_results:
content = agent.scrape_webpage(result['link'])
all_content.append({
'url': result['link'],
'content': content
})
return all_content
# Usage
if __name__ == "__main__":
results = search_flow("artificial intelligence latest developments")
print(results)
For this to work, make sure you have the required dependencies:
bash
pip install marvin
pip install prefect
pip install duckduckgo-search
pip install beautifulsoup4
pip install requests
Note: The exact tool names and imports might vary slightly depending on your Marvin version. If you encounter any specific import errors, please let me know the version you're using, and I can provide the exact imports for that version.
Would you like me to provide more specific examples for any particular functionality (web search, image processing, or audio processing)?Sanaz
06/30/2025, 4:54 PMMarvin
06/30/2025, 4:54 PMMarvin
06/30/2025, 4:55 PMpython
from marvin import Agent
from prefect import flow, task
from typing import List, Dict
# For web search
from duckduckgo_search import ddg
import requests
from bs4 import BeautifulSoup
# For image processing (if needed)
from PIL import Image
import io
# For audio processing (if needed)
import librosa
class WebSearchTool:
def search(self, query: str, max_results: int = 5) -> List[Dict]:
try:
results = ddg(query, max_results=max_results)
return list(results)
except Exception as e:
print(f"Search error: {e}")
return []
def scrape(self, url: str) -> str:
try:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(url, headers=headers, timeout=10)
soup = BeautifulSoup(response.text, 'html.parser')
return soup.get_text()
except Exception as e:
print(f"Scraping error: {e}")
return ""
class ImageTool:
def load_image(self, image_path: str):
try:
return Image.open(image_path)
except Exception as e:
print(f"Image loading error: {e}")
return None
def analyze_image(self, image):
# Add your image analysis logic here
return {
"size": image.size,
"mode": image.mode,
"format": image.format
}
class AudioTool:
def load_audio(self, audio_path: str):
try:
return librosa.load(audio_path)
except Exception as e:
print(f"Audio loading error: {e}")
return None
def analyze_audio(self, audio_data):
# Add your audio analysis logic here
return {
"duration": librosa.get_duration(y=audio_data[0], sr=audio_data[1]),
"sample_rate": audio_data[1]
}
@Agent
class MultiMediaAgent:
model = "openai:gpt-4"
def __init__(self):
self.web_tool = WebSearchTool()
self.image_tool = ImageTool()
self.audio_tool = AudioTool()
def search_and_analyze(self, query: str) -> Dict:
"""
Perform web search and analysis.
"""
search_results = self.web_tool.search(query)
analyzed_results = []
for result in search_results:
content = self.web_tool.scrape(result['link'])
analyzed_results.append({
'url': result['link'],
'title': result['title'],
'content': content
})
return analyzed_results
@flow
def multimedia_research_flow(
query: str,
max_results: int = 5
) -> Dict:
"""
Execute a multimedia research flow.
Args:
query: Search query
max_results: Maximum number of results to return
"""
agent = MultiMediaAgent()
results = agent.search_and_analyze(query)
return results
# Example usage
if __name__ == "__main__":
# Install required dependencies first:
# pip install marvin prefect duckduckgo-search beautifulsoup4 requests Pillow librosa
query = "Latest developments in artificial intelligence"
results = multimedia_research_flow(query)
print(results)
To use this code, you'll need to install the required dependencies:
```bash
# Core requirements
pip install marvin
pip install prefect
# For web search and scraping
pip install duckduckgo-searchSanaz
07/01/2025, 2:12 PMMarvin
07/01/2025, 2:12 PMMarvin
07/01/2025, 2:13 PMbash
pip install marvin prefect smol-agent selenium webdriver-manager
2. Here's an example that combines SmolAgent's web browsing capabilities with Marvin's agent framework:
python
from marvin import Agent
from prefect import flow, task
from smol_agent.agent import SmolAgent
from smol_agent.tools import WebBrowserTool
from typing import Dict, List
import asyncio
class WebBrowserAgent:
def __init__(self):
self.browser_tool = WebBrowserTool()
self.smol_agent = SmolAgent(
tools=[self.browser_tool],
llm_name="gpt-4"
)
async def visit_webpage(self, url: str) -> str:
"""Visit a webpage and extract its content using SmolAgent."""
prompt = f"""
Visit this webpage: {url}
Extract the main content and any relevant information.
Ignore ads and navigation elements.
"""
result = await self.smol_agent.run(prompt)
return result
async def search_and_browse(self, query: str) -> str:
"""Perform a search and browse results using SmolAgent."""
prompt = f"""
Search for: {query}
Visit the most relevant result and extract key information.
"""
result = await self.smol_agent.run(prompt)
return result
@Agent
class MarvinWebResearchAgent:
model = "openai:gpt-4"
def __init__(self):
self.web_browser = WebBrowserAgent()
async def research_topic(self, topic: str) -> Dict:
"""
Research a topic using SmolAgent for web browsing and Marvin for analysis.
"""
# Use SmolAgent to gather information
raw_info = await self.web_browser.search_and_browse(topic)
# Use Marvin to analyze and structure the information
analysis = {
"topic": topic,
"raw_information": raw_info,
"key_points": "Extract key points here",
"summary": "Provide summary here"
}
return analysis
@task
async def perform_research(topic: str) -> Dict:
agent = MarvinWebResearchAgent()
return await agent.research_topic(topic)
@flow
def web_research_flow(topic: str) -> Dict:
"""
Orchestrate web research using both SmolAgent and Marvin.
"""
return asyncio.run(perform_research(topic))
# Example with specific webpage visits
class DetailedWebResearchAgent:
def __init__(self):
self.web_browser = WebBrowserAgent()
async def analyze_multiple_pages(self, urls: List[str]) -> List[Dict]:
"""
Visit and analyze multiple web pages.
"""
results = []
for url in urls:
content = await self.web_browser.visit_webpage(url)
results.append({
"url": url,
"content": content
})
return results
@flow
def detailed_research_flow(urls: List[str]) -> List[Dict]:
"""
Analyze specific web pages using SmolAgent's browsing capabilities.
"""
agent = DetailedWebResearchAgent()
return asyncio.run(agent.analyze_multiple_pages(urls))
# Example usage
if __name__ == "__main__":
# Example 1: General topic research
topic = "Latest developments in quantum computing"
results = web_research_flow(topic)
print("Topic Research Results:", results)
# Example 2: Specific webpage analysis
urls = [
"<https://example.com/page1>",
"<https://example.com/page2>"
]
page_analyses = detailed_research_flow(urls)
print("Webpage Analyses:", page_analyses)
3. For more advanced web interactions, you can extend the SmolAgent capabilities:
```pythonSanaz
07/01/2025, 2:45 PMMarvin
07/01/2025, 2:45 PMMarvin
07/01/2025, 2:46 PMpython
from marvin import Agent
from prefect import flow, task
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager
from bs4 import BeautifulSoup
from typing import Dict, List
import time
class WebBrowserTool:
def __init__(self):
# Set up Chrome options
self.chrome_options = Options()
self.chrome_options.add_argument('--headless') # Run in headless mode
self.chrome_options.add_argument('--no-sandbox')
self.chrome_options.add_argument('--disable-dev-shm-usage')
# Initialize the webdriver
self.driver = webdriver.Chrome(
service=Service(ChromeDriverManager().install()),
options=self.chrome_options
)
def visit_webpage(self, url: str) -> str:
"""Visit a webpage and extract its content."""
try:
self.driver.get(url)
# Wait for dynamic content to load
time.sleep(2)
# Get the page source and parse with BeautifulSoup
soup = BeautifulSoup(self.driver.page_source, 'html.parser')
# Remove unwanted elements
for element in soup(['script', 'style', 'nav', 'footer']):
element.decompose()
return soup.get_text(strip=True)
except Exception as e:
print(f"Error visiting webpage: {e}")
return ""
def __del__(self):
"""Clean up the webdriver when done."""
if hasattr(self, 'driver'):
self.driver.quit()
@Agent
class WebResearchAgent:
model = "openai:gpt-4"
def __init__(self):
self.browser = WebBrowserTool()
def analyze_webpage(self, url: str) -> Dict:
"""
Visit and analyze a webpage using Marvin's capabilities.
"""
content = self.browser.visit_webpage(url)
# Use the agent to analyze the content
analysis = {
"url": url,
"content": content,
"summary": "Summary will be generated here",
"key_points": ["Key points will be extracted here"]
}
return analysis
@flow
def web_research_flow(urls: List[str]) -> List[Dict]:
"""
Research flow that visits and analyzes multiple webpages.
"""
agent = WebResearchAgent()
results = []
for url in urls:
analysis = agent.analyze_webpage(url)
results.append(analysis)
return results
# Example usage
if __name__ == "__main__":
# First, install required dependencies:
# pip install marvin prefect selenium webdriver-manager beautifulsoup4
urls = [
"<https://example.com/page1>",
"<https://example.com/page2>"
]
results = web_research_flow(urls)
print("Research Results:", results)
To use this code, you'll need to install these dependencies:
bash
pip install marvin
pip install prefect
pip install selenium
pip install webdriver-manager
pip install beautifulsoup4
This solution provides:
1. A WebBrowserTool that uses Selenium for actual web browsing
2. Integration with Marvin's Agent for content analysis
3. A Prefect flow for orchestrating the research process
The advantages of this approach:
- Can handle dynamic JavaScript-rendered content
- More reliable than simple HTTP requests
- Can be extended to handle more complex web interactions
- Integrates well with Marvin's AI capabilities
Would you like me to explain any part of this implementation in more detail or show how to extend it for specific use cases?Sanaz
07/01/2025, 8:53 PMMarvin
07/01/2025, 8:53 PMMarvin
07/01/2025, 8:53 PMai_fn
decorator was part of an older version of Marvin, but the import structure has changed.
Here's how to fix it:
1. First, make sure you have the latest version of Marvin installed:
bash
pip install -U marvin
2. Then, update your import statement to:
python
from marvin.components import ai_fn
If you're still having issues after that, you might want to check:
1. Your Python environment is clean (no conflicting packages)
2. You have the correct dependencies installed
3. Your OpenAI API key is properly configured
Also, if you're following any tutorials or documentation, make sure you're looking at the most current version, as the Marvin API has evolved over time.
If you continue to have issues, please let me know what version of Marvin you're using (you can check with pip show marvin
), and I can provide more specific guidance.Sanaz
07/01/2025, 8:57 PMMarvin
07/01/2025, 8:57 PMMarvin
07/01/2025, 8:57 PMai_fn
decorator has been replaced with a new pattern.
Here's how to use function creation in Marvin 3.x:
python
from marvin import fn
# Example usage:
@fn
def your_function(input: str) -> str:
"""Your function description here"""
return NotImplemented
The change from ai_fn
to fn
is part of Marvin's 3.0 release which included significant API changes to make the library more intuitive and consistent.
If you're following any tutorials or documentation that uses ai_fn
, you're likely looking at outdated material for Marvin 1.x or 2.x. I'd recommend referring to the current documentation at:
- https://www.askmarvin.ai/components/functions/
- https://www.askmarvin.ai/welcome/quickstart/
Let me know if you need any clarification on using the new fn
decorator!