Sessions

# First call
result1 = agent.run_sync('My name is Pablo')
# "Nice to meet you, Pablo!"

# Second call (new context, no memory)
result2 = agent.run_sync("What's my name?")
# "I don't know your name. Could you tell me?"

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    instructions='Be helpful. Remember what the user tells you.'
)

# First message
result1 = agent.run_sync('My name is Pablo and I work in fintech.')
print(f"Agent: {result1.output}")

# Continue with history
result2 = agent.run_sync(
    'What do I do for work?',
    message_history=result1.new_messages()  # Pass previous messages
)
print(f"Agent: {result2.output}")
# Agent: You work in fintech!

from pydantic_ai import Agent, ModelMessage, ModelMessagesTypeAdapter
from pydantic_core import to_json
from pathlib import Path

class ChatSession:
    def __init__(self, agent: Agent, session_id: str):
        self.agent = agent
        self.session_id = session_id
        self.messages: list[ModelMessage] = []

    def chat(self, user_message: str) -> str:
        """Send a message and get response, maintaining history."""
        result = self.agent.run_sync(
            user_message,
            message_history=self.messages
        )
        # Update history with all messages
        self.messages = result.all_messages()
        return result.output

    def save(self, filepath: str):
        """Save session to JSON file."""
        data = to_json(self.messages)
        Path(filepath).write_bytes(data)

    def load(self, filepath: str):
        """Load session from JSON file."""
        data = Path(filepath).read_bytes()
        self.messages = ModelMessagesTypeAdapter.validate_json(data)

    def clear(self):
        """Clear conversation history."""
        self.messages = []

# Usage
agent = Agent('openai:gpt-4o')
session = ChatSession(agent, session_id='pablo_001')

# Have a conversation
print(session.chat("Hi, I'm Pablo"))
print(session.chat("What's my name?"))  # Will remember

# Save for later
session.save('/tmp/pablo_session.json')

# Later, load and continue
session.load('/tmp/pablo_session.json')
print(session.chat("Do you still remember me?"))  # Yes!

from dataclasses import dataclass, field
from pydantic_ai import Agent, RunContext

@dataclass
class UserState:
    name: str | None = None
    preferences: dict = field(default_factory=dict)

agent = Agent(
    'openai:gpt-4o',
    deps_type=UserState,
    instructions='You are a helpful assistant. Use tools to remember user information.'
)

@agent.tool
def save_name(ctx: RunContext[UserState], name: str) -> str:
    """Remember the user's name."""
    ctx.deps.name = name
    return f"Got it! I will remember your name is {name}"

@agent.tool
def save_preference(ctx: RunContext[UserState], key: str, value: str) -> str:
    """Save a user preference."""
    ctx.deps.preferences[key] = value
    return f"Saved: {key} = {value}"

@agent.tool
def get_info(ctx: RunContext[UserState]) -> dict:
    """Get all saved user information."""
    return {'name': ctx.deps.name, 'preferences': ctx.deps.preferences}

# Usage
state = UserState()
result1 = agent.run_sync("My name is Pablo and I prefer dark mode", deps=state)
result2 = agent.run_sync("What do you know about me?", deps=state, message_history=result1.new_messages())

# State persists outside the agent
print(f"State: {state}")
# UserState(name='Pablo', preferences={'theme': 'dark'})

from pydantic_ai import Agent, ModelMessage

def keep_recent(messages: list[ModelMessage], keep: int = 10) -> list[ModelMessage]:
    """Keep only the N most recent messages."""
    if len(messages) <= keep:
        return messages
    return messages[-keep:]

# Apply before each run
history = keep_recent(all_messages, keep=10)
result = agent.run_sync(message, message_history=history)