AI Agents for E-commerce with Pydantic AI

E-commerce businesses face incredibly complex challenges to provide an exceptional customer service: customers expect instant, accurate responses 24/7, detailed product information at their fingertips, and efficient order management at scale. Meeting these expectations has traditionally forced businesses to choose between aggressive hiring (expensive) or service compromises (risky).

But what if there was a better way?

The rise of Large Language Models (LLMs) has opened new possibilities for automating complex customer interactions. Consider these real-world scenarios:

A customer asking detailed questions about product materials and care instructions at 3 AM.
Someone needing to update their shipping address on a pending order.
A request for return processing with specific item conditions.

Previously, each of these required human intervention. Now, they can be handled automatically, accurately, and instantly by AI agents. Modern AI agents can understand context, process natural language and execute complex actions. However, there's the catch: given the probabilistic nature of LLMs, building production-ready AI agents that you can actually trust has been challenging.

#Pydantic AI: A Game Changer for AI Agents

Pydantic is my favourite Python library and I use it in practically every Python project I build to handle data parsing and validation, from FastAPI backends to data processing pipelines. And if you are a Python developer, you probably have used it too.

Pydantic has already proven to be essential for LLM applications that need structured outputs or function calling capabilities. It's no coincidence that libraries like Instructor, LangChain, and LlamaIndex, as well as OpenAI's and Anthropic's own SDKs, use it internally. Or to quote Jason Liu's influential talk: "Pydantic is all you need."

Two weeks ago, Pydantic creator Samuel Colvin and his team released PydanticAI, a game-changing library for building AI agents in Python.

What sets PydanticAI apart? Its simplicity and clean architecture. While other frameworks pile on abstractions, PydanticAI provides a thin, type-safe layer for building AI agents. If you've used FastAPI, you'll feel right at home — the library embraces Python's native features and builds on familiar tools, making it both powerful and approachable.

Out of the box, PydanticAI delivers a set of features that are essential for production generative AI applications: type safety, definition and validation of structured outputs, streaming support, a dependency injection system, and model flexibility (supporting different model providers like OpenAI, Anthropic, Gemini, Ollama, Groq and Mistral). Make sure to check the PydanticAI documentation for more details.

In the following two sections, we'll go from theory to practice and learn how to use the PydanticAI library to build two AI agents for a fashion e-commerce store:

A support agent that assists customers by answering their questions using Retrieval-Augmented Generation (RAG).
An order management agent that processes complex operations like updates, returns and cancellations.

You can find all the code in detail in the following GitHub repository.

While these examples are slightly simplified to serve as an introduction to agent building, they mirror the patterns and approaches you'd use in production systems. You will see how PydanticAI helps you build AI agents that work reliably and solve real business problems. Ready to dive in?

#Customer Support RAG Agent

Let's start with a scenario every e-commerce business knows well: customers asking detailed questions about products, shipping policies and store information at all hours. Even if the questions aren't particularly complex, they require accurate, consistent answers drawn from your product catalog and company policies.

Retrieval-Augmented Generation (RAG) is a powerful technique to solve this problem. It combines the capabilities of a Large Language Model (LLM) with your own knowledge base to provide accurate responses to user questions.

The core idea is simple:

We give our support agent access to a knowledge base containing all the relevant information about products, prices and company policies.
When a customer asks a question, the agent will search the knowledge base and deliver an accurate response based on this information, in a natural and conversational manner.

Let's see how to implement it using PydanticAI. We'll build a customer support agent for Maison Noir, a luxury fashion e-commerce store that we will use as a fictional example for this blog post.

The core of any PydanticAI application is the Agent class, which can be configured with a specific model, dependencies and a system prompt:

from pydantic_ai import Agent

SYSTEM_PROMPT = """\
You are a helpful and knowledgeable assistant for the luxury fashion store Maison Noir.
Your role is to provide detailed information and assistance about the store and its products.

Follow these guidelines:
- ALWAYS search the knowledge base using the search_knowledge_base tool to answer user questions.
- Provide accurate product and policy information based ONLY on the information retrieved from the knowledge base. Never make assumptions or provide information not present in the knowledge base.
- Structure your responses in a clear, concise and professional manner, maintaining our premium brand standards
- Highlight unique features, materials, and care instructions when relevant.
- If information is not found in the knowledge base, politely acknowledge this.
"""

