python
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))

research_and_analyze_flow

bash
# 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

bash
export OPENAI_API_KEY="your-api-key-here"

python
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

bash
# Create a virtual environment
python -m venv venv

# Activate it
# On Windows:
venv\Scripts\activate
# On Unix or MacOS:
source venv/bin/activate

docx2txt

pillow

python-pptx

python-magic

bash
     # Ubuntu/Debian
     sudo apt-get install libmagic1
     
     # CentOS/RHEL
     sudo yum install file-devel

python
import prefect
import marvin
import openai
import duckduckgo_search
import magic
import pypdf
import pandas

print("All core dependencies successfully imported!")

requirements.txt

bash
pip freeze > requirements.txt

pyproject.toml

setup.py

python
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)

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

@ai_model

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

@ai_fn

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

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

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

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

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

python
# 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'

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

bash
pip install --upgrade marvin

python
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

bash
pip install "marvin[all]"  # Installs all optional dependencies

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

ai_fn

fn

ai_model

Agent

provider:model_name

bash
# 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

bash
pip install "marvin[all]"
pip install "prefect>=3.0.0"

.env

env
OPENAI_API_KEY=your-api-key-here
MARVIN_AGENT_MODEL=openai:gpt-4

workflow.py

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"])

bash
python workflow.py

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

python
>>> from workflow import research_flow
>>> results = research_flow("The future of quantum computing")
>>> print(results["summary"])

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

bash
# Start the Prefect server
prefect server start

bash
# Point your client to the local server
prefect config set PREFECT_API_URL="<http://127.0.0.1:4200/api>"

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

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)

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()

bash
# Start a worker for your work pool
prefect worker start -p "my-process-pool"

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")

.env

env
PREFECT_API_URL="<http://127.0.0.1:4200/api>"
PREFECT_SERVER_API_HOST="0.0.0.0"
PREFECT_SERVER_API_PORT="4200"

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>"

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

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

bash
pip install "marvin[all]"
pip install prefect
pip install duckduckgo-search
pip install beautifulsoup4
pip install requests

python
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}

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

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
    }

python
# 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

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)

bash
pip install marvin
pip install prefect
pip install duckduckgo-search
pip install beautifulsoup4
pip install requests

python
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)

bash
pip install marvin prefect smol-agent selenium webdriver-manager

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)

python
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)

bash
pip install marvin
pip install prefect
pip install selenium
pip install webdriver-manager
pip install beautifulsoup4

ai_fn

bash
pip install -U marvin

python
from marvin.components import ai_fn

pip show marvin

ai_fn

python
from marvin import fn

# Example usage:
@fn
def your_function(input: str) -> str:
    """Your function description here"""
    return NotImplemented

ai_fn

fn

ai_fn

fn

bash
pip install pandas openpyxl

python
from prefect import flow, task
import pandas as pd

@task
def read_excel_data(file_path: str) -> pd.DataFrame:
    return pd.read_excel(file_path)

@task
def process_data(df: pd.DataFrame):
    # Add your data processing logic here
    return df

@flow
def excel_processing_flow(file_path: str):
    # Read the Excel file
    df = read_excel_data(file_path)
    
    # Process the data
    processed_data = process_data(df)
    
    return processed_data

# Run the flow
if __name__ == "__main__":
    result = excel_processing_flow("path/to/your/excel_file.xlsx")

python
@task
def read_specific_sheet(file_path: str, sheet_name: str) -> pd.DataFrame:
    return pd.read_excel(file_path, sheet_name=sheet_name)

python
@task
def read_all_sheets(file_path: str) -> dict:
    return pd.read_excel(file_path, sheet_name=None)

python
from prefect import flow, task
import pandas as pd

@task
def read_excel_data(file_path: str) -> pd.DataFrame:
    try:
        return pd.read_excel(file_path)
    except FileNotFoundError:
        raise ValueError(f"Excel file not found at: {file_path}")
    except Exception as e:
        raise ValueError(f"Error reading Excel file: {str(e)}")

python
@task
def read_large_excel(file_path: str, chunksize: int = 1000):
    for chunk in pd.read_excel(file_path, chunksize=chunksize):
        # Process each chunk
        yield chunk

usecols

skiprows

parse_dates

na_values

python
@task
def read_excel_advanced(file_path: str) -> pd.DataFrame:
    return pd.read_excel(
        file_path,
        sheet_name='Sheet1',
        usecols='A:D',  # Specify columns to read
        skiprows=1,     # Skip first row
        na_values=['NA', 'missing'],  # Custom NA values
        parse_dates=['Date_Column']  # Parse date columns
    )

@fn

@flow

@task

@fn

@fn

@fn

orchestration_agent

@fn

@flow

@task

.id

@flow

@fn

marvin.run()

id

QA

python
from marvin import fn  # Make sure you're importing the decorator correctly

# Then use it as
@fn
def search_function(query: str) -> list[str]:
    ...

python
from dataclasses import dataclass

@dataclass
class QA:
    Answer: str
    # Add any other fields you need

python
# Instead of passing the function directly
context = {"func": search_function}  # Wrong

# Pass the result or a proper object
context = {"result": search_function(query)}  # Better

QA

python
try:
    result = marvin.run(task["instructions"], agents=[chosen_agent], context=context)
except Exception as e:
    print(f"Error executing agent: {e}")
    result = f"Error: {str(e)}"
return result

python
print(f"Agent decision: {agent_decision}")
print(f"Chosen agent: {chosen_agent.name if chosen_agent else 'None'}")

marvin.run()

python
# Create an orchestration agent instead of using the raw function
orchestration_agent = marvin.Agent(
    name="orchestrator",
    instructions="Route tasks to appropriate specialized agents based on the task requirements.",
    tools=[orchestration_agent]  # Your existing orchestration function becomes a tool
)

python
if __name__ == "__main__":
    with marvin.Thread() as thread:
        question = dataset[20]['Question']
        fileinfo = dataset[20]['file_path']
        Q = question + fileinfo

        plan = marvin.plan(Q)
        for i, task in enumerate(plan, 1):
            print(f"{i}. {task.instructions}")

        outputs = []
        for task in plan:
            print(f"\nExecuting task: {task.instructions}")

            # Use the orchestration agent object, not the function
            res = marvin.run(
                task.instructions,
                agents=[orchestration_agent],  # Now using the Agent object
                context={"question": question}
            )

            outputs.append(res)

        final = marvin.run(
            "Using the gathered outputs, write the final answer.",
            agents=[answer_agent],
            result_type=QA,
            context={"question": question, "info": outputs}
        )

    print("\n:bulb: Final Answer:\n")
    print(final.Answer)

