Get Started with Pydantic AI

pip install pydantic-ai openai 

pip install 'pydantic-ai[logfire]'  # For debugging and monitoring

CEREBRAS_API_KEY=your-cerebras-api-key-here

export CEREBRAS_API_KEY=your-cerebras-api-key-here

import os
import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel

# Create a Cerebras model instance
model = OpenAIChatModel(
    'llama-3.3-70b',
    provider='cerebras'
)

# Create an agent with the Cerebras model
agent = Agent(
    model=model,
    system_prompt='You are a helpful assistant that provides clear, concise answers.'
)

async def main():
    # Run the agent asynchronously
    result = await agent.run('What is the capital of France?')
    print(result)

asyncio.run(main())

import os
import asyncio
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel

# Define your output structure with validation
class CityInfo(BaseModel):
    city: str = Field(description="The name of the city")
    country: str = Field(description="The country where the city is located")
    population: int = Field(description="Approximate population")
    famous_for: list[str] = Field(description="List of things the city is famous for")

# Create the Cerebras model
model = OpenAIChatModel(
    'llama-3.3-70b',
    provider='cerebras'
)

# Create an agent with structured output
agent = Agent(
    model=model,
    output_type=CityInfo,  # Changed from result_type to output_type
    system_prompt='Extract city information from the user query. Provide accurate, factual data.'
)

# Get structured, validated output
async def get_city_info():
    result = await agent.run('Tell me about Paris')
    print(result)

# Run the async function
import asyncio
asyncio.run(get_city_info())

import os
import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel

# Create the Cerebras model
model = OpenAIChatModel(
    'llama-3.3-70b',
    provider='cerebras'
)

agent = Agent(
    model=model,
    system_prompt='You are a helpful assistant.'
)

async def run_concurrent_queries():
    results = await asyncio.gather(
        agent.run('What is 2+2?'),
        agent.run('What is the capital of Spain?'),
        agent.run('Who wrote Romeo and Juliet?')
    )

    for i, result in enumerate(results, 1):
        print(f"Answer {i}: {result.output}")

# Run the async function
asyncio.run(run_concurrent_queries())

import os
import asyncio
from datetime import datetime
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIChatModel

model = OpenAIChatModel(
    'llama-3.3-70b',
    provider='cerebras',
)

agent = Agent(
    model=model,
    system_prompt='You are a helpful assistant with access to tools.'
)

@agent.tool
def get_current_time(ctx: RunContext[None]) -> str:
    """Get the current time in a readable format."""
    return datetime.now().strftime('%I:%M %p')

@agent.tool
def calculate(ctx: RunContext[None], expression: str) -> str:
    """Safely evaluate a mathematical expression.
    
    Args:
        expression: A mathematical expression like '15 * 23'
    """
    try:
        # In production, use a safe eval library like numexpr
        result = eval(expression, {'__builtins__': {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

# The agent can now use these tools automatically
async def test_tools():
    result = await agent.run('What time is it and what is 15 * 23?')
    print(result)

# Run the async function
asyncio.run(test_tools())

import os
import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel

model = OpenAIChatModel(
    'llama-3.3-70b',
    provider='cerebras'
)

agent = Agent(
    model=model,
    system_prompt='You are a creative writing assistant.'
)

# Stream the response token by token
async def stream_response():
    async with agent.run_stream('Write a short poem about AI') as response:
        async for chunk in response.stream_text():
            print(chunk, end='', flush=True)
        print()  # New line at the end

# Run the async function
asyncio.run(stream_response())