rag_agent = Agent(
    model='openai:gpt-4o',
    deps_type=KnowledgeBase,
    system_prompt=SYSTEM_PROMPT
)

PydanticAI uses a clean dependency injection system to provide data and services to your agents. In our case, we're providing a custom KnowledgeBase class as a dependency, so that we are able to search it from the agent's tools.

In RAG applications, you would normally preprocess, chunk, embed and store your documents in a vector database, and ideally combine vector search with full-text search for maximum accuracy. We have explored these ideas in previous blog posts where we build RAG from scratch, and use Redis or PostgreSQL as vector databases for RAG.

For this example, we're using a extremely simplified knowledge base that just returns our store's information:

class KnowledgeBase:
    def search(self, search_query: str) -> str:
        """Search method that would typically implement vector similarity search, full-text search, or a hybrid combination of both."""
        return KB_INFORMATION

The beauty of this design is that in a real application, you would only need to change the implementation of the search method to use your preferred database and search strategy. The rest of the agent code remains exactly the same, completely independent of how the knowledge retrieval is implemented.

Now we need to give our agent the ability to query the knowledge base. PydanticAI makes this straightforward with its tool system. Tools are simply functions that the agent can call when needed. We can easily define a tool using the @agent.tool decorator:

@rag_agent.tool
async def search_knowledge_base(ctx: RunContext[KnowledgeBase], search_query: str) -> str:
    """Search the knowledge base to retrieve information about Maison Noir, the store and its products"""
    return ctx.deps.search(search_query)

This is where the use of Pydantic truly shines. When we create a tool, PydanticAI automatically extracts parameters from the function signature, extracts the function description from the docstring and creates a tool schema that the LLM can understand. All of this happens automatically behind the scenes, making our code simpler and cleaner.

Notice also how the tool's first argument gives access to the agent's context, which includes the agent's dependencies (in our case, the knowledge base that we want to search).

Now our agent is ready to handle customer queries. We can simply run it like this, remembering to pass the knowledge base as a dependency:

kb = KnowledgeBase()
user_message = 'How long does international shipping take?'
result = await rag_agent.run(user_message, deps=kb)
print(result.data)

# International shipping durations from Maison Noir are as follows:
# - **Europe**: 3-5 business days
# - **North America**: 5-7 business days
# - **Asia**: 6-8 business days
# - **Rest of the World**: 7-10 business days

You can also explore the new messages generated during the agent run using result.new_messages() and you will get a result like this:

[
    ModelRequest(
        parts=[
            SystemPromptPart(
                content='You are a helpful and knowledgeable assistant...',
                part_kind='system-prompt'
            ),
            UserPromptPart(
                content='How long does international shipping take?',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt'
            )
        ],
        kind='request'
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='search_knowledge_base',
                args=ArgsJson(args_json='{"search_query":"international shipping duration"}'),
                tool_call_id='call_Pu3t2gnUiOvqMtxnnEixa5Dq',
                part_kind='tool-call'
            )
        ],
        timestamp=datetime.datetime(...),
        kind='response'
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='search_knowledge_base',
                content='# MAISON NOIR - Premium Fashion House\n\n...',
                tool_call_id='call_Pu3t2gnUiOvqMtxnnEixa5Dq',
                timestamp=datetime.datetime(...),
                part_kind='tool-return'
            )
        ],
        kind='request'
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='International shipping durations from Maison Noir are as follows:...',
                part_kind='text'
            )
        ],
        timestamp=datetime.datetime(...),
        kind='response'
    )
]

As you can see above, the agent internally calls the search_knowledge_base tool to query the knowledge base and then builds the response using the retrieved information.

While querying the agent with single questions is useful, most e-commerce applications need to support full conversations with customers. PydanticAI makes this straightforward by allowing us to maintain conversation history. We just need to run the agent in a loop and make sure to update the message history with the new messages generated in every agent call:

async def run_agent():
    kb = KnowledgeBase()
    messages = []
    while True:
		user_message = input()
        result = await rag_agent.run(user_message, message_history=messages, deps=kb)
        messages += result.new_messages()
		print(result.data)