id

python
@fn
def orchestration_agent(task: dict, context: dict) -> str:
    try:
        instructions = task["instructions"].lower()
        
        print(f"Processing instructions: {instructions}")  # Debug logging
        
        agent_decision = marvin.run(
            f"Based on the following task instructions: '{instructions}', which agent should handle this task?",
            agents=[answer_agent],
            context=context
        )
        
        print(f"Agent decision: {agent_decision}")  # Debug logging
        
        # ... rest of your function ...
        
    except Exception as e:
        print(f"Error in orchestration: {e}")
        raise

python
@fn
def orchestration_agent(task: dict, context: dict) -> str:
    instructions = task["instructions"].lower()
    file_path = context.get("file_path", "")  # Get file path from context
    
    # Enhanced decision making for image tasks
    if any(word in instructions.lower() for word in ["image", "picture", "photo", "graph", "plot", "figure"]) or \
       any(ext in file_path.lower() for ext in [".jpg", ".jpeg", ".png", ".gif"]):
        print("Routing to GPT Image Agent")  # Debug logging
        result = marvin.run(
            instructions,
            agents=[gpt_image_agent],
            context={"file_path": file_path, "question": context.get("question", "")}
        )
        return result

    # Regular agent selection for other tasks
    agent_decision = marvin.run(
        f"Based on the following task instructions: '{instructions}', which agent should handle this task?",
        agents=[answer_agent],
        context=context
    )

    agent_map = {
        "search": search_agent,
        "file": file_processing_agent,
        "arxiv": arxiv_agent,
        "image": gpt_image_agent,
        "audio": audio_transcription_agent,
        "youtube": youtube_download_agent
    }

    chosen_agent = None
    for agent_name, agent in agent_map.items():
        if agent_name in agent_decision.lower():
            chosen_agent = agent
            break

    if not chosen_agent:
        chosen_agent = answer_agent

    result = marvin.run(instructions, agents=[chosen_agent], context=context)
    return result

python
if __name__ == "__main__":
    with marvin.Thread() as thread:
        question = dataset[20]['Question']
        fileinfo = dataset[20]['file_path']
        Q = question + fileinfo

        plan = marvin.plan(Q)
        for i, task in enumerate(plan, 1):
            print(f"{i}. {task.instructions}")

        outputs = []
        for task in plan:
            print(f"\nExecuting task: {task.instructions}")

            # Include file_path in the context
            res = marvin.run(
                task.instructions,
                agents=[orchestration_agent],
                context={
                    "question": question,
                    "file_path": fileinfo,  # Add file path to context
                    "full_query": Q
                }
            )

            outputs.append(res)

        final = marvin.run(
            "Using the gathered outputs, write the final answer.",
            agents=[answer_agent],
            result_type=QA,
            context={"question": question, "info": outputs}
        )

    print("\n:bulb: Final Answer:\n")
    print(final.Answer)

python
gpt_image_agent = marvin.Agent(
    name="gpt-image",
    instructions="""Process and describe images with GPT-4o. 
    I can analyze graphs, charts, photographs, and other visual content.
    I will extract relevant information and provide detailed descriptions based on the question.""",
    tools=[gpt_4o_image_function]
)

dataset[20]

python
@fn
def orchestration_agent(task: dict, context: dict) -> str:
    print("\n=== Orchestration Debug ===")
    print(f"Received task: {task}")
    print(f"Context: {context}")
    
    instructions = task["instructions"].lower()
    
    # Debug the instruction analysis
    print("\nAnalyzing instructions...")
    is_image_task = any(word in instructions.lower() for word in ["image", "picture", "photo", "graph", "plot", "figure", "ocr"])
    print(f"Is image task? {is_image_task}")
    
    # If it's clearly an image task, route directly
    if is_image_task:
        print("\nRouting directly to GPT Image Agent")
        try:
            result = marvin.run(
                instructions,
                agents=[gpt_image_agent],
                context=context
            )
            print(f"Image agent result: {result}")
            return result
        except Exception as e:
            print(f"Error in image agent execution: {str(e)}")
            raise

    # If not clearly an image task, use LLM for decision
    print("\nAsking LLM for agent decision...")
    try:
        agent_decision = marvin.run(
            f"Based on the following task instructions: '{instructions}', which agent should handle this task?",
            agents=[answer_agent],
            context=context
        )
        print(f"LLM agent decision: {agent_decision}")
    except Exception as e:
        print(f"Error in agent decision: {str(e)}")
        raise

    agent_map = {
        "search": search_agent,
        "file": file_processing_agent,
        "arxiv": arxiv_agent,
        "image": gpt_image_agent,
        "audio": audio_transcription_agent,
        "youtube": youtube_download_agent
    }

    chosen_agent = None
    for agent_name, agent in agent_map.items():
        if agent_name in agent_decision.lower():
            chosen_agent = agent
            print(f"\nChosen agent: {agent_name}")
            break

    if not chosen_agent:
        print("\nNo specific agent matched, defaulting to answer agent")
        chosen_agent = answer_agent

    print(f"\nExecuting task with {chosen_agent.name} agent...")
    try:
        result = marvin.run(instructions, agents=[chosen_agent], context=context)
        print(f"Execution result: {result}")
        return result
    except Exception as e:
        print(f"Error in task execution: {str(e)}")
        raise

# Also modify your main workflow loop for debugging
if __name__ == "__main__":
    with marvin.Thread() as thread:
        question = dataset[20]['Question']
        fileinfo = dataset[20]['file_path']
        Q = question + fileinfo

        plan = marvin.plan(Q)
        print("\n=== Execution Plan ===")
        for i, task in enumerate(plan, 1):
            print(f"{i}. {task.instructions}")

        outputs = []
        for i, task in enumerate(plan):
            print(f"\n=== Executing Task {i+1} ===")
            print(f"Instructions: {task.instructions}")
            print(f"File path exists: {os.path.exists(fileinfo)}")  # Add this line
            
            try:
                res = marvin.run(
                    task.instructions,
                    agents=[orchestration_agent],
                    context={
                        "question": question,
                        "file_path": fileinfo,
                        "full_query": Q
                    }
                )
                print(f"Task {i+1} Result: {res}")
                outputs.append(res)
            except Exception as e:
                print(f"Error in task {i+1}: {str(e)}")
                print(f"Error type: {type(e)}")
                print(f"Full traceback: {traceback.format_exc()}")

This debugging code will help us understand:
1. Which agent is being selected
2. Whether the file path is valid
3. What decisions are being made in the orchestration process
4. Where exactly any errors are occurring

