Forem: Lukas Hamm

Building Production-Ready RAG in FastAPI with Vector Databases

Lukas Hamm — Mon, 05 Jan 2026 07:55:02 +0000

From Prompting to Production-Ready RAG

Retrieval-Augmented Generation (RAG) is often presented as a prompting technique or a lightweight runtime enhancement for LLMs. While this may work for demos, it breaks down quickly once you try to build a production-ready AI backend system with FastAPI..

The moment you want persistence, reproducibility, scalability, and clear separation of responsibilities, RAG inevitably leads to a vector database, because similarity-based retrieval cannot be treated as a stateless runtime concern.. Not as an optional optimization, but as the central infrastructure component that makes retrieval reliable and operational.

This article focuses exactly on that transition by integrating RAG into a FastAPI backend and treating the vector store as a first-class backend dependency that is configured, injected, and consumed like any other production system component.

RAG as a Backend Component

In current literature, Retrieval-Augmented Generation is not described as a single step or a simple pipeline, but as a composition of multiple responsibilities that together enable grounded generation.

At its core, RAG combines:

persistent knowledge storage,
similarity-based retrieval,
and context-aware generation.

While these responsibilities are often collapsed into a single conceptual flow, in a production backend they naturally separate into distinct backend concerns. Knowledge must first be ingested, transformed, embedded, and stored in a way that allows efficient semantic access. Retrieval then becomes a query-time operation that selects relevant information based on vector similarity. Only after this retrieval step does the generative model come into play.

Seen this way, RAG is not a monolithic process but an architectural pattern that explicitly connects storage, retrieval, and generation. The vector database forms the backbone of this pattern, acting as the system of record for knowledge and as the execution layer for retrieval. This perspective makes it clear why RAG belongs in the backend infrastructure and not inside the AI logic itself.

Scope of RAG in This Article

To keep the focus sharp, this article deliberately excludes:

PDF parsing and document ingestion pipelines
Chunking strategies
Embedding model comparisons
Advanced retrieval or ranking techniques

The goal here is not to explain how to generate embeddings, but how to integrate a vector-based RAG component cleanly into a FastAPI backend.

With the architectural role of RAG clarified, the next step is to materialize it as an actual backend dependency.

Introducing the Vector Store Dependency

At the center of the RAG setup sits the vector store. In this project, it is implemented using Qdrant via LangChain. The QdrantVectorStore used here is a LangChain-provided abstraction that encapsulates all communication with the underlying vector database. It is responsible for embedding queries, executing similarity searches, and mapping results back into document objects. For simplicity and clarity, this project relies on this existing LangChain implementation rather than introducing a custom database layer. By returning the QdrantVectorStore as a VectorStore dependency, the application stays decoupled from Qdrant-specific details while still leveraging a production-ready vector database.

def init_qdrant_vector_store(settings: Settings = Depends(get_settings)) -> VectorStore:
    """
    Initialize and return the vector store used for retrieval.
    """
    embeddings = get_openai_embeddings(
        settings.qdrant_vector_store.embedding_model,
        settings.openai_model.api_key
    )

    client = QdrantClient(path=settings.qdrant_vector_store.path)

    if not client.collection_exists(
        collection_name=settings.qdrant_vector_store.collection_name
    ):
        client.create_collection(
            collection_name=settings.qdrant_vector_store.collection_name,
            vectors_config=VectorParams(
                size=settings.qdrant_vector_store.vector_size,
                distance=settings.qdrant_vector_store.distance
            )
        )

    return QdrantVectorStore(
        client=client,
        collection_name=settings.qdrant_vector_store.collection_name,
        embedding=embeddings,
    )

This dependency is responsible for:

ensuring the vector collection exists,
configuring embeddings,
returning a ready-to-use vector store abstraction.

From the rest of the application’s perspective, this behaves exactly like a database connection.

Uploading Data into the Vector Store

Before retrieval can happen, the vector store must be populated. This is handled via a deliberately minimal upload endpoint /upload/chunks.

@router.post("/chunks", response_model=UploadResponse)
def upload_chunks(
    documents: DocumentChunks,
    vector_store: QdrantVectorStore = Depends(init_qdrant_vector_store)
):
    if len(documents.chunks) == 0:
        raise HTTPException(status_code=400, detail="No chunks to upload found")

    uuids = [str(uuid4()) for _ in range(len(documents.chunks))]

    chunk_ids_added = vector_store.add_documents(
        documents=documents.chunks,
        ids=uuids
    )

    if len(chunk_ids_added) == 0:
        raise HTTPException(
            status_code=500,
            detail="Uploading chunks to vector store failed"
        )

    return UploadResponse(
        success=True,
        message=f"{len(documents.chunks)} chunks uploaded"
    )

This endpoint assumes:

chunks are already prepared,
embeddings are generated implicitly via the vector store,
and data is persisted for future queries.

This reinforces the idea that RAG starts with data ingestion, not with prompting.

Querying with RAG via Dependency Injection

With data in place, the query endpoint becomes straightforward.

@router.post(path="/query", response_model=Insight)
def create_insight(
    request: InsightQuery,
    settings: Settings = Depends(get_settings),
    llm: BaseChatModel = Depends(init_openai_chat_model),
    vector_store: VectorStore = Depends(init_qdrant_vector_store)
):

Here, the vector store is injected alongside the LLM. Neither depends on the other. They are simply resources orchestrated by the endpoint.

The RAG chain then pulls context from the retriever and passes it into the prompt

The important point is not the retrieval logic itself, but where it lives:

Retrieval is fully owned by the vector store
The LLM only receives already-prepared context
No AI component depends on Qdrant, embeddings, or storage details

This clean separation is what allows RAG to exist as a backend component rather than bleeding into AI logic.

def run_rag_insight_chain(
    prompt_messages: ChatModelPrompt,
    llm: BaseChatModel,
    retriever: VectorStoreRetriever,
    question: str
) -> Insight:
    context = retriever.invoke(question)

    prompt_template = ChatPromptTemplate([
        ("system", prompt_messages.system),
        ("human", prompt_messages.human)
    ])

    parser = PydanticOutputParser(pydantic_object=Insight)

    chain = prompt_template | llm | parser

    return chain.invoke({
        "format_instruction": parser.get_format_instructions(),
        "question": question,
        "context": context
    })

The agent itself remains completely unaware of how the context was created — exactly as it should be.

Why This Architecture Matters

By modeling the vector database as a dependency:

RAG becomes configurable and replaceable
The AI layer stays clean and testable
Knowledge management is fully decoupled from generation

Once treated that way, it naturally fits into dependency injection, lifecycle management, and clean system boundaries.

In a production AI backend, RAG works through a vector store rather than through ad-hoc logic embedded in the AI layer. The vector store is the RAG system. Everything else simply consumes it.

Final Thoughts

RAG does not need to complicate your AI backend. When implemented via a vector store and injected like any other backend resource, it becomes predictable, scalable, and maintainable.

By separating ingestion, retrieval, and generation, you gain the freedom to evolve each part independently without turning your AI code into a tightly coupled system.

In the end, RAG is just another backend component. And treating it that way is exactly what makes it powerful.

💻 Code on GitHub: hamluk/fastapi-ai-backend/part-3