For the best possible user experience, PydanticAI supports response streaming. Instead of waiting for complete responses, customers see them being generated in real-time, creating a more natural and engaging interaction. Implementing streaming is straightforward:

async def run_agent_streaming():
    kb = KnowledgeBase()
    messages = []
    while True:
		result_content = ''
        user_message = input()
        async with rag_agent.run_stream(user_message, message_history=messages, deps=kb) as result:
            async for chunk in result.stream_text(delta=True):
		        result_content += chunk
                print(chunk, end = '')
        messages += result.new_messages() + [
            ModelResponse.from_text(content=result_content, timestamp=result.timestamp())
        ]

⚠️The final result message will NOT be added to result messages if you use stream_text(delta=True), so we need to add it manually.

You can try the agent yourself using the code in the GitHub repository. Simply clone the repository, install the dependencies and run:

python rag.py

#Orders Management Agent

Now that we've introduced the main concepts and components of PydanticAI through our RAG agent, let's tackle something more ambitious: order management. This agent will handle a variety of operations including:

Checking order status and details.
Updating shipping addresses.
Processing order cancellations.
Handling return requests.
Escalating complex cases to human support.

Here is where things get interesting. We are moving beyond simple question answering into actually executing actions that affect real orders and customers. Every operation requires careful validation, state management and communication. Let's see how PydanticAI helps us build a robust system that handles these complexities elegantly.

First of all, we need to create a set of enums and Pydantic models for our order management system:

from pydantic import BaseModel
from datetime import datetime
from enum import Enum

class OrderStatus(str, Enum):
    PENDING = 'pending'
    CONFIRMED = 'confirmed'
    SHIPPED = 'shipped'
    DELIVERED = 'delivered'
    CANCELLED = 'cancelled'

class ReturnReason(str, Enum):
    WRONG_SIZE = 'wrong_size'
    WRONG_COLOR = 'wrong_color'
    NOT_AS_DESCRIBED = 'not_as_described'
    CHANGED_MIND = 'changed_mind'
    DAMAGED = 'damaged'

class EscalationReason(str, Enum):
    COMPLEX_REQUEST = 'complex_request'
    CUSTOMER_DISSATISFIED = 'customer_dissatisfied'
    CUSTOMER_REQUEST = 'customer_request'
    CANNOT_RESOLVE_SITUATION = 'cannot_resolve_situation'

class OrderItem(BaseModel):
    product_code: str
    name: str
    size: str
    color: str
    quantity: int
    price: float

class Address(BaseModel):
    street: str
    city: str
    postal_code: str
    country: str

class Order(BaseModel):
    order_id: str
    status: OrderStatus
    items: list[OrderItem]
    total_amount: float
    created_at: datetime
    shipping_address: Address
    
    @property
    def can_modify(self) -> bool:
        return self.status in [OrderStatus.PENDING, OrderStatus.CONFIRMED]
    
    @property
    def can_return(self) -> bool:
        if self.status != OrderStatus.DELIVERED:
            return False
        return (datetime.now() - self.created_at).days <= 30

Notice how as well as defining data structures, we are encoding business rules directly into our Pydantic models that will help during validation. The can_modify and can_return properties ensure that orders are only modified in appropriate states and that returns are only allowed within 30 days of delivery.

Let's now create the orders agent:

from dataclasses import dataclass
from pydantic_ai import Agent

SYSTEM_PROMPT = """\
You are an AI assistant for for the luxury fashion store Maison Noir.
Your role is to help clients with orders, returns, and related inquiries, using the provided tools.

Key Guidelines:
- Orders can only be modified or canceled when PENDING or CONFIRMED.
- Address changes require a valid new address with all fields.
- There is a 30-day return window for delivered items. Items must be in original condition with tags.
- Always maintain a professional, refined communication tone. Be concise in your answers and precise with details.

Escalate to a human for:
- Complex modifications or special requests.
- Dissatisfied customers.
- When the user specifically requests to talk with a human.
- Any situation you cannot fully resolve.
"""

@dataclass
class Deps:
    db: Database