Run this modified code and share the debug output - it will help us pinpoint exactly where the process is failing and why the image processing isn't being handled correctly.

Also, for this specific task (extracting red and green numbers), you might want to modify the gpt_image_agent's instructions to be more specific:

```python
gpt_image_agent = marvin.Agent(
    name="gpt-image",
    instructions="""Process images with GPT-4o to extract numerical information. 
    I can identify and extract numbers of different colors, read text and numbers from images,
    and perform OCR-like tasks. For colored numbers, I will explicitly state the color and value of each number found.""",
    tools=[gpt_4o_image_function]
)

Also, update the GPT-4o image agent to be more specific about number extraction:

```python
gpt_image_agent = marvin.Agent(
    name="gpt-image",
    instructions="""
    Analyze images to extract colored numbers. When asked about numbers:
    1. Identify all numbers in the image
    2. Classify them by color (specifically red and green)
    3. Return them in a structured format:
       Red numbers: [list of numbers]
       Green numbers: [list of numbers]
    Be precise in color identification and number extraction.
    """,
    tools=[gpt_4o_image_function]
)

Key improvements in this version:

1. *Better Task Analysis*: The orchestration agent now uses a structured prompt to select the appropriate agent.

2. *Standardized Context*: All tasks receive consistent context including previous results.

3. *Robust Error Handling*: Each level has proper error handling and debugging output.

4. *Clear Agent Definitions*: Each agent has specific, detailed instructions about its capabilities.

5. *Task State Management*: Previous results are passed forward in the context.

To use this system:

1. Make sure all your agent functions (`search_function`, `file_analysis_function`, etc.) are properly defined and return results in a consistent format.

2. Test with a simple task first to verify the orchestration is working:
```python
result = marvin.run(
    "Search for information about Python",
    agents=[orchestration_agent],
    context={"question": "What is Python?"}
)

python
@fn
def gpt_4o_image_function(image_path: str, Q: str) -> str:
    try:
        print(f"\n=== Processing Image ===")
        print(f"Image path: {image_path}")
        print(f"Question: {Q}")
        
        # Verify file exists
        if not os.path.exists(image_path):
            return f"Error: Image file not found at {image_path}"
            
        # Read the image
        with open(image_path, "rb") as image_file:
            image_data = image_file.read()
            
        # Create a more direct prompt for the vision model
        prompt = """
        Look at this image and:
        1. Identify all numbers that are colored in RED. List them as numbers, not words.
        2. Identify all numbers that are colored in GREEN. List them as numbers, not words.
        
        Format your response EXACTLY like this:
        RED NUMBERS: [x, y, z]
        GREEN NUMBERS: [a, b, c]
        
        Only provide the lists of numbers, no explanation or additional text.
        """
            
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": prompt
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{base64.b64encode(image_data).decode('utf-8')}"
                            }
                        }
                    ]
                }
            ],
            max_tokens=100  # Limit response length to encourage concise output
        )
        
        result = response.choices[0].message.content
        print(f"Raw GPT-4o response: {result}")
        
        # Validate the response format
        if "RED NUMBERS:" not in result or "GREEN NUMBERS:" not in result:
            return "Error: Vision model did not return properly formatted number lists"
            
        return result

    except Exception as e:
        print(f"Error in image processing: {str(e)}")
        return f"Error processing image: {str(e)}"

python
gpt_image_agent = marvin.Agent(
    name="gpt-image",
    instructions="""You are a specialized image analysis agent that:
    1. ONLY extracts colored numbers from images
    2. Returns EXACTLY two lists: red numbers and green numbers
    3. Uses the format:
       RED NUMBERS: [x, y, z]
       GREEN NUMBERS: [a, b, c]
    4. Never provides explanations or theoretical examples
    5. Returns only the actual numbers found in the image""",
    tools=[gpt_4o_image_function]
)

python
@fn
def orchestration_agent(task: dict, context: dict) -> str:
    print("\n=== Orchestration Debug ===")
    print(f"Task received: {task}")
    print(f"Context received: {context}")

    # Get instructions from task
    instructions = task.get("instructions") if isinstance(task, dict) else str(task)
    
    # Get file path from context
    file_path = context.get("fileinfo")
    
    # If this is an image-related task
    if any(word in instructions.lower() for word in ["image", "picture", "photo", "ocr", "numbers"]):
        print(f"\nRouting to image agent with path: {file_path}")
        chosen_agent = gpt_image_agent
        
        # Pass the full task instructions and file path to the image agent
        modified_context = {
            "file_path": file_path,
            "task_instructions": instructions,
            "image_path": file_path,  # Explicit path for the image function
            "Q": instructions  # Pass full instructions as the question
        }
    else:
        # ... other agent selection logic ...
        chosen_agent = answer_agent
        modified_context = context

    print(f"Selected agent: {chosen_agent.name}")
    print(f"Modified context: {modified_context}")

    # Execute with modified context
    result = marvin.run(
        instructions,
        agents=[chosen_agent],
        context=modified_context
    )
    
    return result

@fn
def gpt_4o_image_function(image_path: str, Q: str) -> str:
    print("\n=== Image Function Debug ===")
    print(f"Received image_path: {image_path}")
    print(f"Received task: {Q}")
    
    return get_image_description_gpt4o(image_path, Q)

def get_image_description_gpt4o(image_path: str, Q: str) -> str:
    print(f"\n=== GPT-4o Processing ===")
    print(f"Processing image at: {image_path}")
    print(f"Processing task: {Q}")
    print(f"File exists: {os.path.exists(image_path)}")

    try:
        with open(image_path, "rb") as image_file:
            image_data = image_file.read()

        # Modified prompt to focus on the specific task
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"Complete this specific task for the image: {Q}"
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64.b64encode(image_data).decode('utf-8')}"
                        }
                    }
                ]
            }]
        )
        return response.choices[0].message.content

    except FileNotFoundError:
        return f"Error: Image file not found at {image_path}"
    except Exception as e:
        return f"Error processing image: {str(e)}"

# Main workflow remains the same, but ensure context includes file path
if __name__ == "__main__":
    with marvin.Thread() as thread:
        question = dataset[20]['Question']
        fileinfo = dataset[20]['file_path']
        
        plan = marvin.plan(question)
        outputs = []
        
        for task in plan:
            print(f"\nExecuting task: {task.instructions}")
            
            res = marvin.run(
                task.instructions,
                agents=[orchestration_agent],
                context={
                    "question": question,
                    "fileinfo": fileinfo,
                    "task": task
                }
            )
            outputs.append(res)

And in your main workflow, ensure you're passing the file path correctly:

```python
if __name__ == "__main__":
    with marvin.Thread() as thread:
        question = dataset[20]['Question']
        fileinfo = dataset[20]['file_path']
        
        plan = marvin.plan(question)
        outputs = []
        
        for task in plan:
            print(f"\nExecuting task: {task.instructions}")
            
            # Ensure file path is included in context
            res = marvin.run(
                task.instructions,
                agents=[orchestration_agent],
                context={
                    "question": question,
                    "fileinfo": fileinfo,  # This should be the full path to your image
                    "task": task,
                    "file_path": fileinfo  # Include both for compatibility
                }
            )
            outputs.append(res)