From LangChain Demos to a Production-Ready FastAPI Backend

Lukas Hamm — Fri, 02 Jan 2026 02:26:00 +0000

Why LangChain Needs a Proper Backend Architecture

Most LangChain examples stop where real backend work actually begins.

Many AI examples live in notebooks, scripts, or Streamlit demos, but they quickly break down once they need to run inside a production backend system. As soon as AI becomes part of an API, it must follow the same rules as any other backend component. Inputs and outputs must be well defined, dependencies need to be explicit, and the overall structure must allow change without rewriting everything.

This article addresses exactly this starting point.

We will establish a clean and maintainable FastAPI endpoint that integrates LangChain in a backend friendly way. The goal is to create a solid architectural foundation that can be extended step by step. At this stage, the implementation is intentionally kept simple. Later articles will gradually introduce more advanced LLM and agent capabilities on top of this baseline.

The focus here is not on showcasing LangChain features. Instead, it is about defining a clear and robust endpoint architecture that remains understandable, testable, and scalable as complexity increases.

Thinking of AI as a Backend Component

Before looking at code, it is important to align on how AI should be treated inside a backend system. The goal is not to expose an LLM directly, but to embed AI logic behind a stable and predictable API.

A backend ready AI endpoint should provide the following guarantees:

Clear request and response contracts
Explicit orchestration of dependencies
Encapsulation of AI logic away from HTTP concerns
Predictable outputs that can be validated and consumed by other systems

FastAPI fits naturally into this model because it already enforces structure through Pydantic models and dependency injection. This makes it possible to integrate LangChain without special cases or ad hoc glue code.

Defining the Contract with Pydantic

The first building block is a strict API contract. Input and output are defined explicitly using Pydantic models.

# RequestModel
class InsightQuery(BaseModel):
    question: str
    context: str

# ResponseModel
class Insight(BaseModel):
    title: str
    summary: str
    confidence: float

    @field_validator("confidence")
    @classmethod
    def clamp_confidence(cls, v):
        if v is None:
            return 0.0
        if v < 0:
            return 0.0
        if v > 1:
            return 1.0
        return float(v)

This contract ensures that the API remains predictable regardless of how the underlying AI logic evolves. The confidence validator also demonstrates an important principle. Even if AI produces imperfect values, the backend enforces consistency before returning a response. Without it, LLM output quickly becomes unpredictable and hard to integrate into real systems.

Injecting the LLM via FastAPI Depends

Instead of creating the LLM directly inside the endpoint or inside the chain, it is injected using FastAPI dependencies.

# FastAPI Endpoint definition
@router.post(path="/query", response_model=Insight)
def create_insight(
        request: InsightQuery,
        settings: Settings = Depends(get_settings),
        llm: BaseChatModel = Depends(init_openai_chat_model)
):

The language model itself is initialized in a separate dependency function.

def init_openai_chat_model(settings: Settings = Depends(get_settings)):
    """
    Initializes and returns the LangChain OpenAI chat model
    """
    return ChatOpenAI(
        model=settings.openai_model.model_name,
        temperature=settings.openai_model.temperature,
        api_key=settings.openai_model.api_key,
    )

This approach has several advantages. The endpoint stays focused on orchestration, configuration is centralized, and the LLM can be replaced or mocked easily during testing. From FastAPI’s perspective, the language model is just another dependency, no different from a database session or a service client.

Encapsulating the LangChain Logic

The LangChain logic itself is encapsulated in a dedicated function. The endpoint does not need to know how the chain is built or executed.

def run_insight_chain(prompt_messages: ChatModelPrompt, llm: BaseChatModel, question: str, context: str) -> Insight:
    """
    Builds and runs the LangChain insight chain
    """
    prompt_template = ChatPromptTemplate([
        ("system", prompt_messages.system),
        ("human", prompt_messages.human)
    ])

    parser = PydanticOutputParser(pydantic_object=Insight)

    chain = prompt_template | llm | parser

    response = chain.invoke({
        "format_instruction": parser.get_format_instructions(),
        "question": question,
        "context": context
    })

    return response

This design cleanly separates concerns. Prompt construction, model execution, and output parsing live in one place. The rest of the application only deals with inputs and outputs.

Orchestrating Everything in the FastAPI Endpoint

The endpoint now becomes a thin orchestration layer.

@router.post(path="/query", response_model=Insight)
def create_insight(
        request: InsightQuery,
        settings: Settings = Depends(get_settings),
        llm: BaseChatModel = Depends(init_openai_chat_model)
):
    """
    Post Insights Endpoint: Creates a new insight to a given context and related question.
    """

    prompt_messages = load_prompt_messages(
        settings.prompt.insight_path,
        settings.prompt.insight_version
    )

    response = run_insight_chain(
        prompt_messages,
        llm,
        request.question,
        request.context
    )

    return response

The endpoint coordinates configuration, prompt loading, and chain execution without embedding business logic. This keeps the API readable and makes future extensions straightforward.

Why this Structure Scales

Even though the example is simple, the structure is intentionally forward compatible. Retrieval can later be added as another dependency. Agent logic can replace the chain function without touching the endpoint contract. State handling and error management can be layered on top without rewriting the core flow.

Most importantly, AI is treated as a backend concern, not a special case. It follows the same architectural rules as any other component in a production system.

Final Thoughts

This article shows the difference between experimenting with AI and operating it as part of a backend system. It establishes the first building block of a production oriented AI backend. From here, adding retrieval, memory, or agents becomes an architectural decision instead of a refactor.

💻 Code on GitHub: hamluk/fastapi-ai-backend/part-2
📘 Read Part 3 of the Series: Building Production-Ready RAG in FastAPI with Vector Databases

Building Production-Ready AI Backends with FastAPI

Lukas Hamm — Tue, 30 Dec 2025 05:42:43 +0000

Why AI Backends Fail in Production (and how to build them properly)

Building AI demos has never been easier. With notebooks, Streamlit or Gradio, you can create something impressive in minutes.

But once AI is supposed to live inside a real system that serves requests, integrates with data sources, handles errors and evolves over time, most of these approaches start to fall apart. This article series focus on exactly that gap:
How to turn AI PoCs into production-ready backend services.

Thinking of AI as a backend service instead of a demo

A production-ready AI backend needs to do more than generate text.

It must provide:

clearly defined input and output contracts
predictable behavior despite probabilistic models
clean integration with databases, APIs and business logic
controlled error handling
extensibility (RAG, tools, agents, async workflows)
testability and long-term maintainability

A demo optimizes for speed and visibility. A backend service optimizes for reliability, structure and integration. That distinction becomes critical as soon as AI is not the product itself, but a capability inside a larger system.

This is where backend frameworks, and especially FastAPI, start to matter.

Why FastAPI fits production-ready AI backends so well

Its core concepts map extremely well to the architectural challenges of AI systems. Instead of treating AI as a special case, FastAPI allows it to be handled like any other backend component with clear boundaries and responsibilities.

Structured contracts instead of free-form AI

Large language models are probabilistic. Backend systems are not.

Using Pydantic as a contract layer makes AI outputs machine-consumable:

validated prompt inputs
structured LLM responses
tool schemas for agents
explicit agent actions

