Developing Multi-Agent Applications: A Deep Dive into AI Agent Frameworks
In the rapidly evolving landscape of Artificial Intelligence, multi-agent systems have emerged as a powerful approach to managing complex workflows. This blog explores the intricacies of building multi-agent applications using the Pydantic AI Agent framework, with a focus on Agent Delegation and sophisticated workflow management. Understanding AI Agent Levels or Architectures Multi-agent systems are not created equal. We can categorize AI agents into several sophisticated levels: Single Agent: The most basic form, similar to a simple chatbot interacting with a Large Language Model (LLM) to perform tasks or respond to queries. Agent Delegation: A more advanced setup where one agent delegates tasks to another using specialized tools, creating a more structured and efficient workflow. Supervisor Agent: Drawing inspiration from the SAGA Orchestration Pattern, this agent strategically determines the next agent to invoke in the workflow. [Coordination] Hierarchy of Supervisors: Mirroring corporate organizational structures, this approach includes: Top-level supervisor (CEO equivalent) Intermediate supervisors (Department Heads) Individual agents handling specialized tasks (Team members) Practical Example: Resume Processing Multi-Agent Application Let's dive into a practical example of a multi-agent application designed to extract information from resumes and store it in a database. This application features two primary agents: a Resume Parser Agent and a Candidate Profile Creator Agent. Database Setup We'll use SQLAlchemy ORM for a code-first approach to database schema design, with SQLite as the backend. Here's how we set up our database: # database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base DB_URL = 'sqlite:///./test.db' engine = create_engine(DB_URL, connect_args={'check_same_thread': False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() Defining the Candidate Model Our Candidate model captures essential candidate information: # models.py from sqlalchemy import Column, Integer, String from database import Base class Candidate(Base): __tablename__ = 'candidates' id = Column(Integer, primary_key=True, index=True) first_name = Column(String) last_name = Column(String) email = Column(String) phone = Column(String) skills = Column(String) experience = Column(Integer) title = Column(String) Agent Implementation Let's look at the core of our multi-agent system: # agents.py from dataclasses import dataclass from anthropic import BaseModel from pydantic_ai import Agent, RunContext from dotenv import load_dotenv from contextlib import contextmanager from database import SessionLocal, engine, Base from models import Candidate from pydantic import Field import asyncio load_dotenv() @contextmanager def get_db(): db = SessionLocal() try: yield db finally: db.close() class DatabaseConn: @classmethod async def get_candidate(cls, candidate_id: int) -> Candidate | None: with get_db() as db: candidate = db.query(Candidate).filter_by(id=candidate_id).first() return candidate @classmethod async def add_candidate(cls, candidate: Candidate) -> int: with get_db() as db: db.add(candidate) db.commit() db.refresh(candidate) return candidate.id @classmethod async def get_all_candidates(cls) -> list[Candidate]: with get_db() as db: candidates = db.query(Candidate).all() return candidates @dataclass class SupportDependencies: candidate_id: int | None db: DatabaseConn candidateInfo: Candidate class CandidateInfo(BaseModel): first_name: str = Field(description="First name of the candidate") last_name: str = Field(description="Last name of the candidate") email: str = Field(description="Email of the candidate") phone: str = Field(description="Phone of the candidate") skills: str = Field(description="Skills of the candidate") experience: int = Field(description="Experience of the candidate") title: str = Field(description="Title of the candidate") class ResultType(BaseModel): first_name: str = Field(description="First name of the candidate") last_name: str = Field(description="Last name of the candidate") email: str = Field(description="Email of the candidate") phone: str = Field(description="Phone of the candidate") resume_parser_agent = Agent(model="openai:gpt-4", result_type=CandidateInfo, system_prompt="You are a resume parser agent. You have to parse the resume and get the candidate details") agent = Agent(model="openai:gpt-4", deps_type=[SupportDependencies], result_type=list[ResultType], sy
In the rapidly evolving landscape of Artificial Intelligence, multi-agent systems have emerged as a powerful approach to managing complex workflows. This blog explores the intricacies of building multi-agent applications using the Pydantic AI Agent framework, with a focus on Agent Delegation and sophisticated workflow management.
Understanding AI Agent Levels or Architectures
Multi-agent systems are not created equal. We can categorize AI agents into several sophisticated levels:
Single Agent: The most basic form, similar to a simple chatbot interacting with a Large Language Model (LLM) to perform tasks or respond to queries.
Agent Delegation: A more advanced setup where one agent delegates tasks to another using specialized tools, creating a more structured and efficient workflow.
Supervisor Agent: Drawing inspiration from the SAGA Orchestration Pattern, this agent strategically determines the next agent to invoke in the workflow. [Coordination]
-
Hierarchy of Supervisors: Mirroring corporate organizational structures, this approach includes:
- Top-level supervisor (CEO equivalent)
- Intermediate supervisors (Department Heads)
- Individual agents handling specialized tasks (Team members)
Practical Example: Resume Processing Multi-Agent Application
Let's dive into a practical example of a multi-agent application designed to extract information from resumes and store it in a database. This application features two primary agents: a Resume Parser Agent and a Candidate Profile Creator Agent.
Database Setup
We'll use SQLAlchemy ORM for a code-first approach to database schema design, with SQLite as the backend. Here's how we set up our database:
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
DB_URL = 'sqlite:///./test.db'
engine = create_engine(DB_URL, connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Defining the Candidate Model
Our Candidate model captures essential candidate information:
# models.py
from sqlalchemy import Column, Integer, String
from database import Base
class Candidate(Base):
__tablename__ = 'candidates'
id = Column(Integer, primary_key=True, index=True)
first_name = Column(String)
last_name = Column(String)
email = Column(String)
phone = Column(String)
skills = Column(String)
experience = Column(Integer)
title = Column(String)
Agent Implementation
Let's look at the core of our multi-agent system:
# agents.py
from dataclasses import dataclass
from anthropic import BaseModel
from pydantic_ai import Agent, RunContext
from dotenv import load_dotenv
from contextlib import contextmanager
from database import SessionLocal, engine, Base
from models import Candidate
from pydantic import Field
import asyncio
load_dotenv()
@contextmanager
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
class DatabaseConn:
@classmethod
async def get_candidate(cls, candidate_id: int) -> Candidate | None:
with get_db() as db:
candidate = db.query(Candidate).filter_by(id=candidate_id).first()
return candidate
@classmethod
async def add_candidate(cls, candidate: Candidate) -> int:
with get_db() as db:
db.add(candidate)
db.commit()
db.refresh(candidate)
return candidate.id
@classmethod
async def get_all_candidates(cls) -> list[Candidate]:
with get_db() as db:
candidates = db.query(Candidate).all()
return candidates
@dataclass
class SupportDependencies:
candidate_id: int | None
db: DatabaseConn
candidateInfo: Candidate
class CandidateInfo(BaseModel):
first_name: str = Field(description="First name of the candidate")
last_name: str = Field(description="Last name of the candidate")
email: str = Field(description="Email of the candidate")
phone: str = Field(description="Phone of the candidate")
skills: str = Field(description="Skills of the candidate")
experience: int = Field(description="Experience of the candidate")
title: str = Field(description="Title of the candidate")
class ResultType(BaseModel):
first_name: str = Field(description="First name of the candidate")
last_name: str = Field(description="Last name of the candidate")
email: str = Field(description="Email of the candidate")
phone: str = Field(description="Phone of the candidate")
resume_parser_agent = Agent(model="openai:gpt-4", result_type=CandidateInfo,
system_prompt="You are a resume parser agent. You have to parse the resume and get the candidate details")
agent = Agent(model="openai:gpt-4", deps_type=[SupportDependencies], result_type=list[ResultType],
system_prompt=("You are a SaaS Candidate agent help users with their queries. You have access to the database to add and get the candidate details. "))
# Agent tools and prompts are defined here...
Main Application Logic
Here's how we tie everything together:
# main.py
from Resume_parser_agent import DatabaseConn, SupportDependencies, ResultType, agent, resume_parser_agent
import asyncio
from models import Candidate
async def main():
resume_text = "Sri Krishna Solution Architect | Email: sri.krishna@provizient.com | Phone: 223-456-7890 | Experienced Solution Architect with 10 years in software development and system design, specializing in Python, SQL, and cloud architectures (AWS, Azure). Skilled in microservices, REST APIs, Docker, Kubernetes, and CI/CD pipelines, with a strong background in designing scalable and resilient enterprise applications. Adept at leading cross-functional teams, optimizing database performance, and integrating AI/ML solutions into business applications. Passionate about leveraging modern technologies to drive digital transformation and align technical solutions with business objectives."
result1 = await resume_parser_agent.run(resume_text)
candidate = Candidate(first_name=result1.data.first_name, last_name=result1.data.last_name, email=result1.data.email, phone=result1.data.phone, skills=result1.data.skills, experience=result1.data.experience, title=result1.data.title)
add_deps = SupportDependencies(candidate_id="", db=DatabaseConn(), candidateInfo=candidate)
result_candidate_added = await agent.run("Add candidate details", deps=add_deps)
print(result_candidate_added.data)
deps = SupportDependencies(candidate_id="", db=DatabaseConn(), candidateInfo=None)
result = await agent.run("Get all candidate details", deps=deps)
print(result.data)
if __name__ == "__main__":
asyncio.run(main())
Advanced Possibilities
While our example focuses on resume processing, multi-agent architectures can be extended to:
- Automated candidate classification
- Job posting matching
- Advanced natural language processing for resume insights
Conclusion
Multi-agent AI systems represent a transformative approach to building intelligent, automated applications. By enabling sophisticated task delegation and workflow management, these frameworks unlock unprecedented potential for intelligent decision-making and process automation.
As AI technologies continue to evolve, multi-agent architectures will play an increasingly critical role in developing complex, adaptive software solutions. The example provided here is just the tip of the iceberg, demonstrating the power and flexibility of these systems in handling real-world tasks efficiently and intelligently.
Thanks
Sreeni Ramadorai.