python
# Example of context usage in Marvin agents
@fn
def orchestration_agent(task: dict, context: dict) -> str:
    # Context received from marvin.run() call
    print(f"Context received: {context}")
    
    # Context can include things like:
    file_path = context.get("file_path")  # File paths
    question = context.get("question")     # Original question
    task_info = context.get("task")       # Task information
    previous_results = context.get("previous_results")  # Results from previous tasks

    # You can also access global context using marvin.context
    global_context = marvin.context.get()

    # When calling another agent, you can pass modified context
    modified_context = {
        "image_path": file_path,
        "Q": question,
        "instructions": task_info,
        # Add any other relevant information
    }

    result = marvin.run(
        instructions,
        agents=[chosen_agent],
        context=modified_context  # This context will be available to the next agent
    )

python
# In main workflow
res = marvin.run(
    task.instructions,
    agents=[orchestration_agent],
    context={
        "question": original_question,
        "file_path": image_path,
        "previous_results": previous_outputs
    }
)

python
@fn
def gpt_4o_image_function(image_path: str, Q: str) -> str:
    # Get additional context beyond function parameters
    context = marvin.context.get()
    specific_instructions = context.get("instructions")
    original_task = context.get("task")

python
@fn
def process_with_history(input_data: str) -> str:
    context = marvin.context.get()
    previous_results = context.get("previous_results", [])
    # Use previous results to inform current processing

python
# Set up common configuration in main workflow
common_context = {
    "model_config": {"temperature": 0.7},
    "api_keys": {"service_key": "xyz"},
    "file_paths": {"base_dir": "/path/to/files"}
}

res = marvin.run(
    instructions,
    agents=[agent],
    context=common_context
)

python
@fn
def safe_processing_agent(task: dict, context: dict) -> str:
    try:
        # Access debug information from context
        debug_mode = context.get("debug", False)
        if debug_mode:
            print(f"Processing task with context: {context}")
            
        result = process_task(task)
        return result
    except Exception as e:
        # Add error information to context
        error_context = {
            "error": str(e),
            "task": task,
            "timestamp": datetime.now()
        }
        marvin.context.update(error_context)
        raise

python
@fn
def my_agent(task: dict, context: dict) -> str:
    """
    Agent that processes tasks with specific context requirements.
    
    Expected context:
    - file_path: str - Path to input file
    - question: str - Original question
    - task: dict - Task information
    - previous_results: list - Results from previous tasks
    """
    # Agent implementation

python
@fn
def image_processing_agent(task: dict, context: dict) -> str:
    required_keys = ["image_path", "instructions"]
    missing_keys = [key for key in required_keys if key not in context]
    
    if missing_keys:
        return f"Error: Missing required context keys: {missing_keys}"

For your specific case with image processing, you might want to structure your context like this:

```python
# In main workflow
context = {
    "question": dataset[20]['Question'],
    "file_path": dataset[20]['file_path'],
    "task": current_task,
    "execution_info": {
        "step": current_step,
        "total_steps": total_steps
    },
    "image_processing": {
        "image_path": image_path,
        "instructions": specific_instructions,
        "expected_output": "list_of_numbers"
    }
}

# This context will be available throughout the execution chain
res = marvin.run(
    task.instructions,
    agents=[orchestration_agent],
    context=context
)

python
# First, define the image processing function
@fn
def gpt_4o_image_function(image_path: str, Q: str) -> str:
    print("\n>>>>> IMAGE FUNCTION START <<<<<")
    print(f"IMAGE PATH: {image_path}")
    print(f"QUESTION: {Q}")
    
    try:
        with open(image_path, "rb") as image_file:
            image_data = image_file.read()
            print("Image file read successfully")

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": f"Complete this specific task for the image: {Q}"},
                    {"type": "image_url", "image_url": {
                        "url": f"data:image/jpeg;base64,{base64.b64encode(image_data).decode('utf-8')}"
                    }}
                ]
            }]
        )
        return response.choices[0].message.content

    except Exception as e:
        print(f"Error: {str(e)}")
        return f"Error processing image: {str(e)}"

# Define the image agent properly
gpt_image_agent = marvin.Agent(
    name="gpt-image",
    instructions="""You are an image analysis specialist that uses GPT-4 Vision to:
    1. Process and analyze images
    2. Extract text and numbers from images
    3. Identify colors and patterns
    4. Return specific information as requested
    
    Always use the provided image processing function to analyze images.""",
    tools=[gpt_4o_image_function]
)

# Modify the orchestration agent
@fn
def orchestration_agent(task: dict, context: dict) -> str:
    print("\n>>>>> ORCHESTRATION START <<<<<")
    print(f"TASK RECEIVED: {task}")
    print(f"CONTEXT RECEIVED: {context}")

    instructions = task.get("instructions") if isinstance(task, dict) else str(task)
    
    # Check for image task
    is_image_task = any(word in instructions.lower() for word in [
        "image", "picture", "photo", "ocr", "extract from image", "numbers in the image"
    ])

    if is_image_task:
        print("\n>>>>> ROUTING TO IMAGE AGENT")
        # Use the image agent instead of the raw function
        result = marvin.run(
            instructions,
            agents=[gpt_image_agent],  # Use the agent, not the function
            context={
                "image_path": context.get("fileinfo"),
                "Q": instructions
            }
        )
        return result

    # ... rest of the orchestration logic ...

# Main workflow
if __name__ == "__main__":
    with marvin.Thread() as thread:
        question = dataset[20]['Question']
        fileinfo = dataset[20]['file_path']
        
        print(f"\n>>>>> PROCESSING:")
        print(f"QUESTION: {question}")
        print(f"FILE PATH: {fileinfo}")
        
        plan = marvin.plan(question)
        outputs = []
        
        for task in plan:
            print(f"\n>>>>> EXECUTING TASK: {task.instructions}")
            
            res = marvin.run(
                task.instructions,
                agents=[orchestration_agent],
                context={
                    "question": question,
                    "fileinfo": fileinfo,
                    "task": task
                }
            )
            outputs.append(res)

python
orchestration_agent = marvin.Agent(
    name="orchestrator",
    tools=[get_image_description_gpt4o]  # Direct function as tool
)

python
gpt_image_agent = marvin.Agent(
    name="gpt-image",
    tools=[gpt_4o_image_function]
)

orchestration_agent = marvin.Agent(
    name="orchestrator",
    tools=[...]  # Other tools
)

Agent "orchestrator" (1396b4fc)
Tool:    get_image_description_gpt4o
Status:  :white_check_mark:
Input    {
    'image_path': 'C:\\Users\\a\\.cache\\huggingface\\hub\\datasets--gaia-benchmark--GAIA\\snaps…
    'Q': 'Identify the red and green numbers in the image.'
}

marvin.plan()

marvin.plan()

Task

python
from marvin import plan

tasks = plan(
    instructions="Build a data pipeline that processes customer data",
    context={"data_source": "CSV files", "output_format": "PostgreSQL"}
)

# Each returned task will have properties like:
# - description
# - status
# - dependencies
# - etc.

marvin.plan()

marvin.plan()

python
import marvin

# Change the default model for all agents
marvin.defaults.model = "openai:gpt-4"  # or any other supported model

python
from marvin import Agent

# Create an agent with a specific model
agent = Agent(
    model="openai:gpt-4",  # Specify the model
    model_settings={
        "temperature": 0.7,  # Optional: Adjust model parameters
        "max_tokens": 1000
    }
)

provider:model_name

python
# OpenAI models
"openai:gpt-4"
"openai:gpt-3.5-turbo"

bash
export OPENAI_API_KEY="your-api-key"

python
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"

python
model_settings = {
    "temperature": 0.7,      # Controls randomness (0.0 to 1.0)
    "max_tokens": 1000,      # Maximum tokens in response
    "top_p": 1.0,           # Nuclear sampling parameter
    "frequency_penalty": 0,  # Penalty for frequent token use
    "presence_penalty": 0    # Penalty for new token introduction
}

agent = Agent(model="openai:gpt-4", model_settings=model_settings)

marvin.defaults.model

bash
pip install marvin

bash
# Install uv first if you haven't already
curl -LsSf <https://astral.sh/uv/install.sh> | sh

# Then install Marvin
uv pip install marvin

bash
export OPENAI_API_KEY="your-api-key"

python
import marvin

bash
# Create a virtual environment
python -m venv venv

# Activate it (on Unix/macOS)
source venv/bin/activate

# Activate it (on Windows)
.\venv\Scripts\activate

# Install Marvin
pip install marvin

bash
pip install "marvin[dev]"  # Install with development dependencies

python
from marvin import ai_fn

@ai_fn
def greet(name: str) -> str:
    """Generate a friendly greeting."""
    pass

# Test the function
print(greet("World"))

python
from marvin import Agent, Memory

# Create a memory instance
memory = Memory()

# Create an agent with memory
agent = Agent(
    name="MyAgent",
    memories=[memory]
)

python
# Store information in memory
await memory.store("The user prefers Python over JavaScript")

# Store multiple facts
await memory.store([
    "The user is working on a data pipeline",
    "The user's name is Alice"
])

python
# Retrieve relevant memories
relevant_memories = await memory.recall("What does the user prefer?")

python
from marvin import Agent, Memory, assistant

@assistant()
async def coding_assistant(message: str, memory: Memory):
    # Recall relevant information before responding
    context = await memory.recall(message)
    
    # Use the context in your response
    response = f"Based on what I remember: {context}\n"
    response += "Here's my response to your question..."
    
    # Store new information from the conversation
    await memory.store(f"User asked about: {message}")
    
    return response

BaseTool

langchain_core.tools

python
from marvin import tool

@tool
async def my_custom_tool(input_parameter: str) -> str:
    """
    Description of what this tool does
    """
    # Your tool implementation here
    return result

python
   from marvin import tool
   from langchain_core.tools import BaseTool
   
   # Wrap your LangChain tool functionality in a Marvin tool
   @tool
   async def langchain_wrapper(input_parameter: str) -> str:
       """
       Description of what this wrapped tool does
       """
       # Implement your LangChain tool logic here
       # but wrap it in Marvin's tool decorator
       return result

python
from marvin import tool

@tool
def calculate_metrics(data: list[float], threshold: float = 0.5) -> dict:
    """
    Calculate metrics for a list of numbers based on a threshold.
    
    Args:
        data: List of numerical values to analyze
        threshold: Cutoff value for analysis (default: 0.5)
    
    Returns:
        Dictionary containing calculated metrics
    """
    # Function implementation
    pass

list[float]

float

python
from marvin import Agent
from prefect import flow

# Define your tools
@tool
def process_data(input_text: str, max_length: int = 100) -> str:
    """
    Process input text with a maximum length constraint.
    
    Args:
        input_text: The text to process
        max_length: Maximum allowed length
    """
    return input_text[:max_length]

# Create an agent with tools
agent = Agent(
    tools=[process_data],
    instructions="You are an agent that processes text data."
)

@flow
def process_text_flow(text: str):
    # The agent will automatically understand how to use the tool
    # and pass the appropriate arguments
    result = agent.run(f"Process this text: {text}")
    return result

python
from marvin import Agent

# Using OpenAI's GPT-4
agent = Agent(
    model="openai:gpt-4",
    instructions="Your agent instructions here"
)

# Using GPT-3.5-turbo
agent = Agent(
    model="openai:gpt-3.5-turbo",
    instructions="Your agent instructions here"
)

# Using Anthropic's Claude
agent = Agent(
    model="anthropic:claude-2",
    instructions="Your agent instructions here"
)

python
agent = Agent(
    model="openai:gpt-4",
    model_settings={
        "temperature": 0.7,
        "max_tokens": 500,
        "top_p": 1.0
    },
    instructions="Your agent instructions here"
)

python
import marvin

# Set default model for all agents
marvin.settings.model = "openai:gpt-4"

openai:

anthropic:

cohere:

google:

bedrock:

groq:

mistral:

bash
# For OpenAI
export OPENAI_API_KEY=your_api_key

# For Anthropic
export ANTHROPIC_API_KEY=your_api_key

# For other providers, check their specific environment variable requirements

python
from pydantic_ai.models import OpenAIModel

custom_model = OpenAIModel(
    model="gpt-4",
    temperature=0.7,
    max_tokens=500
)

agent = Agent(
    model=custom_model,
    instructions="Your agent instructions here"
)

python
import os
from marvin import Agent

# Set the environment variables
os.environ["OPENAI_API_BASE"] = "<http://127.0.0.1:your_port>"  # Your base URL
os.environ["OPENAI_API_KEY"] = "SS"  # Your API key

# Create the agent (it will automatically use the environment variables)
agent = Agent(
    model="openai:your-model-name",  # Replace with your model name
    instructions="Your instructions here"
)

python
from marvin import Agent
from pydantic_ai.models import OpenAIModel
from pydantic_ai.providers import OpenAIProvider
import openai

# Create custom OpenAI client
client = openai.Client(
    base_url="<http://127.0.0.1:your_port>",
    api_key="SS"
)

# Create custom OpenAI provider
provider = OpenAIProvider(openai_client=client)

# Create custom model configuration
model = OpenAIModel(
    model="your-model-name",  # Replace with your model name
    provider=provider
)

# Create the agent with the custom model
agent = Agent(
    model=model,
    instructions="Your instructions here"
)

python
import marvin
from marvin import Agent

# Configure Marvin settings
marvin.settings.openai.api_base = "<http://127.0.0.1:your_port>"
marvin.settings.openai.api_key = "SS"

# Create the agent
agent = Agent(
    model="openai:your-model-name",  # Replace with your model name
    instructions="Your instructions here"
)

your_port

your-model-name

python
# Test the agent
result = agent.run("Hello, are you connected to the correct model?")
print(result)

marvin.plan()

python
import os
import marvin

# Set environment variables before using plan()
os.environ["MARVIN_AGENT_MODEL"] = "openai:gpt-4"  # or your desired model
os.environ["OPENAI_API_KEY"] = "your-api-key"

# Now use plan()
result = marvin.plan("Your planning prompt here")

python
import marvin

# Configure settings before using plan()
marvin.settings.model = "openai:gpt-3.5-turbo"
marvin.settings.model_settings = {
    "temperature": 0.7,
    "max_tokens": 500
}

result = marvin.plan("Your planning prompt here")

python
import marvin
import os

# Set custom endpoint
os.environ["OPENAI_API_BASE"] = "<http://127.0.0.1:your_port>"
os.environ["OPENAI_API_KEY"] = "SS"
os.environ["MARVIN_AGENT_MODEL"] = "your-model-name"

# Use plan()
result = marvin.plan("Your planning prompt here")

.env

env
MARVIN_AGENT_MODEL=your-model-name
OPENAI_API_BASE=<http://127.0.0.1:your_port>
OPENAI_API_KEY=SS

python
from dotenv import load_dotenv
import marvin

load_dotenv()
result = marvin.plan("Your planning prompt here")

python
import marvin
from pydantic_ai.models import OpenAIModel
from pydantic_ai.providers import OpenAIProvider
import openai

# Create custom client
client = openai.Client(
    base_url="<http://127.0.0.1:your_port>",
    api_key="SS"
)

# Configure provider and model
provider = OpenAIProvider(openai_client=client)
model = OpenAIModel(
    model="your-model-name",
    provider=provider
)

# Set as default model
marvin.settings.model = model

# Use plan()
result = marvin.plan("Your planning prompt here")

marvin.plan()

python
# Test the configuration
print(f"Current model: {marvin.settings.model}")
print(f"Current model settings: {marvin.settings.model_settings}")

marvin.settings.model = model

marvin.plan()

python
import os
import marvin

# Set these before using marvin.plan()
os.environ["OPENAI_API_BASE"] = "<http://127.0.0.1:your_port>"
os.environ["OPENAI_API_KEY"] = "SS"

# Use plan()
result = marvin.plan("Your planning prompt here")

.env

env
OPENAI_API_BASE=<http://127.0.0.1:your_port>
OPENAI_API_KEY=SS

python
from dotenv import load_dotenv
import marvin

load_dotenv()
result = marvin.plan("Your planning prompt here")

python
import os

# Make sure the URL format is exactly correct - common issues include:
# - Missing http:// or https://
# - Wrong port number
# - Missing or extra slashes
os.environ["OPENAI_API_BASE"] = "<http://127.0.0.1:1234>"  # adjust port number
os.environ["OPENAI_API_KEY"] = "SS"

python
import openai

client = openai.Client(
    base_url="<http://127.0.0.1:1234>",  # your actual port
    api_key="SS"
)

# Test the connection
try:
    # Basic test call
    response = client.chat.completions.create(
        model="your-model-name",
        messages=[{"role": "user", "content": "Hello"}]
    )
    print("Connection successful!")
except Exception as e:
    print(f"Connection failed: {str(e)}")

python
import os
os.environ["OPENAI_API_BASE"] = "<http://127.0.0.1:1234>"
os.environ["OPENAI_API_KEY"] = "SS"
# If needed for non-SSL endpoints
os.environ["OPENAI_VERIFY_SSL"] = "false"

python
import marvin
from typing import Dict, List
import time
import json
from datetime import datetime

class AgentEvaluator:
    def __init__(self, test_cases: List[dict], models: List[str]):
        self.test_cases = test_cases
        self.models = models
        self.results = {}
        
    def run_evaluation(self):
        for model in self.models:
            self.results[model] = self._evaluate_model(model)
            
    def _evaluate_model(self, model_identifier: str) -> Dict:
        results = []
        
        # Configure environment for the model
        self._setup_model_env(model_identifier)
        
        for test_case in self.test_cases:
            try:
                start_time = time.time()
                
                # Create agent with the specific model
                agent = marvin.Agent(
                    model=model_identifier,
                    instructions=test_case.get('instructions', "You are a helpful AI assistant.")
                )
                
                # Run the test
                response = agent.run(test_case['input'])
                
                end_time = time.time()
                
                result = {
                    'test_case_id': test_case['id'],
                    'input': test_case['input'],
                    'output': response,
                    'execution_time': end_time - start_time,
                    'success': True,
                    'error': None
                }
                
            except Exception as e:
                result = {
                    'test_case_id': test_case['id'],
                    'input': test_case['input'],
                    'output': None,
                    'execution_time': None,
                    'success': False,
                    'error': str(e)
                }
                
            results.append(result)
            
        return results
    
    def _setup_model_env(self, model_identifier):
        # Configure environment variables based on model
        if model_identifier.startswith('openai:'):
            import os
            os.environ["OPENAI_API_KEY"] = "your-key"
        # Add other model configurations as needed
    
    def save_results(self, filename=None):
        if filename is None:
            filename = f"agent_evaluation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        
        with open(filename, 'w') as f:
            json.dump(self.results, f, indent=2)
            
    def print_summary(self):
        for model, results in self.results.items():
            success_rate = len([r for r in results if r['success']]) / len(results)
            avg_time = sum(r['execution_time'] for r in results if r['success']) / len(results)
            
            print(f"\nModel: {model}")
            print(f"Success Rate: {success_rate*100:.2f}%")
            print(f"Average Execution Time: {avg_time:.2f} seconds")

python
test_cases = [
    {
        'id': 1,
        'input': "Analyze this data and create a summary report",
        'instructions': "You are a data analysis expert.",
        'expected_output_type': 'report',
    },
    {
        'id': 2,
        'input': "Create a marketing strategy for a new product",
        'instructions': "You are a marketing strategist.",
        'expected_output_type': 'strategy',
    },
    # Add more test cases
]

# Define models to test
models_to_test = [
    "openai:gpt-4",
    "openai:gpt-3.5-turbo",
    "anthropic:claude-2",
    # Add your custom endpoint
    "your-custom-model"  # Configure this appropriately
]

4. *Add Custom Metrics*
```python
from typing import Callable