This is essential if AI is supposed to interact reliably with existing systems. Without strict contracts, small deviations in model output quickly lead to runtime errors, brittle integrations and difficult debugging.

RAG and agents as first-class dependencies

Retrieval pipelines, vector stores, agent tools or memory components are often treated as something “special” in AI projects. FastAPI’s dependency injection model removes that distinction.

RAG components, agents and tools can be injected exactly like:

database sessions
configuration objects
external services

This leads to a clean separation of concerns:

API layer
AI logic
retrieval
tooling
infrastructure

The result is an architecture where AI components are replaceable, testable and composable, rather than being hard wired into endpoint logic.

Backend primitives that matter for AI in production

Production AI systems often require:

file uploads (documents for RAG pipelines)
background tasks for long-running AI jobs
async execution for performance
security layers around AI endpoints

FastAPI provides these primitives out of the box, which makes it easier to move from experimentation to real backend services without changing the entire stack.

A minimal architectural idea in code

Not as a draft, but as a visual concept:

@app.post("/chat", response_model=ChatResponse)
asyncdefchat(
    prompt: ChatRequest,
    llm=Depends(get_llm),
):
    result = llm.invoke(prompt.message)
return ChatResponse(answer=result)

This is intentionally simple, but it already enforces:

a strict input and output schema
a decoupled AI component
a testable endpoint
an extensible foundation for RAG or agents

That is the difference between “AI as a demo” and “AI in a production backend”.

Conclusion: production-ready AI is an architectural problem

Building production-ready AI backends is not about better prompts or bigger models.

It is about:

contracts
separation of concerns
controlled integration
predictable behavior

Good architecture turns AI into a reliable and powerful backend capability. It allows you to design systems in which AI creates real and sustainable value instead of merely generating text.

📘 Read Part 2 of the Series: From LangChain Demos to a Production-Ready FastAPI Backend

Advanced Pydantic AI Agents: Building a Multi-Agent System in Pydantic AI

Lukas Hamm — Thu, 06 Nov 2025 07:56:40 +0000

In the previous part of this series, we introduced the concept of message history and explored how agents in Pydantic AI can maintain conversational context across multiple interactions.

In this part, we temporarily set aside message history and focus instead on a new concept in Pydantic AI: the Multi-Agent Pattern. This pattern allows us to design modular, cooperative AI systems that distribute responsibilities and work between agents to complete a task.

Understanding the Multi-Agent Pattern in Pydantic AI

The Multi-Agent Pattern is a design principle in which several agents cooperate to complete complex tasks.

Each agent is responsible for a specific domain or capability, and together they form a distributed intelligence system.

Pydantic AI supports several forms of multi-agent coordination, each designed for different control flow strategies.

Agent Delegation, where one agent calls another agent to perform a specific subtask.
Programmatic Agent Hand-Off, where control is explicitly transferred from one agent to another during runtime.
Graph-Based Control Flow, where agents are organized in a directed graph and the execution path is dynamically determined based on the context and results.

In this tutorial, we focus exclusively on the Agent Delegation pattern. It demonstrates how one agent can call another in a structured, type-safe way to delegate part of its work to another specialized agent.

Implementing Two Cooperative Agents

In our example, we divide the responsibilities between two agents.

TaskAgent handles user queries related to task creation.
ProjectManagementAgent manages project-level updates such as tracking open tasks.

The TaskAgent is responsible for creating tasks and delegates the update of project details to the ProjectManagementAgent through a tool call.

Let’s start by adapting the Task Agent.

task_agent.py

class TaskAgent():
    def __init__(self):
        self.agent = self._init_agent()
        self.pm_agent = ProjectManagementAgent()

    def _init_agent(self) -> Agent:
        agent = Agent(
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=TaskModel | Failed,
            deps_type=TaskDependency,
        )

        @agent.system_prompt
        def create_system_prompt(ctx: RunContext[TaskDependency]) -> str:
            return (f"""
                    You are a task creation agent.
                    Always follow the following instructions:
                    1. Update the project {ctx.deps.project} using the 'update_project' tool.
                    2. Create a brief task from the user's query. It is created by {ctx.deps.username} and belongs to the project {ctx.deps.project}.
                    """)

        @agent.tool_plain
        async def update_project(project_name: str) -> str:
            """Update the new created task's project."""
            result = await self.pm_agent.run(query=f"Update open tasks for {project_name}")

            if isinstance(result, Failed):
                return Failed(reason=f"Task creation failed: {result.reason}")

            return f"Updated project details for project {project_name}: {result.project}."

        return agent

pm_agent.py

class ProjectManagementAgent:
    def __init__(self):
        self.agent = self._init_agent()

    def _init_agent(self) -> Agent[ProjectDetails | Failed, None]:
        agent = Agent[ProjectDetails | Failed, None](
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=ProjectDetails | Failed,
            system_prompt=("""
                           You are a project management agent with the task of updating the number of open tasks for a project.
                           """)
        )

        @agent.tool_plain
        def update_open_tasks(project_name: str) -> ProjectDetails | Failed:
            """Updates the number of open tasks for a given project."""
            project = PROJECTS.get(project_name, None)
            if project is None:
                return Failed(reason=f"Updating open tasks failed for project {project_name}. Project does not exist.")

            # create internal ProjectDetails python object and increase number of open tasks by one
            project_details = ProjectDetails(owned_by_group=project.get("group"), number_of_open_tasks=project.get("open_tasks"))
            project_details.number_of_open_tasks += 1

            # write the new project details back to the data
            # this reflects a database update instruction for a real-world application
            project["open_tasks"] = project_details.number_of_open_tasks

            return project_details

        return agent

With this structure, the TaskAgent no longer manages project data directly. Instead, it delegates this part of the process to the ProjectManagementAgent, which ensures a clear separation of functionality and improves maintainability.

This architectural separation also affects the data model. Each agent now has its own dedicated data class that represents its specific domain. The project-related data is handled by the ProjectManagementAgent using the new ProjectDetails class, while the TaskAgent works with the TaskModel, which now includes project-specific information through a nested field. Additionally, a Failed class was introduced to handle error states in a structured way.

model.py

class ProjectDetails(BaseModel):
    owned_by_group: str = Field(description="The group this project is owned by.")
    number_of_open_tasks: int = Field(description="The number of open tasks for the project.")

class TaskModel(BaseModel):
    task: str = Field(description="The title of the task.")
    description: str = Field(description="A brief description of the task specifing its steps.")
    priority: int = Field(description="The priority of the task on a scale from 1 (high) to 5 (low).")
    project: str = Field(description="The project the task belongs to.")
    created_by:str = Field(description="The name of the person who created the task.")
    project_details: ProjectDetails = Field(description="The details of the project this task belongs to.")

class Failed(BaseModel):
    reason: str = Field(description="The reason for the failure.")

The Failed class standardizes error handling across agents. Instead of returning unstructured strings or exceptions, an agent can return a Failed object with a descriptive reason. The calling agent can then interpret the failure, propagate it, or respond accordingly.

This design keeps success and failure paths explicit and type-safe, making debugging and testing much easier. In essence, the separation of agents is mirrored by a separation of data models, which allows a clear, maintainable, and scalable approach for building cooperative multi-agent systems.

Running the Multi-Agent System

