Building a Personal CRM with PydanticAI
Introduction
There are a lot of LLM frameworks.
From LangGraph to LangChain, LlamaIndex and others. There’s a ton going on in the LLM frameworks.
I decided to take the time to build several applications in each as a part of a series.
I’m going to keep the app the same so that I can focus strictly on the LLM frameworks - comparing them.
PydanticAI
If you’ve used Python in the last 4 years, you’re probably familiar with Pydantic. Pydantic is a Python library that provides data validation and settings management using Python type annotations. It’s a powerful tool for building robust and maintainable applications.
The creators of Pydantic have founded a company and created a new framework called PydanticAI.
PydanticAI, in my opinion, is a pretty no frills framework that provides a simple and effective way to build AI applications. Culturally, it reminds me of Pydantic’s philosophy of simplicity and clarity.
That means that it’s a bit lower level that you might expect from a framework like LangChain or LlamaIndex. It’s designed to be flexible and customizable, allowing you to build your own AI applications from the ground up. It’s a great choice for those who want to have full control over their AI applications.
Why a CRM?
I chose a CRM because it’s a practical application, requires long term storage and tool use, and makes for a conceptually simple example.
This project serves as an excellent example of how to leverage a database and various tools to create a functional service. By implementing a CRM, we can maintain context and history, which are crucial for effective relationship management. Throughout this article, we will walk through the steps of setting up the database, defining the CRM service, creating the agent, and ultimately running the CRM system.
Let’s get started.
Setting Up the Database
To set up our personal CRM, we will use SQLite as a simple and effective database system. The first step is to initialize the database and create the necessary tables that will hold our data. We will create four tables: contacts, notes, tags, and a junction table contact_tags to manage the many-to-many relationship between contacts and tags.
Here’s the code to initialize the database and create the required tables:
import os
import asyncio
import sqlite3
from datetime import datetime
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import ModelMessageNext, we define the path for our database:
# Database setup
DB_PATH = "crm.db"Now, we can implement the function to initialize the database and create the tables:
def init_db():
"""Initialize the SQLite database with required tables."""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Create contacts table
cursor.execute("""
CREATE TABLE IF NOT EXISTS contacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create notes table
cursor.execute("""
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
contact_id INTEGER,
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (contact_id) REFERENCES contacts (id)
)
""")
# Create tags table
cursor.execute("""
CREATE TABLE IF NOT EXISTS tags (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL
)
""")
# Create contact_tags junction table
cursor.execute("""
CREATE TABLE IF NOT EXISTS contact_tags (
contact_id INTEGER,
tag_id INTEGER,
PRIMARY KEY (contact_id, tag_id),
FOREIGN KEY (contact_id) REFERENCES contacts (id),
FOREIGN KEY (tag_id) REFERENCES tags (id)
)
""")
conn.commit()
conn.close()This setup will ensure that we have a structured way to store and retrieve our CRM data, allowing us to manage contacts, notes, and tags effectively.
Defining the CRM Service
The CRMService class is designed to handle all operations related to managing contacts, notes, and tags within our CRM system. It provides a straightforward interface for performing CRUD (Create, Read, Update, Delete) operations, ensuring that users can easily interact with their data.
Here’s an overview of the CRMService class and its responsibilities:
-
Database Connection: The service manages connections to the SQLite database, ensuring that all operations are performed in a consistent manner.
-
Adding Contacts: Users can add new contacts to the CRM. The service checks for existing contacts to prevent duplicates.
-
Updating Contacts: The service allows users to update the information of existing contacts.
-
Managing Notes: Users can add notes to contacts, which can be useful for keeping track of interactions or important information.
-
Tag Management: The service supports adding tags and associating them with contacts, enabling better organization and search capabilities.
-
Searching Contacts: Users can search for contacts based on names, notes, or tags, making it easy to find specific information.
Here’s the code for the CRMService class, which implements these functionalities:
@dataclass
class CRMService:
"""Service class for CRM operations."""
def __post_init__(self):
# Ensure DB exists
if not os.path.exists(DB_PATH):
init_db()
def get_conn(self):
"""Get a database connection."""
return sqlite3.connect(DB_PATH)
def add_contact(self, name: str) -> int:
"""Add a new contact to the CRM."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute("INSERT INTO contacts (name) VALUES (?)", (name,))
conn.commit()
return cursor.lastrowid
finally:
conn.close()
def add_note(self, contact_id: int, content: str) -> int:
"""Add a note to a contact."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"INSERT INTO notes (contact_id, content) VALUES (?, ?)",
(contact_id, content),
)
conn.commit()
return cursor.lastrowid
finally:
conn.close()
def add_tag(self, name: str) -> int:
"""Add a tag if it doesn't exist and return its ID."""
conn = self.get_conn()
cursor = conn.cursor()
try:
# Try to get existing tag
cursor.execute("SELECT id FROM tags WHERE name = ?", (name,))
result = cursor.fetchone()
if result:
return result[0]
# Create new tag
cursor.execute("INSERT INTO tags (name) VALUES (?)", (name,))
conn.commit()
return cursor.lastrowid
finally:
conn.close()
def tag_contact(self, contact_id: int, tag_id: int) -> bool:
"""Associate a tag with a contact."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"INSERT OR IGNORE INTO contact_tags (contact_id, tag_id) VALUES (?, ?)",
(contact_id, tag_id),
)
conn.commit()
return cursor.rowcount > 0
finally:
conn.close()
def get_contact_by_name(self, name: str) -> Optional[Dict[str, Any]]:
"""Get a contact by name."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"SELECT id, name, created_at, updated_at FROM contacts WHERE name = ?",
(name,),
)
result = cursor.fetchone()
if not result:
return None
return {
"id": result[0],
"name": result[1],
"created_at": result[2],
"updated_at": result[3],
}
finally:
conn.close()
def update_contact(self, contact_id: int, name: str) -> bool:
"""Update a contact's information."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"UPDATE contacts SET name = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(name, contact_id),
)
conn.commit()
return cursor.rowcount > 0
finally:
conn.close()
def get_note_by_id(self, note_id: int) -> Optional[Dict[str, Any]]:
"""Get a note by its ID."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"SELECT id, contact_id, content, created_at FROM notes WHERE id = ?",
(note_id,),
)
result = cursor.fetchone()
if not result:
return None
return {
"id": result[0],
"contact_id": result[1],
"content": result[2],
"created_at": result[3],
}
finally:
conn.close()
def update_note(self, note_id: int, content: str) -> bool:
"""Update a note's content."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"UPDATE notes SET content = ? WHERE id = ?", (content, note_id)
)
conn.commit()
return cursor.rowcount > 0
finally:
conn.close()
def get_all_contacts(self) -> List[Dict[str, Any]]:
"""Get all contacts."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute("SELECT id, name FROM contacts ORDER BY name")
return [{"id": row[0], "name": row[1]} for row in cursor.fetchall()]
finally:
conn.close()
def get_contact_notes(self, contact_id: int) -> List[Dict[str, Any]]:
"""Get all notes for a contact."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"SELECT id, content, created_at FROM notes WHERE contact_id = ? ORDER BY created_at DESC",
(contact_id,),
)
return [
{"id": row[0], "content": row[1], "created_at": row[2]}
for row in cursor.fetchall()
]
finally:
conn.close()
def get_contact_tags(self, contact_id: int) -> List[str]:
"""Get all tags for a contact."""
conn = self.get_conn()
cursor = conn.cursor()
try:
cursor.execute(
"""
SELECT t.name
FROM tags t
JOIN contact_tags ct ON t.id = ct.tag_id
WHERE ct.contact_id = ?
ORDER BY t.name
""",
(contact_id,),
)
return [row[0] for row in cursor.fetchall()]
finally:
conn.close()
def search_contacts(self, query: str = "", tags: str = "") -> List[Dict[str, Any]]:
"""
Search contacts by name, notes content, or tags.
Args:
query: Search term for name or notes
tags: Comma-separated list of tags to filter by
"""
conn = self.get_conn()
cursor = conn.cursor()
try:
if not query and not tags:
return self.get_all_contacts()
# Base query to get distinct contacts
sql = """
SELECT DISTINCT c.id, c.name
FROM contacts c
"""
params = []
where_clauses = []
# Add join for notes if query is present
if query:
sql += " LEFT JOIN notes n ON c.id = n.contact_id"
where_clauses.append("(c.name LIKE ? OR n.content LIKE ?)")
search_term = f"%{query}%"
params.extend([search_term, search_term])
# Add joins and conditions for tags if tags are present
if tags:
tag_list = [t.strip() for t in tags.split(",")]
if tag_list:
sql += """
JOIN contact_tags ct ON c.id = ct.contact_id
JOIN tags t ON ct.tag_id = t.id
"""
tag_placeholders = ", ".join(["?" for _ in tag_list])
where_clauses.append(f"t.name IN ({tag_placeholders})")
params.extend(tag_list)
# Add WHERE clause if we have conditions
if where_clauses:
sql += " WHERE " + " AND ".join(where_clauses)
sql += " ORDER BY c.name"
cursor.execute(sql, params)
return [{"id": row[0], "name": row[1]} for row in cursor.fetchall()]
finally:
conn.close()This class serves as the backbone of our CRM application, providing all the necessary methods to interact with the database and manage user data effectively. By encapsulating the database operations within this service, we maintain a clean separation of concerns, making the codebase easier to manage and extend in the future.
Creating the CRM Agent
To create the CRM agent, we’ll instantiate an Agent, which allows us to define a stateless agent that can handle various CRM operations. The agent is designed to assist users in managing their contacts, notes, and tags effectively.
Here are some of the capabilities of Agents in Pydantic:
They consist of several main components:
-
System Prompts: These are instructions written by developers that guide the LLM on how to behave and what tasks to perform. System prompts help establish the agent’s role and capabilities.
-
Function Tools: These are callable functions that the LLM can use to retrieve information during response generation. Tools provide the agent with the ability to interact with external systems and data sources.
-
Structured Output Types: Agents can be configured to return specific structured data types at the end of their runs. This ensures responses follow a predictable format.
-
Dependency Type Constraints: System prompts, tools, and output validators can utilize dependencies during execution, allowing for flexible and modular design.
-
LLM Model Selection: Agents can be associated with a default LLM model, though this can be overridden at runtime. This allows for flexibility in choosing the appropriate model for different scenarios.
-
Model Settings: Fine-tuning parameters can be configured as defaults or specified during runtime to optimize the agent’s performance.
The combination of these components creates a powerful framework for building AI applications that are both flexible and maintainable.
The following code snippet illustrates how to create the CRM agent:
# Create the CRM agent
crm_agent = Agent[CRMService, Any](
"openai:gpt-4o", # You can change this to your preferred model
deps_type=CRMService,
system_prompt="""
You are a CRM (Customer Relationship Management) assistant.
You help users manage their contacts, notes, and tags.
You should use the available tools to:
1. Add and update contacts
2. Add and update notes for contacts
3. Add tags to contacts
4. Search for contacts by name, notes, or tags
Always confirm actions you've taken and provide helpful summaries.
Include the date in new notes when appropriate.
""",
)This agent is equipped with a system prompt that outlines its capabilities, ensuring that it can assist users in adding and updating contacts, managing notes, and searching for contacts based on different criteria. By leveraging dependency injection, the agent maintains a clean separation of concerns, allowing it to operate without retaining state between requests.
In summary, the creation of the CRM agent is a crucial step in building our personal CRM system, enabling us to interact with the underlying data and perform various operations seamlessly.
Creating tools
To enhance the functionality of our CRM system, we will create several tools that allow users to perform various operations seamlessly. Each tool is defined as an asynchronous function that interacts with the CRMService class. Below are the code examples for the different functionalities:
First, we define a tool to add a new contact:
@crm_agent.tool
async def add_contact(
ctx: RunContext[CRMService], name: str, tags: str = "", notes: str = ""
) -> str:
"""
Add a new contact to the CRM. Returns a status message.
Args:
name: Name of the contact
tags: Comma-separated list of tags (optional)
notes: Initial notes about the contact (optional)
"""
try:
# Check if contact already exists
existing = ctx.deps.get_contact_by_name(name)
if existing:
return f"Contact '{name}' already exists with ID {existing['id']}"
# Add the contact
contact_id = ctx.deps.add_contact(name)
# Add tags if provided
if tags:
tag_list = [t.strip() for t in tags.split(",")]
for tag in tag_list:
if tag:
tag_id = ctx.deps.add_tag(tag)
ctx.deps.tag_contact(contact_id, tag_id)
# Add notes if provided
if notes:
ctx.deps.add_note(contact_id, notes)
return f"Contact '{name}' added successfully with ID {contact_id}"
except Exception as e:
return f"Error adding contact: {str(e)}"Next, we create a tool for updating a contact’s information:
@crm_agent.tool
async def update_contact(
ctx: RunContext[CRMService], contact_name: str, new_name: str
) -> str:
"""
Update a contact's name. Returns a status message.
Args:
contact_name: Current name of the contact
new_name: New name for the contact
"""
try:
contact = ctx.deps.get_contact_by_name(contact_name)
if not contact:
return f"Contact '{contact_name}' not found"
# Check if new name already exists
if contact_name != new_name and ctx.deps.get_contact_by_name(new_name):
return f"Cannot update: Contact '{new_name}' already exists"
success = ctx.deps.update_contact(contact["id"], new_name)
if success:
return f"Contact updated from '{contact_name}' to '{new_name}' (ID: {contact['id']})"
else:
return f"Failed to update contact '{contact_name}'"
except Exception as e:
return f"Error updating contact: {str(e)}"We also need a tool to add notes to a contact:
@crm_agent.tool
async def add_note(ctx: RunContext[CRMService], contact_name: str, note: str) -> str:
"""
Add a note to an existing contact. Returns a status message.
Args:
contact_name: Name of the contact
note: Note text to add
"""
try:
contact = ctx.deps.get_contact_by_name(contact_name)
if not contact:
return f"Contact '{contact_name}' not found"
# Add the date to the note if it doesn't already have one
if not any(
word in note.lower() for word in ["today", "yesterday", "date:", "date is"]
):
today = datetime.now().strftime("%Y-%m-%d")
note = f"Date: {today}\\n{note}"
note_id = ctx.deps.add_note(contact["id"], note)
return f"Note (ID: {note_id}) added to contact '{contact_name}'"
except Exception as e:
return f"Error adding note: {str(e)}"Additionally, we can list all notes for a contact:
@crm_agent.tool
async def list_contact_notes(ctx: RunContext[CRMService], contact_name: str) -> str:
"""
List all notes for a contact with their IDs. Returns a formatted string.
Args:
contact_name: Name of the contact
"""
contact = ctx.deps.get_contact_by_name(contact_name)
if not contact:
return f"Contact '{contact_name}' not found."
notes = ctx.deps.get_contact_notes(contact["id"])
if not notes:
return f"No notes found for contact '{contact_name}'."
output = f"Notes for {contact_name} (ID: {contact['id']}):\\n"
for note in notes:
output += f" - ID: {note['id']}, Created: {note['created_at']}\\n Content: {note['content']}\\n"
return outputWe can edit those notes…
@crm_agent.tool
async def edit_note_by_id(
ctx: RunContext[CRMService], note_id: int, new_note: str
) -> str:
"""
Edit a note using its ID. Returns a status message.
Args:
note_id: ID of the note to edit
new_note: New text for the note
"""
try:
note = ctx.deps.get_note_by_id(note_id)
if not note:
return f"Note with ID {note_id} not found"
success = ctx.deps.update_note(note_id, new_note)
if success:
return f"Note with ID {note_id} updated successfully"
else:
return f"Failed to update note with ID {note_id}"
except Exception as e:
return f"Error updating note: {str(e)}"We can add tags to help organize our data.
@crm_agent.tool
async def add_tags_to_contact(
ctx: RunContext[CRMService], contact_name: str, tags: str
) -> str:
"""
Add tags to a contact. Returns a status message.
Args:
contact_name: Name of the contact
tags: Comma-separated list of tags
"""
try:
contact = ctx.deps.get_contact_by_name(contact_name)
if not contact:
return f"Contact '{contact_name}' not found"
tag_list = [t.strip() for t in tags.split(",")]
added_tags = []
for tag in tag_list:
if tag:
tag_id = ctx.deps.add_tag(tag)
if ctx.deps.tag_contact(contact["id"], tag_id):
added_tags.append(tag)
if added_tags:
return f"Added tags [{', '.join(added_tags)}] to contact '{contact_name}' (ID: {contact['id']})"
else:
# Check if all tags already existed for the contact
current_tags = set(ctx.deps.get_contact_tags(contact["id"]))
if all(t in current_tags for t in tag_list if t):
return f"All specified tags already exist for contact '{contact_name}'."
else:
return f"No new tags were added to contact '{contact_name}'. Some might already exist."
except Exception as e:
return f"Error adding tags: {str(e)}"List tags…
@crm_agent.tool
async def list_tags(ctx: RunContext[CRMService]) -> str:
"""List all existing tags in the system. Returns a formatted string."""
conn = ctx.deps.get_conn()
cursor = conn.cursor()
try:
cursor.execute("SELECT name FROM tags ORDER BY name")
tags = [row[0] for row in cursor.fetchall()]
if not tags:
return "No tags found in the system."
return f"Existing tags:\\n- {', '.join(tags)}"
except Exception as e:
return f"Error listing tags: {str(e)}"
finally:
conn.close()We can also search our contacts by name or tag.
@crm_agent.tool
async def search_contacts(
ctx: RunContext[CRMService], query: str = "", tags: str = ""
) -> str:
"""
Search for contacts by name, notes, or tags. Returns a formatted string.
Args:
query: Search term for name or notes (optional)
tags: Comma-separated list of tags to filter by (optional)
"""
try:
contacts = ctx.deps.search_contacts(query, tags)
count = len(contacts)
if count == 0:
search_criteria = []
if query:
search_criteria.append(f"query '{query}'")
if tags:
search_criteria.append(f"tags '{tags}'")
return f"No contacts found matching {' and '.join(search_criteria)}."
output = f"Found {count} contact(s):\\n"
for contact in contacts:
output += f" - ID: {contact['id']}, Name: {contact['name']}\\n"
return output
except Exception as e:
return f"Error searching contacts: {str(e)}"What I do like is how straightforward it is to define the tools. No magic, little boilerplate, relatively simple types / type transfers.
Running the CRM Agent
To run the CRM agent, we will implement a command-line interface (CLI) that allows users to interact with the system in a natural language format. This interface will enable users to perform various operations such as adding contacts, updating information, and managing notes and tags.
The CLI will continuously prompt the user for input until they decide to exit. Each user request will be processed by the CRM agent, which will handle the logic for interacting with the database and returning appropriate responses.
Here’s the code for the CLI interface:
# CLI interface
def run_cli():
"""Run the CRM agent as a command-line interface."""
crm_service = CRMService()
message_history: List[ModelMessage] = []
print("Starting CRM Agent...")
print("Type your requests in natural language. Type 'exit' to quit.")
print()
user_input = ""
while user_input.lower() not in ["exit", "quit"]:
user_input = input("\nWhat would you like to do? > ")
if user_input.lower() in ["exit", "quit"]:
break
# Run the agent synchronously, passing the history
result = crm_agent.run_sync(
user_input, deps=crm_service, message_history=message_history
)
print("\n" + result.output)
# Update history with all messages from the run
message_history = result.all_messages()Here’s what it will look like:
In this code, we initialize the CRMService and set up a loop that prompts the user for input. The agent processes the input using the run_sync method, which allows it to handle requests synchronously (async is also available). The results are printed to the console, and the message history is updated after each interaction.
Finally, we can run the CLI by calling the run_cli function in the main block of our application:
if __name__ == "__main__":
# Initialize the database
init_db()
# Run the CLI
run_cli()This setup ensures that the CRM agent is ready to handle user requests effectively, providing a user-friendly interface for managing contacts and their associated information.
Conclusion
In conclusion, the personal CRM system we built using PydanticAI showcases the power and flexibility of this framework.
By leveraging SQLite for our database, we created a robust backend that supports essential CRUD operations, allowing users to add, update, and search for contacts seamlessly.
The stateless nature of the CRM agent, facilitated by dependency injection, ensures that the system remains efficient and easy to maintain. The memory is a little funky, but the fact that it’s clear that I’m responsible for managing it is pretty nice. I’m not sure that I want to cede that to a framework at this point.
As we explored various functionalities, such as adding notes and managing tags, it became evident that PydanticAI offers a solid foundation for building complex applications with minimal overhead. The ability to define clear interfaces and utilize tools for specific tasks further streamlines the development process.
Looking ahead, there are numerous opportunities for enhancing this CRM system. Potential improvements could include integrating more advanced search capabilities, implementing user authentication, or expanding the system to support additional data types. Overall, this project has provided valuable insights into the capabilities of PydanticAI and its applicability in real-world scenarios.
We hope this article inspires you to explore building your own applications using PydanticAI or similar frameworks, and to consider the potential of personal CRMs in managing your relationships and information effectively.