class MetricsCalculator:
    @staticmethod
    def response_length(response: str) -> int:
        return len(response.split())
    
    @staticmethod
    def contains_key_elements(response: str, required_elements: List[str]) -> float:
        found = sum(1 for elem in required_elements if elem.lower() in response.lower())
        return found / len(required_elements)

class EnhancedAgentEvaluator(AgentEvaluator):
    def __init__(self, test_cases: List[dict], models: List[str], metrics: Dict[str, Callable] = None):
        super().__init__(test_cases, models)
        self.metrics = metrics or {
            'response_length': MetricsCalculator.response_length,
        }
    
    def _evaluate_model(self, model_identifier: str) -> Dict:
        results = super()._evaluate_model(model_identifier)
        
        # Add metrics calculations
        for result in results:
            if result['success']:
                result['metrics'] = {
                    metric_name: metric_func(result['output'])
                    for metric_name, metric_func in self.metrics.items()
                }
        
        return results

python
# Define custom metrics
custom_metrics = {
    'response_length': MetricsCalculator.response_length,
    'completeness': lambda x: MetricsCalculator.contains_key_elements(
        x, ['analysis', 'recommendation', 'conclusion']
    )
}

# Initialize enhanced evaluator
evaluator = EnhancedAgentEvaluator(
    test_cases=test_cases,
    models=models_to_test,
    metrics=custom_metrics
)