Let’s now execute our Pydantic AI multi-agent system and observe how the two agents interact under both normal and failure conditions.

app.py

from dotenv import load_dotenv
from LearnPydanticAI.task_agent import TaskAgent
from LearnPydanticAI.dependecies import TaskDependency

import asyncio

async def main():
    load_dotenv()  # take environment variables from .env.

    agent = TaskAgent()

    # successful run
    deps = TaskDependency(username="sudo", project="LearnPydanticAI")
    answer = await agent.run("I want to learn more about Pydantic AI.", deps=deps)
    print(answer)

    # failed run due to a non-existing project
    deps_fail = TaskDependency(username="sudo", project="NonExistingProject")
    failed_answer = await agent.run("I want to learn more about Pydantic AI.", deps=deps_fail)
    print(failed_answer)

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

In the first case, the agent successfully creates a task and updates the project information.

In the second case, we simulate an error scenario by providing a project name that does not exist. The ProjectManagementAgent detects that the project cannot be found and returns a Failedobject, which is then handled by the TaskAgent.

% poetry run python src/LearnPydanticAI/app.py
task='Learn about Pydantic AI' 
description='1. Research about Pydantic AI.\n2. Learn about its features and capabilities.\n3. Explore its documentation and tutorials.' 
priority=3 
project='LearnPydanticAI' 
created_by='sudo' 
project_details=ProjectDetails(owned_by_group='dev_team', number_of_open_tasks=3)

reason='Task creation failed: Updating open tasks failed for project NonExistingProject. Project does not exist.'

This shows how Pydantic AI agents can coordinate their behavior while maintaining robust error handling across agent boundaries.

What Happens Behind the Scenes

When the query "I want to learn more about Pydantic AI." is executed, the following sequence occurs:

The TaskAgent receives the query and prepares to create a new task.
Before creating the task, it invokes the update_project tool.
This tool delegates the call to the ProjectManagementAgent, which updates the number of open tasks for the specified project.
Once the update is successful, the result is returned to the TaskAgent.
The TaskAgent then completes the task creation process and outputs a structured TaskModel.

When a non-existing project name is passed to the agent, a different control flow is triggered.

The ProjectManagementAgent attempts to update the specified project but cannot find a corresponding entry in the PROJECTS dictionary.
It returns a Failed object with a descriptive error message that explains why the update could not be completed.
The TaskAgent receives this Failed response instead of a project update result.
As a result, the TaskAgent stops the task creation process and propagates the failure, ensuring that no inconsistent data is produced.

This control flow illustrates how multiple agents in Pydantic AI can coordinate their tasks, exchange structured information, and gracefully recover from errors. It also demonstrates the value of well-defined data models and typed outputs, which make error handling explicit and maintain the transparency of the system’s internal logic.

Relevance for Real-World Applications

The The Multi-Agent Pattern in Pydantic AI is a powerful concept for designing scalable and modular AI systems. It allows developers to distribute complex logic across specialized agents that can interact and support one another.

In production environments, this approach enables teams to:

Divide large workflows into modular components.
Maintain clear boundaries between domains and functionalities.
Extend or replace individual agents without changing the overall architecture.
Achieve parallel and distributed reasoning across multiple AI modules.
Integrate domain-specific intelligence through dedicated agents.

Using Agent Delegation in Pydantic AI combines the flexibility of natural language interfaces with the robustness of structured and typed agent communication. It provides a scalable foundation for building collaborative AI systems in Python that are easy to maintain, extend, and adapt to complex real-world scenarios.

Conclusion to this Series

With this fifth part, the current tutorial series on Pydantic AI Agents comes to a close. We’ve gone from building a simple, structured agent to exploring advanced features such as dependencies, function tools, message history, and finally, the multi-agent pattern with agent delegation.

Together, these concepts form a solid foundation for designing intelligent, modular, and production-ready AI systems in Python. While this marks the end of the series, it’s only the beginning of what can be done with Pydantic AI. Continue exploring advanced concepts and mechanisms, ranging from graph-based agent orchestration to persistent memory architectures and Pydantic Evals.

Thank you for following along this journey into the world of structured AI agent design with Pydantic AI. If you’ve enjoyed this series, stay tuned for more insights on AI programming in Python and modern AI frameworks. I’ll continue sharing practical tutorials and engineering perspectives on this blog to help bridge the gap between AI research and real-world software development.

💻 Code on GitHub: hamluk/LearnPydanticAI/part-5

📘 Read Part 4: Extending Pydantic AI Agents with Chat History

Extending Pydantic AI Agents with Chat History - Messages and Chat History in Pydantic AI

Lukas Hamm — Thu, 06 Nov 2025 07:35:34 +0000

Up to now we built an agent that processes user input and returns structured, predictable responses. The agent uses dependencies for contextual information and can access and modify project data via Function Tools.

We now extend the agent so it can use past interactions: it will access and reuse prior messages through a shared conversation history. In Part 3 we already examined the agent’s internal communication using result.all_messages() . Looking at this concept was actually the first step toward understanding how the conversation memory in Pydantic AI works.

Introducing Pydantic AI Message History

Every time our agent runs, it exchanges a sequence of structured messages: system prompts, user queries, model responses and any function/tool calls. Until now each run_sync() call was isolated and stateless. Pydantic AI Agent in Python allows us to pass a message_history argument to run_sync() so that we can reuse previous messages across runs. Using a message_historyis important because it makes the agent’s behavior consistent across multiple interactions and moves the agent from isolated single-turn responses toward reproducible, context-aware multi-turn workflows.

Extending Our Agent with a Session

To make this process easier to manage, we’ll introduce a simple session class that wraps around our existing TaskAgent. This AgentSession will handle the message history automatically by storing every new message after each run into a list and reusing them for the next one. It does so by passing the list of past results to the agent on every run via the message_history parameter.

session.py

class AgentSession:
    def __init__(self, agent):
        self.agent = agent
        self.history = []

    def run(self, query: str, deps) -> str:
        result = self.agent.run_sync(query, deps=deps, message_history=self.history)
        self.history.extend(result.new_messages())
        return result.output

This way, the agent uses previous user inputs and model outputs to answer subsequent queries. With this step, we move from single-run executions to a persistent session model that maintains state across interactions.

Putting the Session into Action

Let’s see this in practice by updating our app to use our existing TaskAgent together with the new AgentSession.

app.py

from dotenv import load_dotenv
from LearnPydanticAI.agent import TaskAgent
from LearnPydanticAI.session import AgentSession
from LearnPydanticAI.dependecies import TaskDependency

load_dotenv()  # take environment variables from .env.

agent = TaskAgent()
session = AgentSession(agent.agent)
deps = TaskDependency(username="sudo", project="LearnPydanticAI")

answer1 = session.run(query="I want to learn more about Pydantic AI.", deps=deps)
print(answer1)

answer2 = session.run(query="Update its creator to joe", deps=deps)
print(answer2)

In this example:

We initialize the TaskAgent and wrap it with an AgentSession.
The first query stays the same and asks the agent to create a task.
The second query refers to “its user”. Because the session preserves the message history, the agent can resolve this to the previously created task and apply the change.

This demonstrates the easiest way our agent can maintain continuity across multiple runs.