orders_agent = Agent(
    model='openai:gpt-4o',
    deps_type=Deps,
    system_prompt=SYSTEM_PROMPT
)

We are now using a dataclass for dependencies instead of passing the database directly. While not strictly necessary here, this pattern is more convenient when you need to inject multiple dependencies into your agent.

The orders agent will run in a simple loop that maintains conversation history and handles streaming responses, very similar to the run_agent_streaming function we saw previously, so there is no need to repeat it here too.

You can try it yourself by cloning the GitHub repository, installing the dependencies and running:

python orders.py

The most interesting part of the orders agent is the toolkit that it has available to perform all the necessary functionalities. Each tool represents a specific order management operation, complete with its own validation and business logic. Let's take a look at each one in detail.

#Order Status & Details Retrieval

Our first tool handles order status and details inquiries:

@orders_agent.tool
async def get_order_details(ctx: RunContext[Deps], order_id: str) -> str:
    """Get the current status and details of an order."""
    order = ctx.deps.db.orders.get(order_id)
    if not order:
        return 'Order not found'
    
    items = '\n'.join(
        f"- {item.name} ({item.size}, {item.color}) €{item.price}"
        for item in order.items
    )
    address = order.shipping_address

    return (
        f'Status: {order.status.value}\n'
        f'Order date: {order.created_at.strftime('%Y-%m-%d %H:%M')}\n'
        f'Items:\n{items}\n'
        f'Total: €{order.total_amount}\n'
        f'Shipping to: {address.street}, {address.city}, {address.postal_code} {address.country}\n'
    )

The tool takes an order_id parameter and uses the database dependency to fetch the corresponding order, which is automatically parsed into our Order Pydantic model. It then formats the order details into a clear, readable response. The agent can therefore use this tool to answer any questions about a specific order.

Let's see it in action:

USER: What's the status of my order 001?
AGENT: Your order (ID: 001) is currently confirmed. Here are the details:

Order Date: December 19, 2024

Item: Classic Noir Biker Jacket (M, Black)

Total Amount: €1499.0

Shipping Address: 123 Avenue des Champs-Élysées, Paris, 75008, France

#Shipping Address Updates

Our next tool handles shipping address updates, moving beyond retrieval to actual action execution:

@orders_agent.tool
async def update_shipping_address(
    ctx: RunContext[Deps],
    order_id: str,
    street: str,
    city: str,
    postal_code: str,
    country: str
) -> str:
    """Update the shipping address for an order if possible."""
    order = ctx.deps.db.orders.get(order_id)
    if not order:
        return 'Order not found'
    
    if not order.can_modify:
        return f'Cannot modify order - current status is {order.status.value}'
    
    new_address = Address(street=street, city=city, postal_code=postal_code, country=country)
    
    if ctx.deps.db.update_shipping_address(order_id, new_address):
        return f'Successfully updated shipping address'
    return 'Failed to update shipping address'

Notice how the tool performs all necessary validations, verifying that the order exists and that it can be modified. Then it creates a new Address object using our Pydantic model and uses the database dependency to update the address. Whether successful or not, it returns clear status messages that the agent can relay to the customer. Let's see some example conversations:

USER: I would like to update the shipping address for the order 002.
AGENT: Unfortunately, I am unable to update the shipping address for order 002, as it has already been delivered. If you need further assistance, please let me know.

USER: I would like to update the shipping address for the order 001.
AGENT: Could you please provide the new shipping address, including the street, city, postal code, and country?
USER: 14 Rue des Roses, 75008 Paris, France.
AGENT: The shipping address for your order 001 has been successfully updated to 14 Rue des Roses, 75008 Paris, France. If you need further assistance, feel free to ask.

#Handling Returns & Cancellations

One of the most challenging aspects of order management is handling returns and cancellations. Let's see how we can implement these tools:

@orders_agent.tool
async def cancel_order(ctx: RunContext[Deps], order_id: str) -> str:
    """Cancel an order if possible."""
    order = ctx.deps.db.orders.get(order_id)
    if not order:
        return 'Order not found'
    
    if not order.can_modify:
        return f'Cannot modify order - current status is {order.status.value}'

    if ctx.deps.db.update_order_status(order_id, OrderStatus.CANCELLED):
        return f'Successfully cancelled order'
    return 'Failed to cancel order'