# Run evaluation
evaluator.run_evaluation()

python
import os

# Configure your custom endpoint before running evaluation
os.environ["OPENAI_API_BASE"] = "<http://127.0.0.1:your_port>"
os.environ["OPENAI_API_KEY"] = "SS"

# Add your custom model to the test
models_to_test = [
    "your-model-name",  # Your custom model
    "openai:gpt-4",     # For comparison
    # Add other models as needed
]

marvin.plan()

python
import os
import marvin
from openai import AsyncOpenAI
import asyncio

# Method 1: Using environment variables (Most reliable)
os.environ["OPENAI_API_BASE"] = "<http://your-endpoint>"  # Your actual endpoint
os.environ["OPENAI_API_KEY"] = "EMPTY"

# Optional: If your endpoint uses a specific model name
os.environ["MARVIN_MODEL_NAME"] = "default"  # Your model name

async def main():
    try:
        steps = await marvin.plan("Design a launch strategy for a mobile app.")
        print(steps)
    except Exception as e:
        print(f"Error occurred: {str(e)}")
        # Print more details about the error
        print(f"Error type: {type(e)}")

asyncio.run(main())

python
import os
import marvin
from openai import AsyncOpenAI
import asyncio
import httpx

# Configure the base settings
os.environ["OPENAI_API_BASE"] = "<http://your-endpoint>"
os.environ["OPENAI_API_KEY"] = "EMPTY"

# Optional: Disable SSL verification if needed
os.environ["OPENAI_VERIFY_SSL"] = "false"

async def test_connection():
    """Test the connection to the endpoint"""
    client = AsyncOpenAI(
        base_url=os.environ["OPENAI_API_BASE"],
        api_key=os.environ["OPENAI_API_KEY"],
        timeout=httpx.Timeout(30.0)  # Increase timeout if needed
    )
    
    try:
        # Test the connection with a simple completion
        response = await client.chat.completions.create(
            model="default",  # Your model name
            messages=[{"role": "user", "content": "Hello"}]
        )
        print("Connection test successful!")
        return True
    except Exception as e:
        print(f"Connection test failed: {str(e)}")
        return False

async def main():
    # First test the connection
    connection_ok = await test_connection()
    
    if connection_ok:
        try:
            steps = await marvin.plan("Design a launch strategy for a mobile app.")
            print(steps)
        except Exception as e:
            print(f"Error in marvin.plan(): {str(e)}")
    else:
        print("Skipping marvin.plan() due to connection test failure")

if __name__ == "__main__":
    asyncio.run(main())

marvin.plan()

python
import os
import marvin
from openai import AsyncOpenAI
import asyncio
import httpx
from marvin import settings

# More comprehensive environment setup
os.environ.update({
    "OPENAI_API_BASE": "<http://your-endpoint>",
    "OPENAI_API_KEY": "EMPTY",
    "MARVIN_OPENAI_API_BASE": "<http://your-endpoint>",  # Try adding this
    "MARVIN_OPENAI_API_KEY": "EMPTY",                  # And this
    "OPENAI_VERIFY_SSL": "false",                      # If needed
})

# Try to directly configure Marvin's settings
settings.openai.api_base = "<http://your-endpoint>"
settings.openai.api_key = "EMPTY"

async def test_connection():
    """Test the connection to the endpoint"""
    client = AsyncOpenAI(
        base_url=os.environ["OPENAI_API_BASE"],
        api_key=os.environ["OPENAI_API_KEY"],
        timeout=httpx.Timeout(30.0),
        http_client=httpx.AsyncClient(verify=False)  # If needed for SSL
    )
    
    try:
        response = await client.chat.completions.create(
            model="default",
            messages=[{"role": "user", "content": "Hello"}]
        )
        print("Connection test successful!")
        print(f"Model responded with: {response.choices[0].message.content}")
        return True
    except Exception as e:
        print(f"Connection test failed: {str(e)}")
        return False