% poetry run python src/LearnPydanticAI/app.py
task='Learn about Pydantic AI' 
description='1. Research about Pydantic AI.\n2. Learn about its features and capabilities.\n3. Experiment with simple examples.' 
priority=3 
project='LearnPydanticAI' 
created_by='sudo' 
number_of_open_tasks=3 
success=True

task='Learn about Pydantic AI' 
description='1. Research about Pydantic AI.\n2. Learn about its features and capabilities.\n3. Experiment with simple examples.' 
priority=3 
project='LearnPydanticAI' 
created_by='joe' 
number_of_open_tasks=3 
success=True

Now that we’ve seen the session in action, let’s take a look under the hood to understand how PydanticAI manages message history internally.

How It Works Behind the Scenes

When run_sync() is called with a message_history, Pydantic AI automatically:

Combines the historical messages with the new user input,
Generates a complete conversational context for the model,
Returns all messages including historical and newly created messages.

By using the message_history we effectively allow the agent to “remember” everything that has happened so far in the session.

Further Pydantic AI Message History Mechanisms

For the purpose of this tutorial, we are not going to extend our agent more than with a session to automatically handle message history. However, Pydantic AI offers additional capabilities for working with message history that go beyond a simple session. We’ll explore in the following sections more mechanisms that Pydantic AI offers to persist history, process it before each run and dynamically adapt it based on the agents context.

Persisting and Reloading Message History

In many cases, we might want to preserve the conversation history beyond a single runtime — for example, to store it on disk, log it for later inspection, or share it between services.

Pydantic AI provides a convenient way to serialize and reload the entire message history using the ModelMessagesTypeAdapter utility. This ensures that all messages, including user queries, system prompts, and function calls, can be stored as structured data and later validated when reloaded.

result1 = self.agent.run_sync("Create a new task", deps=deps, message_history=self.history)
history_step_1 = result1.all_messages()

# Convert history into JSON-serializable Python objects
python_objects = to_jsonable_python(history_step_1)

# Validate and reload the same history structure
history_from_step_1 = ModelMessagesTypeAdapter.validate_python(python_objects)

# Reuse the restored history in a new run
result2 = self.agent.run_sync("Change its user to joe", deps=deps, message_history=history_from_step_1)

This pattern is ideal when you want your agent’s state to persist between sessions or when running long-lived applications where context continuity matters. The agent now can pause, resume, or transfer its memory safely.

Processing and Filtering History

Sometimes, it’s neither necessary nor desirable to forward the entire message history to the model. Pydantic AI provides history_processors, which allow you to inspect and modify the history before each run.

Common use cases include:

Privacy filtering: removing user-identifying information before sending context to the model.
Token optimization: truncating older messages to save cost and improve performance.
Custom logic: transforming messages, rephrasing context, or aggregating prior results.

async def keep_recent_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
    """Keep only the last 5 messages to manage token usage."""
    return messages[-5:] if len(messages) > 5 else messages

def _init_agent(self) -> Agent:
        agent = Agent(
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=TaskModel,
            deps_type=TaskDependency,
            history_processors=[keep_recent_messages]
            )

Context-Aware History Processing

For even more control, Pydantic AI supports the context_aware_processor, which lets you modify message history based on the current run context.

This is useful when certain dependencies or runtime parameters should influence how much of the history is visible to the model. For instance, you might:

Hide system prompts when the user has limited access rights,
Provide additional context only for specific projects or tasks, or
Apply specialized summarization logic depending on the user’s role.

async def keep_recent_messages(ctx: RunContext[TaskDependency], messages: list[ModelMessage]) -> list[ModelMessage]:
    """Keep only the last 5 messages for non sudo users."""
    if ctx.deps.username != "sudo":
          return messages[-5:] if len(messages) > 5 else messages

       return messages

def _init_agent(self) -> Agent:
        agent = Agent(
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=TaskModel,
            deps_type=TaskDependency,
            history_processors=[keep_recent_messages]
            )

With context_aware_processor, the history can dynamically adapt to the agent’s current execution context, giving you fine-grained control over what information the model sees at each step.

Relevance for Real-World Applications

Message history plays an essential role when integrating AI agents into real-world systems. It enables agents to maintain context across interactions and to behave more consistently over time. By storing and reusing messages, developers can create agents that understand past actions, follow long-term goals, and build on previous results.

In practical scenarios, conversation memory allows agents to:

Maintain context over multiple user sessions
Improve accuracy by referencing earlier information
Adapt behavior based on previous interactions
Provide personalized experiences for different users
Reduce redundancy by avoiding repeated inputs

Using message history in PydanticAI combines the flexibility of natural language interfaces with the reliability of structured, contextual memory. This makes it a foundational concept for building stateful, context-aware AI agents that can operate effectively in complex, real-world environments.

Continuing the Series

With this part, we have extended our agent to support persistent message history, a key feature for building context-aware and stateful AI agents in Python. In the next part, we will shift our focus to multi-agent systems in Pydantic AI and explore how multiple agents can collaborate, delegate tasks, and work together to solve more complex problems in a structured and predictable way.

💻 Code on GitHub: hamluk/LearnPydanticAI/part-4

📘 Read Part 5: Advanced Pydantic AI Agents: Multi-Agent System

Extending Pydantic AI Agents with Function Tools — Reading and Updating Project Data in Python

Lukas Hamm — Thu, 16 Oct 2025 07:40:58 +0000

In the previous part of this series, we extended our Pydantic AI agent with dependencies, allowing it to use contextual information such as the user and project name when generating tasks.

In this part, we’ll further enhance our agent by enabling it to access and modify project data through Function Tools in PydanticAI.

This functionality allows the agent to perform controlled interactions with external data, such as retrieving project details or updating records. By introducing Function Tools, we make our agent more autonomous, structured, and adaptable for real-world applications.

Why Function Tools Are Crucial for Python AI Agents in PydanticAI

In real-world software systems, an AI agent often needs to interact with its environment. It might query project information, update task counts, or verify user permissions before executing an action.

To achieve this reliably, we need a structured interface between the agent and the systems it operates in. PydanticAI Function Tools provide exactly that interface. They allow the agent to call predefined Python functions in a type-safe and predictable way, ensuring consistent behavior across multiple runs. This makes Function Tools an essential component for integrating AI agents into production-grade Python applications.

Defining a Simple Data Source

Disclaimer:

For demonstration purposes, this tutorial uses a Python dictionary to simulate a database. In a real application, these operations would typically be implemented through actual database queries or API requests.

data.py

PROJECTS = {
    "LearnPydanticAI": {"group": "dev_team", "open_tasks": 2},
    "InternalResearch": {"group": "data_team", "open_tasks": 4}
}

This dictionary represents our data source. Each project includes information about the assigned group and the current number of open tasks.

Extending the Response Model

Next, we will extend our TaskModel of the agent to include additional fields that reflect the operation’s outcome. With this extension we enable the agent to communicate whether the operation was successful and how many open tasks remain for a given project.

model.py

from pydantic import BaseModel, Field