@orders_agent.tool
async def request_return(
    ctx: RunContext[Deps],
    order_id: str,
    reason: ReturnReason
) -> str:
    """Process a return request for an order."""
    order = ctx.deps.db.orders.get(order_id)
    if not order:
        return 'Order not found'
    
    if not order.can_return:
        if order.status != OrderStatus.DELIVERED:
            return f'Cannot return order - current status is {order.status.value}'
        return 'Cannot return order - outside our 30-day return window.'

    return_id = 'RET-' + uuid4().hex[:12]
    
    return (
        f"Return request approved for order {order_id}:\n"
        f"Reason: {reason.value}\n\n"
        f"A return label with ID {return_id} has been emailed to you. Please ship items within 14 days with all original tags attached."
    )

PydanticAI makes it quite simple to handle the complexity of these operations elegantly and concisely. The agent automatically knows:

Which parameters each operation requires (like return reasons and order IDs).
When each tool is appropriate to use (based on order status and timing).
How to handle error conditions (invalid states, expired windows).
How to communicate results back to the customer (from success messages to explanations of why an operation can't proceed).

Let's see an example of a return request:

USER: My order is damaged and I would like to get a refund.
AGENT: I can assist you with processing a return for your damaged order. Could you please provide me with your order ID so I can help you further?
USER: 002.
AGENT: Your return request for the damaged order has been approved. You will receive a return label via email with ID RET-eedc4cdf4494. Please ship the item back within 14 days, ensuring it includes all original tags. Once we've received the item and verified its condition, your refund will be processed. If you need further assistance, please let me know.

#Human Escalation

Finally, a critical feature of any automated system is knowing when to hand over to human support. Our escalation tool will transfer conversations to a human during complex requests or unresolved issues, when the user specifically demands it, or when the agent detects customer dissatisfaction. The tool is also able to flag high-priority cases, ensuring urgent matters or clearly dissatisfied customers receive immediate attention.

Here's the implementation:

@orders_agent.tool
async def escalate_to_human(
    ctx: RunContext[Deps],
    reason: EscalationReason,
    details: str,
    high_priority: bool = False
) -> str:
    """Escalate the conversation to a human.
    Set high_priority=True for urgent matters or when customer is clearly dissatisfied.
    """
    response_time = '1 hour' if high_priority else '12 hours'
    return f'This matter has been escalated to our support team. We will contact you within {response_time}.'

This implementation is simplified for demonstration purposes. A production system would need to integrate with your support desk software, trigger notifications, and possibly create support tickets with appropriate priority levels and routing. Those are additional details to add inside the tool, but the fundamental agent logic would remain the same.

Let's take a look at an example where the agent detects user frustration:

USER: I would like to cancel my order 002.
AGENT: Your order with ID 002 has already been delivered and cannot be canceled. However, if you wish to return the item, please let me know, and I can assist you with the return process.
USER: You are not being helpful.
AGENT: I've escalated your concern to our support team for further assistance. You will be contacted within 1 hour. Thank you for your patience.

Behind the scenes, the agent generates a detailed tool call that provides the support team with valuable context:

ToolCallPart(
    tool_name='escalate_to_human',
    args=ArgsJson(
        args_json='{
		        "reason": "customer_dissatisfied",
		        "details": "Customer is not satisfied with the inability to cancel a delivered order and is seeking further assistance.",
		        "high_priority": true
		    }'
    ),
    tool_call_id='call_Lw115FnRP5h8UIgvvJkbdaIK',
    part_kind='tool-call'
)

#Beyond the Demo: Production Considerations

While our example agents illustrate the core concepts, deploying AI agents in production requires additional considerations:

Security, customer authentication and access control.
Integration with your existing order management systems, inventory and customer databases.
Agent performance monitoring and human feedback for continuous improvement.
Logging and audit trails.
Error handling and graceful degradation when external services are unavailable.
GDPR compliance and proper data privacy.

If you're interested in exploring how AI agents and automations could transform your e-commerce operations, I'd love to have a conversation about your specific needs and challenges. You can reach me at guillermo@codeawake.com.