async def custom_plan():
    """Alternative to marvin.plan() using direct API calls"""
    client = AsyncOpenAI(
        base_url=os.environ["OPENAI_API_BASE"],
        api_key=os.environ["OPENAI_API_KEY"],
        timeout=httpx.Timeout(30.0),
        http_client=httpx.AsyncClient(verify=False)  # If needed for SSL
    )
    
    messages = [
        {"role": "system", "content": "You are a helpful AI assistant that creates detailed plans."},
        {"role": "user", "content": "Design a launch strategy for a mobile app."}
    ]
    
    try:
        response = await client.chat.completions.create(
            model="default",
            messages=messages,
            temperature=0.7,
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"Custom plan failed: {str(e)}")
        return None

async def main():
    # First test the connection
    connection_ok = await test_connection()
    
    if connection_ok:
        print("\nTrying marvin.plan()...")
        try:
            steps = await marvin.plan("Design a launch strategy for a mobile app.")
            print("Marvin plan succeeded:")
            print(steps)
        except Exception as e:
            print(f"Error in marvin.plan(): {str(e)}")
            
            print("\nTrying custom plan implementation...")
            custom_steps = await custom_plan()
            if custom_steps:
                print("Custom plan succeeded:")
                print(custom_steps)

if __name__ == "__main__":
    asyncio.run(main())

marvin.plan()

python
import logging

# Add this before running the main code
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("marvin")
logger.setLevel(logging.DEBUG)

marvin.plan()

marvin.plan()

marvin.plan()

python
import marvin
from marvin import Agent
import asyncio

# Create a planning-specific agent
planagent = Agent(
    instructions="""You are an expert planning assistant. When creating plans:
    1. Always start with a root task (task 0)
    2. Break down complex tasks into clear subtasks
    3. Ensure each subtask has a clear parent task
    4. Provide specific, actionable items
    """,
    model="your-model-name"  # Your specific model
)

# Configure the environment
import os
os.environ["OPENAI_API_BASE"] = "your-endpoint"
os.environ["OPENAI_API_KEY"] = "your-key"

async def create_plan(question: str):
    try:
        # Create initial root task
        root_task = {
            "id": 0,
            "description": "Main objective",
            "parent_id": None
        }
        
        # Use marvin.plan() with the configured agent
        plan = await marvin.plan(
            instructions=question,
            agent=planagent,
            initial_tasks=[root_task]  # Provide the root task
        )
        return plan
    except Exception as e:
        print(f"Error creating plan: {str(e)}")
        return None

# Run the planning
async def main():
    question = "Design a launch strategy for a mobile app."
    try:
        result = await create_plan(question)
        if result:
            print("Plan created successfully:")
            print(result)
    except Exception as e:
        print(f"Error in main: {str(e)}")

if __name__ == "__main__":
    asyncio.run(main())

python
import marvin
from marvin import Agent
import asyncio

class PlanningAgent(Agent):
    def __init__(self, model_name: str):
        super().__init__(
            instructions="""You are an expert planning assistant. Create detailed, 
            hierarchical plans with clear parent-child relationships between tasks.""",
            model=model_name
        )
    
    async def create_hierarchical_plan(self, question: str):
        # Create a planning prompt that ensures proper task hierarchy
        planning_prompt = f"""
        Create a detailed plan for: {question}
        
        Format your response as a hierarchical list where:
        - Each task has a unique ID
        - Each subtask references its parent task ID
        - Start with task ID 0 as the main objective
        
        Example format:
        0. Main objective
            1. First major task (parent: 0)
                1.1 Subtask (parent: 1)
                1.2 Subtask (parent: 1)
            2. Second major task (parent: 0)
                2.1 Subtask (parent: 2)
        """
        
        try:
            response = await self.run(planning_prompt)
            return response
        except Exception as e:
            print(f"Error in planning: {str(e)}")
            return None

async def main():
    # Configure the environment
    os.environ["OPENAI_API_BASE"] = "your-endpoint"
    os.environ["OPENAI_API_KEY"] = "your-key"
    
    # Create the planning agent
    planner = PlanningAgent(model_name="your-model-name")
    
    # Test the planning
    question = "Design a launch strategy for a mobile app."
    result = await planner.create_hierarchical_plan(question)
    
    if result:
        print("Generated Plan:")
        print(result)

if __name__ == "__main__":
    asyncio.run(main())

marvin.plan()

python
import marvin
from marvin import Agent
from openai import AsyncOpenAI
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
import asyncio

# Your client setup
client = AsyncOpenAI(
    base_url="your-endpoint",
    api_key="your-key"
)

# Create the planning agent with structured task handling
planagent = marvin.Agent(
    model=OpenAIModel(
        model_name="qwen2.5vl",
        provider=OpenAIProvider(openai_client=client),
    ),
    name="Planner",
    instructions="""You are a planning assistant that creates structured task lists.
    When creating plans:
    1. Always start with a root task (task 0)
    2. Each subsequent task must have a clear parent task
    3. Break down complex tasks into manageable subtasks
    4. Provide specific, actionable instructions
    
    Format your responses as:
    0. Root Task: [Main Objective]
        1. Task: [Description] (Parent: 0)
            1.1 Subtask: [Description] (Parent: 1)
            1.2 Subtask: [Description] (Parent: 1)
        2. Task: [Description] (Parent: 0)
    """,
)

async def create_structured_plan(question: str, file_info: str):
    # Create the initial root task
    root_task = {
        "id": 0,
        "description": f"Plan for: {question}",
        "parent_id": None
    }
    
    try:
        # Combine question and file info
        full_prompt = f"{question}\nContext: {file_info}"
        
        # Create the plan with explicit root task
        plan = await marvin.plan(
            instructions=full_prompt,
            agent=planagent,
            initial_tasks=[root_task]
        )
        return plan
    except Exception as e:
        print(f"Error creating plan: {str(e)}")
        return None

if __name__ == "__main__":
    all_logs = []
    
    async def process_question():
        with marvin.Thread() as thread:
            index = 160  # Change this to loop if needed
            print(dataset[index])
            question = dataset[index]["Question"]
            fileinfo = dataset[index]["file_path"]
            finalanswer = dataset[index]["Final answer"]
            
            try:
                # Create the plan
                plan = await create_structured_plan(question, fileinfo)
                
                if plan:
                    print("\nGenerated Plan:")
                    for i, task in enumerate(plan):
                        print(f"{i}. {task.instructions}")
                        if hasattr(task, 'subtasks') and task.subtasks:
                            for j, subtask in enumerate(task.subtasks, 1):
                                print(f"   {i}.{j} {subtask.instructions}")
            except Exception as e:
                print(f"Error processing question: {str(e)}")
    
    # Run the async function
    asyncio.run(process_question())