class TaskModel(BaseModel):
    task: str = Field(description="The title of the task.")
    description: str = Field(description="A brief description of the task specifying its steps.")
    priority: int = Field(description="The priority of the task on a scale from 1 (high) to 5 (low).")
    project: str = Field(description="The project the task belongs to.")
    created_by: str = Field(description="The name of the person who created the task.")
    number_of_open_tasks: int = Field(description="The number of open tasks for the project after creating this task.")
    success: bool = Field(description="Indicates if the task was created successfully.")

Integrating PydanticAI Function Tools into the Agent

Now, let’s update our agent to use two function tools and interact with the simulated data source. The first retrieves project information, while the second updates the number of open tasks after a new task is created.

agent.py

from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
from pydantic_ai import Agent, RunContext

from LearnPydanticAI.dependencies import TaskDependency
from LearnPydanticAI.model import TaskModel
from LearnPydanticAI.projects import PROJECTS

import os

class TaskAgent:
    def __init__(self):
        self.agent = self._init_agent()

    def _init_agent(self) -> Agent:
        agent = Agent(
            model=MistralModel(
                model_name=os.getenv("LLM_MISTRAL_MODEL"),
                provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))
            ),
            output_type=TaskModel,
            deps_type=TaskDependency,
        )

        @agent.system_prompt
        def create_system_prompt(ctx: RunContext[TaskDependency]) -> str:
            return (f"""
                    Create a brief task from the user's request.
                    The task is created by {ctx.deps.username}.
                    The task belongs to the project {ctx.deps.project}.
                    Before creating the task, verify that the project {ctx.deps.project} exists by retrieving its details.
                    If you verified that it exists, update the number of open tasks for this project.
                    If the project does not exist, mark the task creation as failed and leave the task and its description empty.
                    """)

        @agent.tool_plain
        def get_project_info(project_name: str) -> str:
            """Retrieves project information."""
            project = PROJECTS.get(project_name, None)
            if not project:
                return f"Project {project_name} not found."
            return project

        @agent.tool_plain
        def update_open_tasks(project_name: str, new_open_tasks: int) -> str:
            """Updates the number of open tasks for a given project."""
            project = PROJECTS.get(project_name, None)
            if not project:
                return f"Updating open tasks failed for project {project_name}. Project does not exist."
            project["open_tasks"] = new_open_tasks
            return f"Successfully updated open tasks for project {project_name} to {project['open_tasks']}."

        return agent

    def run(self, query: str) -> str:
        deps = TaskDependency(username="sudo", project="LearnPydanticAI")
        result = self.agent.run_sync(query, deps=deps)
        return result.output

This setup allows the agent to retrieve project data and update it when creating new tasks. Both functions are registered using the @agent.tool_plain decorator.

In addition to this, PydanticAI also provides the @agent.tool decorator. Both decorators register Function Tools, but they serve different purposes. Let’s take a closer look at how they differ and when to use each of them.

`@agent.tool_plain` decorator

The @agent.tool_plain decorator registers a synchronous function as a tool that can be directly executed by the agent. These function tools typically perform simple operations, like retrieving values from memory, performing calculations or extending the agents capabilities in another way.

In this example, both get_project_info() and update_open_tasks() are synchronous, and their execution does not depend on external network requests.

`@agent.tool` decorator

In contrast, the @agent.tool decorator is designed for asynchronous operations, such as API calls. When using @agent.tool, the function also receives a RunContext object, which allows access to dependency values like API hosts, authentication keys and database operations.

Example 1: External API Call

@agent.tool
async def get_project_info(ctx: RunContext[TaskDependency], project_name: str) -> str:
    """Retrieve project info from an external API."""
    api_host = ctx.deps.api_host
    api_key = ctx.deps.api_key
    response = await httpx.get(f"{api_host}/projects/{project_name}", headers={"Authorization": f"Bearer {api_key}"})
    return response.text

Example 2: Internal Database Call

@agent.tool
async def get_project_info(ctx: RunContext[DBDependencies], project_name: str) -> str:
    if ctx.deps is None:
        raise ValueError("Dependencies (ctx.deps) are not set.")

    db = ctx.deps.db
    projects = db.get_project_info(project_name)
    return f"{projects}"

This approach is essential when connecting your AI agent to real-world data sources, where security, authentication, and asynchronous execution are required.

Running the Agent

When we execute the agent with this configuration, it validates whether the project exists, creates a new task, and updates the number of open tasks.

% poetry run python src/LearnPydanticAI/app.py
task='Learn about Pydantic AI'
description='1. Research about Pydantic AI. 2. Understand its core concepts and applications. 3. Experiment with simple examples and exercises.'
priority=2
project='LearnPydanticAI'
created_by='sudo'
number_of_open_tasks=3
success=True

We can validate the agents behavior by I changing the dependencies handed to the agent, to include a project that does not exist.

agent.py

...
deps = TaskDependency(username="sudo", project="UnknownProject")
result = self.agent.run_sync(query, deps=deps)
return result.output
...

After doing so, the agent responds accordingly.

% poetry run python src/LearnPydanticAI/app.py
task=''
description=''
priority=1
project='UnknownProject'
created_by='sudo'
number_of_open_tasks=0
success=False

This demonstrates how Function Tools enhance the agent’s ability to make context-aware decisions and update its environment accordingly.

Bonus: Inspecting Function Tool Calls in PydanticAI

When working with Function Tools, it is often helpful to analyze how the agent interacts with them during execution. PydanticAI provides a simple way to inspect this process through the all_messages() method of the result object.

This method returns the complete internal message history — including system prompts, tool calls, intermediate reasoning, and final responses. It allows developers to understand how the agent decides when and why to call a tool, which can be invaluable when debugging or optimizing complex workflows.

agent.py

...
# Bonus: inspecting function tool calls
for message in result.all_messages():
    print(message)
...

The output reveals each step of the conversation, such as:

The system prompt created for the current run
The function call the agent decided to make (get_project_info or update_open_tasks)
The parameters passed to each call
The resulting updates before the final structured output

By inspecting these internal messages, developers can gain deeper insight into how the PydanticAI runtime orchestrates reasoning and function execution. This kind of introspection is especially valuable when designing complex, multi-tool agents or optimizing prompt behavior for production environments.

Relevance for Real-World Applications

Function Tools are a critical feature when integrating AI agents into real-world systems.

They allow the agent to go beyond generating text and to perform meaningful actions such as fetching data, validating information, and modifying state.

In practice, agents often need to:

Validate input or synchronize state
Fetch data from APIs or internal services
Update records in databases
Interact with external environments or data stores

By implementing these patterns through PydanticAI Function Tools, developers gain both the flexibility of AI and the reliability of typed, structured interfaces.

What’s Next

With Function Tools implemented, our agent can now access, validate, and update contextual project data.

In the next part of this PydanticAI tutorial, we will explore how to integrate real external APIs and securely manage dependencies for more advanced automation workflows.

💻 Code on GitHub: hamluk/LearnPydanticAI/part-3

📘 Read Part 4: Extending Pydantic AI Agents with Chat History

Extending Pydantic AI Agents with Dependencies — Adding Context to Your AI Agents

Lukas Hamm — Wed, 15 Oct 2025 23:50:38 +0000

In the first part of this series, we built a simple Pydantic AI agent in Python that could process user input and return structured, predictable responses.

In this part, we’ll take it a step further by giving our agent contextual information with dependencies, defining who is asking and which project the request belongs to.

This is where dependencies come into play. By using deps_type in Pydantic AI, we can define structured data that provides the agent with essential background information. This makes the agent more intelligent, consistent, and better suited for real-world applications.

Why Dependencies Are Crucial for Python AI Agents in PydanticAI

In real-world software systems, an AI agent rarely operates in isolation. It usually acts within a larger environment — for example, handling requests for specific users, working on particular projects, or referencing external data sources.

This context helps create more reliable and relevant agent behavior. Pydantic AI offers a clean and type-safe way to provide that context through dependencies. By explicitly defining what the agent should know about its environment, we ensure that every run starts with the right information.

Defining Dependencies

Let’s start by defining a simple dependency for our Python AI Task Agent that specifies who is creating a task and to which project it belongs to. We’ll use a Python dataclass to store this structured information, enabling our AI agent to access contextual data reliably.

dependencies.py

from dataclasses import dataclass

@dataclass
class TaskDependency:
    username: str
    project: str

This class defines a structured data object that our agent can later access when it needs to generate a task description.

Updating the Agent

Now we’ll update our existing TaskAgent so that it can make use of this dependency. In a first step we enhance the output model to include project and creator information.

model.py

from pydantic import BaseModel, Field

class TaskModel(BaseModel):
    task: str = Field(description="The title of the task.")
    description: str = Field(description="A brief description of the task specifying its steps.")
    priority: int = Field(description="The priority of the task on a scale from 1 (high) to 5 (low).")
    project: str = Field(description="The project the task belongs to.")
    created_by: str = Field(description="The name of the person who created the task.")

This updated model ensures that each task generated by the agent not only contains the task title, description, and priority, but also explicitly includes who created the task and which project it belongs to.

For the agent the key additions are the deps_type parameter and the @agent.system_prompt decorator.

agent.py

from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
from pydantic_ai import Agent, RunContext

from LearnPydanticAI.dependecies import TaskDependency
from LearnPydanticAI.model import TaskModel

import os


class TaskAgent():
    def __init__(self):
        self.agent = self._init_agent()

    def _init_agent(self) -> Agent:
        agent = Agent(
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=TaskModel,
            deps_type=TaskDependency,
            )

        @agent.system_prompt
        def create_system_prompt(ctx: RunContext[TaskDependency]) -> str:
            return (f"""
                    Create a brief task from the user's request. 
                    The task is created by {ctx.deps.username}.
                    The taks belongs to the project {ctx.deps.project}.
                    """)

        return agent

    def run(self, query: str) -> str:
        deps = TaskDependency(username="sudo", project="LearnPydanticAI")
        result = self.agent.run_sync(query, deps=deps)
        return result.output

Let’s look more closely at what changed here:

deps_type=TaskDependency:
This tells the agent what kind of structured data it will receive as dependency information. In this case, it expects a TaskDependency object with a username and project name.
@agent.system_prompt decorator:
This decorator defines a function that dynamically generates the system prompt every time the agent runs. The function receives a RunContext object, which includes the dependency data as ctx.deps. By referencing ctx.deps.username and ctx.deps.project, we can build a prompt that adapts to the given context at runtime.

This approach provides much more flexibility than using a static prompt string. It allows your agent to respond differently based on who uses it or what project it’s currently handling.

Running the Agent with Context

Now that we’ve added dependencies in PydanticAI, the agent can include user- and project-specific information, generating responses that are context-aware and more useful in real-world automation scenarios.

% poetry run python src/LearnPydanticAI/app.py
task='Learn about Pydantic AI' description='Research and learn about Pydantic AI, its features, and applications.' priority=3 project='LearnPydanticAI' created_by='sudo'

Instead of producing a generic task, the output now reflects who created the task and where it belongs. This makes the agent’s responses far more meaningful and adaptable to different use cases.

Why Dependencies Are Essential in Real Projects

Dependencies are a key component when integrating AI agents into production systems. In real projects, agents often need to:

Access user-specific data (permissions, preferences, or roles)
Operate within project contexts (repositories, workspaces, or tenants)
Use environmental settings (API tokens, configuration values, or runtime parameters)

With Pydantic AI, you can manage all of this using strongly typed dependencies. It enforces structure and reliability while keeping your code clean and predictable. This approach aligns well with common software development practices, allowing developers to design AI features with the same structure and reliability as any other part of their system.

What’s Next

With PydanticAI dependencies in place, our agent can now understand its environment and produce more meaningful, context-aware results.

In the next part of this PydanticAI tutorial, we’ll extend this foundation by integrating external tools, enabling the agent to interact with databases, APIs, and other real-world systems.

💻 Code on GitHub: hamluk/LearnPydanticAI/part-2

📘 Read Part 3: Extending Pydantic AI Agents with Function Tools

Getting Started with PydanticAI — Basics for AI Agents in Python

Lukas Hamm — Tue, 14 Oct 2025 07:51:16 +0000

What's the Reason behind all of that?

For many developers, AI Agents are still mostly an idea rather than something implemented in real projects. These agents have the potential to change the way software projects are built, becoming a standard part of future codebases. They can organize, validate, and structure information in ways that make AI-driven workflows more predictable and reliable.

In this series, we'll take a closer look at Pydantic AI - a Python agent framework - and how to bring these AI agents to life, starting with the basics and gradually exploring more advanced possibilities.

Getting Started with Pydantic AI and Mistral

We’ll start this series by building our first working example of an AI agent. We'll cover the basics of PydanticAI and how it works, including step-by-step instructions and practical code examples. This will establish the foundation for the following tutorials, where we’ll explore more advanced features and workflows.

Setting Up the Project

Let's start by setting up a clean Python environment and installing the required dependencies. I’ll be using Poetry, a dependency management and packaging tool for Python, but feel free to use any setup method you prefer.

In this series, I’ll work with the Mistral model. Since Pydantic AI is a model-agnostic framework, you can easily use a different model if you like.

# create project folder structure
poetry new LearnPydanticAI
cd LearnPydanticAI

# install PydanticAi and Mistral dependencies
poetry add "pydantic-ai-slim[mistral]"

To use your model, make sure the API key is available as an environment variable. I prefer to store these values in a .env file and load them in my application using load_dotenv().

.env

LLM_MISTRAL_MODEL=<mistral_model>
MISTRAL_API_KEY=<mistral_api_key>

For this, we’ll also install the python-dotenv package.

# install python-dotenv as dependency
poetry add python-dotenv

Core Concepts — Agents and Structured Output

The first two core concepts of Pydantic AI that we are going to cover are Agents and Structured Output.

An Agent in Pydantic AI is a self-contained component that can take instructions, interact with a language model, and return results. Unlike calling an LLM directly, agents are designed to be reusable, predictable, and easily integrated into larger workflows. Each agent can be configured with specific behaviors, prompts, and validation rules, which allows you to build reliable AI-driven processes.

The second concept - and one of Pydantic AI's key features - is its ability to produce structured output. Instead of receiving free-form text from the model, you define a data model using Pydantic. The agent ensures that the AI’s response conforms to this model, which reduces errors and makes downstream processing straightforward.
Structured output allows developers to focus on practical use cases, rather than parsing and cleaning model responses. It makes AI Agents not just an experimental tool, but a component that can reliably interact with other parts of a software project.

Build Your First Agent

Now that we’ve covered the basic concepts, it’s time to build our first working AI agent. We’ll start small — creating an agent that takes a user request and returns a concise task description.

agent.py

from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
from pydantic_ai import Agent
import os


class TaskAgent:
    def __init__(self):
        self.agent = self._init_agent()

    def _init_agent(self) -> Agent:
        agent = Agent(
            model=MistralModel(
                model_name=os.getenv("LLM_MISTRAL_MODEL"),
                provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))
            ),
            system_prompt="Create a brief task from the user's request."
        )
        return agent

    def run(self, query: str) -> str:
        result = self.agent.run_sync(query)
        return result.output

With our agent defined, we can now write a simple app.py to run it for the first time.

app.py

from dotenv import load_dotenv
from LearnPydanticAI.agent import TaskAgent

load_dotenv()  # load environment variables from .env

agent = TaskAgent()
answer = agent.run("I want to learn more about Pydantic AI")
print(answer)

Here’s an example of what the output might look like:

% poetry run python src/LearnPydanticAI/app.py
Task: Research Pydantic AI

1. Objective: Gather information about Pydantic AI to understand its features, use cases, and benefits.

2. Steps:
   - Visit the official Pydantic documentation or website.
   - Read about Pydantic's core features and how it integrates with AI/ML workflows.
   - Explore tutorials or case studies showing Pydantic AI in action.
   - Check community discussions on GitHub or Stack Overflow.
   - Summarize key insights such as data validation and type safety for AI projects.

3. Deliverable: A concise overview of Pydantic AI — what it is, how it works, and why it matters.

Add Structured Output

Now that the agent is working, let’s make its responses structured and predictable. We’ll do this by defining a response schema using Pydantic models.

model.py

from pydantic import BaseModel

class TaskModel(BaseModel):
    task: str
    description: str
    priority: int

We then connect this model to the agent by using the output_type argument in its constructor.

agent.py

from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
from pydantic_ai import Agent
from LearnPydanticAI.model import TaskModel

import os


class TaskAgent:
    def __init__(self):
        self.agent = self._init_agent()

    def _init_agent(self) -> Agent:
        agent = Agent(
            model=MistralModel(
                model_name=os.getenv("LLM_MISTRAL_MODEL"),
                provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))
            ),
            output_type=TaskModel, # new line
            system_prompt="Create a brief task from the user's request."
        )
        return agent

    def run(self, query: str) -> str:
        result = self.agent.run_sync(query)
        return result.output

Running the agent again now produces a structured response that matches the schema we defined:

% poetry run python src/LearnPydanticAI/app.py
task='Research Pydantic AI' description='Learn about the features, use cases, and benefits of Pydantic AI.' priority=3

Bonus — Add Field Descriptions

Defining the output with a Pydantic BaseModel is one of Pydantic AI’s most powerful features.
Not only does it guarantee structured output and eliminate the need for manual parsing — it also allows you to guide the model by describing each field in more detail.

You can do this by using Pydantic’s Field() function:

model.py

from pydantic import BaseModel, Field

class TaskModel(BaseModel):
    task: str = Field(description="The title of the task.")
    description: str = Field(description="A brief description of the task specifying its goal.")
    priority: int = Field(description="The task priority on a scale from 1 (high) to 5 (low).")

With these field descriptions, the agent receives clearer context about the expected structure and meaning of each value — which results in more detailed and reliable outputs.

% poetry run python src/LearnPydanticAI/app.py
task='Research Pydantic AI' description="1. Open a web browser and go to a search engine like Google. 2. Type 'Pydantic AI' in the search bar and press enter. 3. Review relevant links and gather key information." priority=3

Wrapping Up

In this first part, we created a simple yet functional AI Agent using Pydantic AI and the Mistral model.
You learned how to initialize an agent, run it, and structure its responses using Pydantic models — making the output both predictable and ready for integration into real applications. With this foundation in place, you’re ready to move on to more complex scenarios.

In the next part, we’ll build on this foundation by exploring how to handle dependencies, extend agent capabilities, and connect your AI Agent with external data sources.

💻 Code for part 1: hamluk/LearnPydanticAI/part-1

📘 Read Part 2: Extending Pydantic AI Agents with Dependencies

Forem: Lukas Hamm

Building Production-Ready RAG in FastAPI with Vector Databases

From Prompting to Production-Ready RAG

RAG as a Backend Component

Scope of RAG in This Article

Introducing the Vector Store Dependency

Uploading Data into the Vector Store

Querying with RAG via Dependency Injection

Why This Architecture Matters

Final Thoughts

From LangChain Demos to a Production-Ready FastAPI Backend

Why LangChain Needs a Proper Backend Architecture

Thinking of AI as a Backend Component

Defining the Contract with Pydantic

Injecting the LLM via FastAPI Depends

Encapsulating the LangChain Logic

Orchestrating Everything in the FastAPI Endpoint

Why this Structure Scales

Final Thoughts

Building Production-Ready AI Backends with FastAPI

Why AI Backends Fail in Production (and how to build them properly)

Thinking of AI as a backend service instead of a demo

Why FastAPI fits production-ready AI backends so well

Structured contracts instead of free-form AI

RAG and agents as first-class dependencies

Backend primitives that matter for AI in production

A minimal architectural idea in code

Conclusion: production-ready AI is an architectural problem

Advanced Pydantic AI Agents: Building a Multi-Agent System in Pydantic AI

Understanding the Multi-Agent Pattern in Pydantic AI

Implementing Two Cooperative Agents

Running the Multi-Agent System

What Happens Behind the Scenes

Relevance for Real-World Applications

Conclusion to this Series

Extending Pydantic AI Agents with Chat History - Messages and Chat History in Pydantic AI

Introducing Pydantic AI Message History

Extending Our Agent with a Session

Putting the Session into Action

How It Works Behind the Scenes

Further Pydantic AI Message History Mechanisms

Persisting and Reloading Message History

Processing and Filtering History

Context-Aware History Processing

Relevance for Real-World Applications

Continuing the Series

Extending Pydantic AI Agents with Function Tools — Reading and Updating Project Data in Python

Why Function Tools Are Crucial for Python AI Agents in PydanticAI

Defining a Simple Data Source

Extending the Response Model

Integrating PydanticAI Function Tools into the Agent

@agent.tool_plain decorator

@agent.tool decorator

Example 1: External API Call

Example 2: Internal Database Call

Running the Agent

Bonus: Inspecting Function Tool Calls in PydanticAI

Relevance for Real-World Applications

What’s Next

Extending Pydantic AI Agents with Dependencies — Adding Context to Your AI Agents

Why Dependencies Are Crucial for Python AI Agents in PydanticAI

Defining Dependencies

Updating the Agent

Running the Agent with Context

Why Dependencies Are Essential in Real Projects

What’s Next

Getting Started with PydanticAI — Basics for AI Agents in Python

What's the Reason behind all of that?

Getting Started with Pydantic AI and Mistral

Setting Up the Project

Core Concepts — Agents and Structured Output

Build Your First Agent

Add Structured Output

Bonus — Add Field Descriptions

Wrapping Up

`@agent.tool_plain` decorator

`@agent.tool` decorator