Agentic AI frameworks

CrewAI, Langgraph, LlamaIndex workflow, Pydantic AI, OpenAI swarm, Huggingface Smolagents

Xin Cheng
13 min readFeb 28, 2025

Quick experiment of popular open-source Agentic AI frameworks.

TLDR

  1. openai swarm is educational purpose only
  2. Langgraph, CrewAI are most popular and complement each other
  3. Pydantic AI focuses on type safety
  4. Huggingface smolagents coder agent uses Python code which has benefit for code-solvable problem
  5. LlamaIndex workflow has low-level event-based model, but AgentWorkflow paradigm is similar to CrewAI

Agent: building block of agentic AI, with each agent specializes in performing a task

Workflow control/Handoff: how agents work together (e.g. for complex task, which agent to break down it to smaller steps, which agent processes which step, sequence, how these decisions are made, etc.)

Tool: including access to external knowledge; interaction module with external environment; guardrail

CrewAI Langgraph

There is no magic in this integration. Basically langgraph each node maps to a function, in this function, it can retrieve any state (e.g. customer question) from the engine and then process state using CrewAI. This way, it can integrate with Autogen, Pydantic AI, LlamaIndex.

langgraph node definition, when engine calls “draft_responses” node, it calls the function “EmailFilterCrew().kickoff”.

workflow.add_node("draft_responses", EmailFilterCrew().kickoff)

The function basically processes state and call CrewAI method, and pass back new state which langgraph engine will pass to next node

def kickoff(self, state):
print("### Filtering emails")
tasks = EmailFilterTasks()
crew = Crew(
agents=[self.filter_agent, self.action_agent, self.writer_agent],
tasks=[
tasks.filter_emails_task(self.filter_agent, self._format_emails(state['emails'])),
tasks.action_required_emails_task(self.action_agent),
tasks.draft_responses_task(self.writer_agent)
],
verbose=True
)
result = crew.kickoff()
return {**state, "action_required_emails": result}

Use CrewAI vision tool to analyze image_url

LlamaIndex

Workflow (step function can respond to multiple events and output multiple events)

Sub question as workflow notebook

Define tool (most common is FunctionTool)

from llama_index.core.tools import FunctionTool


def get_weather(location: str) -> str:
"""Usfeful for getting the weather for a given location."""
...


tool = FunctionTool.from_defaults(
get_weather,
# async_fn=aget_weather, # optional!
)

agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)

New AgentWorkflow is supported for agentic AI. Single agent

Define tool (no specific annotation, just function)

from duckduckgo_search import DDGS
from llama_index.core.workflow import Context


async def search_web(query: str) -> str:
"""Useful for using the web to answer questions."""
client = DDGS()
return str(client.text(query, max_results=1))


async def record_notes(ctx: Context, notes: str, notes_title: str) -> str:
"""Useful for recording notes on a given topic. Your input should be notes with a title to save the notes under."""
current_state = await ctx.get("state")
if "research_notes" not in current_state:
current_state["research_notes"] = {}
current_state["research_notes"][notes_title] = notes
await ctx.set("state", current_state)
return "Notes recorded."


async def write_report(ctx: Context, report_content: str) -> str:
"""Useful for writing a report on a given topic. Your input should be a markdown formatted report."""
current_state = await ctx.get("state")
current_state["report_content"] = report_content
await ctx.set("state", current_state)
return "Report written."


async def review_report(ctx: Context, review: str) -> str:
"""Useful for reviewing a report and providing feedback. Your input should be a review of the report."""
current_state = await ctx.get("state")
current_state["review"] = review
await ctx.set("state", current_state)
return "Report reviewed."

Define agents (web search, writer, reviewer). Each agent specify which other agent it can handoff to (should be defined on workflow level or send event, each agent should complete its work and hand back to a central assembly line, the main supervisor decides how to route to next agent or other agents decide which event the respond).


from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini")

from llama_index.core.agent.workflow import FunctionAgent, ReActAgent

research_agent = FunctionAgent(
name="ResearchAgent",
description="Useful for searching the web for information on a given topic and recording notes on the topic.",
system_prompt=(
"You are the ResearchAgent that can search the web for information on a given topic and record notes on the topic. "
"Once notes are recorded and you are satisfied, you should hand off control to the WriteAgent to write a report on the topic. "
"You should have at least some notes on a topic before handing off control to the WriteAgent."
),
llm=llm,
tools=[search_web, record_notes],
can_handoff_to=["WriteAgent"],
)

write_agent = FunctionAgent(
name="WriteAgent",
description="Useful for writing a report on a given topic.",
system_prompt=(
"You are the WriteAgent that can write a report on a given topic. "
"Your report should be in a markdown format. The content should be grounded in the research notes. "
"Once the report is written, you should get feedback at least once from the ReviewAgent."
),
llm=llm,
tools=[write_report],
can_handoff_to=["ReviewAgent", "ResearchAgent"],
)

review_agent = FunctionAgent(
name="ReviewAgent",
description="Useful for reviewing a report and providing feedback.",
system_prompt=(
"You are the ReviewAgent that can review the write report and provide feedback. "
"Your review should either approve the current report or request changes for the WriteAgent to implement. "
"If you have feedback that requires changes, you should hand off control to the WriteAgent to implement the changes after submitting the review."
),
llm=llm,
tools=[review_report],
can_handoff_to=["WriteAgent"],
)

Define workflow with initial state and run

from llama_index.core.agent.workflow import AgentWorkflow

agent_workflow = AgentWorkflow(
agents=[research_agent, write_agent, review_agent],
root_agent=research_agent.name,
initial_state={
"research_notes": {},
"report_content": "Not written yet.",
"review": "Review required.",
},
)

from llama_index.core.agent.workflow import (
AgentInput,
AgentOutput,
ToolCall,
ToolCallResult,
AgentStream,
)

handler = agent_workflow.run(
user_msg=(
"Write me a report on the history of the internet. "
"Briefly describe the history of the internet, including the development of the internet, the development of the web, "
"and the development of the internet in the 21st century."
)
)

current_agent = None
current_tool_calls = ""
async for event in handler.stream_events():
if (
hasattr(event, "current_agent_name")
and event.current_agent_name != current_agent
):
current_agent = event.current_agent_name
print(f"\n{'='*50}")
print(f"🤖 Agent: {current_agent}")
print(f"{'='*50}\n")

# use AgentStream for LLM output
# if isinstance(event, AgentStream):
# if event.delta:
# print(event.delta, end="", flush=True)
# elif isinstance(event, AgentInput):
# print("📥 Input:", event.input)
elif isinstance(event, AgentOutput):
if event.response.content:
print("📤 Output:", event.response.content)
if event.tool_calls:
print(
"🛠️ Planning to use tools:",
[call.tool_name for call in event.tool_calls],
)
elif isinstance(event, ToolCallResult):
print(f"🔧 Tool Result ({event.tool_name}):")
print(f" Arguments: {event.tool_kwargs}")
print(f" Output: {event.tool_output}")
elif isinstance(event, ToolCall):
print(f"🔨 Calling Tool: {event.tool_name}")
print(f" With arguments: {event.tool_kwargs}")

Pydantic AI

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.usage import UsageLimits
from pydantic import BaseModel
from datetime import datetime
from dotenv import load_dotenv
import asyncio

model_name = "gpt-4o-mini"

# Configure OpenRouter API with OpenAI-compatible base URL
model = OpenAIModel(
model_name=model_name, # Specify the desired model
)

Define input, output structure and agents (associating tool function with specific agent is not intuitive, what if multiple agents need to use same tools)

class GetCurrentDateInput(BaseModel):
"""No inputs required for current date"""
pass
class GetCurrentDateOutput(BaseModel):
"""Response for getting current date"""
current_date: str
class GetWeatherInput(BaseModel):
"""Input for getting weather"""
city: str
class GetWeatherOutput(BaseModel):
"""Response for getting weather"""
weather: str
temperature: float

# Router Agent
router_agent = Agent(
model=model,
system_prompt="""You are a helpful assistant. You have access to other agents who can help you answer questions. You can call them by calling their tools. \
1. Assess which agent you should use to answer the question. \
2. If you think the question is too complex or not relevant, respond with 'I don't know how to help you with that'. \
3. Use call_get_current_date_agent tool to reach out to current date agent. \
4. Use call_get_weather_agent tool to reach out to weather agent. Provide the city. \
Finally, respond once you have a final answer.""",
)
# Current Date Agent
current_date_agent = Agent(
model=model,
system_prompt="""You are a date expert. Get the current date. \
1. Use get_current_date tool to get the current date. \
2. If you think the question is too complex or not relevant, respond with 'I don't know how to help you with that'. \
Finally, respond once you have a final answer.""",
)
# Weather Agent
weather_agent = Agent(
model=model,
system_prompt="""You are a weather expert. Get the weather for a specified city. \
1. Use get_weather tool to get the weather for a specified city. \
2. If you think the question is too complex or not relevant, respond with 'I don't know how to help you with that'. \
Finally, respond once you have a final answer.""",
)

# Tool 1: Get Current Date
@current_date_agent.tool
def get_current_date(_: RunContext[GetCurrentDateInput]) -> GetCurrentDateOutput:
print("Getting current date...")
current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return GetCurrentDateOutput(current_date=current_date)

# Tool 2: Get Weather
@weather_agent.tool
def get_weather(_: RunContext[GetWeatherInput], city: str) -> GetWeatherOutput:
print(f"Received city: {city}")
if not city:
raise ValueError("City is missing!")
# Simulated weather data
weather = "Sunny"
temperature = 24.5
return GetWeatherOutput(weather=weather, temperature=temperature)

Call agent (nest_asyncio is for running in Databricks notebook)

import nest_asyncio
nest_asyncio.apply()

async def main():
result = await router_agent.run("What is the date today and what is the weather in New York?", usage_limits=UsageLimits(request_limit=3))
print(result.data)
# print(result.all_messages())


# Run the main function
asyncio.run(main())

The response only prints date agent response

[ModelRequest(parts=[SystemPromptPart(content="You are a helpful assistant. You have access to other agents who can help you answer questions. You can call them by calling their tools.         1. Assess which agent you should use to answer the question.         2. If you think the question is too complex or not relevant, respond with 'I don't know how to help you with that'.         3. Use current_date_agent tool to reach out to current date agent.         4. Use get_weather tool to reach out to weather agent. Provide the city.         Finally, respond once you have a final answer.", dynamic_ref=None, part_kind='system-prompt'), UserPromptPart(content='What is the date today and what is the weather in New York?', timestamp=datetime.datetime(2025, 2, 19, 5, 40, 41, 358276, tzinfo=datetime.timezone.utc), part_kind='user-prompt')], kind='request'), ModelResponse(parts=[TextPart(content="I will first check the current date and then the weather in New York.\n\nLet's start with the current date.", part_kind='text')], model_name='gpt-4o-mini-2024-07-18', timestamp=datetime.datetime(2025, 2, 19, 5, 40, 43, tzinfo=datetime.timezone.utc), kind='response')]

A rag agent example, external vector DB connection can be passed as dependency

from dataclasses import dataclass
from asyncpg import Pool

@dataclass
class Deps:
pool: Pool # Database connection pool

while vector search function can be decorated as agent tool, the dependency is made available in RunContext.deps parameter

from pydantic_ai import Agent

agent = Agent(
model="openai:gpt-4o", # Specify your LLM model
deps_type=Deps,
system_prompt="You are a helpful assistant answering questions using provided documentation.",
)

@agent.tool
async def search(ctx: RunContext[Deps], query: str) -> str:
"""Searches the documentation database."""
# Generate query embedding (use your LLM's embedding API)
query_embedding = await ctx.deps.pool.fetchval("SELECT embedding_function($1)", query)
return await search_docs(ctx, query_embedding)

async def main():
# Initialize dependencies (e.g., database connection pool)
async with database_connect() as pool:
deps = Deps(pool=pool)

# Ask a question using the agent
question = "How do I configure logfire with FastAPI?"
result = await agent.run(question, deps=deps)
print(result.data)

The programming model of Pydantic AI feels taking more time to get familiar than CrewAI, e.g. agent tool annotation. However, it focuses on Type safety which can benefit non-chatbot scenario, e.g. agent can specify result_type

agent = Agent(
'anthropic:claude-3-5-sonnet-latest',
retries=3,
result_type=NeverResultType,
system_prompt='Any time you get a response, call the `infinite_retry_tool` to produce another response.',
)

You can pass message history with

from pydantic_ai import Agent

agent = Agent(
'openai:gpt-4o-mini',
system_prompt='Be concise, reply with one sentence.',
)

result = agent.run_sync('Where does "hello world" come from?')
print(result.data)

# "Hello, World!" originated from the 1972 programming language tutorial for the B language, created by Brian Kernighan, and has since become a standard example for introducing programming concepts.
result2 = agent.run_sync('Explain more', message_history=result.all_messages())
print(result2.data)
# The phrase "Hello, World!" is often used as the first program written by beginners learning a new programming language, serving as a simple way to demonstrate the syntax and functionality of the language. Its use was popularized by Brian Kernighan in the book "The C Programming Language," co-authored with Dennis Ritchie, where it was used as an illustrative example of basic output. Over the years, it has become a cultural reference within the programming community, symbolizing the start of a programmer's journey.

Huggingface smolagents

babyagi, autogpt rely on too much on LLM to make decisions, while frameworks like langgraph control degree of autonomy. CodeAgent is inspired from papers like PAL: Program-aided Language Models. You can specify prompt and additional_authorized_imports to allow more sophisticated code.

  • TransformersModel takes a pre-initialized transformers pipeline to run inference on your local machine using transformers.
  • HfApiModel leverages a huggingface_hub.InferenceClient under the hood and supports all Inference Providers on the Hub.
  • LiteLLMModel similarly lets you call 100+ different models and providers through LiteLLM!
  • AzureOpenAIServerModel allows you to use OpenAI models deployed in Azure.
  • MLXModel creates a mlx-lm pipeline to run inference on your local machine.

Also comparison of CodeAgent vs. ToolCallingAgent

Minimal sample, download arxiv paper

Main agent is CodeAgent (good for technical domain where Python has advantage), which writes its actions in code, and can securely execute code in sandboxed environments via E2B. It works mostly like classical ReAct agents — the exception being that the LLM engine writes its actions as Python code snippets. On top of this CodeAgent class, we still support the standard ToolCallingAgent that writes actions as JSON/text blobs (good for API interaction, database queries). But we recommend always using CodeAgent. Import could be more organized, e.g. model under models, tool under tools sub-folders.

Define model

import os
from smolagents import OpenAIServerModel

model = OpenAIServerModel(
model_id="gpt-4o-mini",
)

Define tool

import arxiv
from smolagents import tool

@tool
def download_paper_by_id(paper_id: str) -> None:
"""
This tool gets the id of a paper and downloads it from arxiv. It saves the paper locally
in the current directory as "paper.pdf".

Args:
paper_id: The id of the paper to download.
"""
paper = next(arxiv.Client().results(arxiv.Search(id_list=[paper_id])))
paper.download_pdf(filename="paper.pdf")
return None

Above use tool annotation. We can also extend Tool base class.

Define agent and execute

from smolagents import CodeAgent

agent = CodeAgent(tools=[
download_paper_by_id],
model=model,
add_base_tools=True)

agent.run(
"download paper 2502.11211",
)

Output

╭──────────────────────────────────────────────────── New run ────────────────────────────────────────────────────╮
│ │
│ download paper 2502.11211 │
│ │
╰─ OpenAIServerModel - gpt-4o-mini ───────────────────────────────────────────────────────────────────────────────╯
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Step 1 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
─ Executing parsed code: ────────────────────────────────────────────────────────────────────────────────────────
download_paper_by_id(paper_id="2502.11211")
─────────────────────────────────────────────────────────────────────────────────────────────────────────────────
Out: None
[Step 0: Duration 3.40 seconds| Input tokens: 2,154 | Output tokens: 60]
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Step 2 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
─ Executing parsed code: ────────────────────────────────────────────────────────────────────────────────────────
final_answer("The paper with ID 2502.11211 has been downloaded successfully.")
─────────────────────────────────────────────────────────────────────────────────────────────────────────────────
Out - Final answer: The paper with ID 2502.11211 has been downloaded successfully.
[Step 1: Duration 1.36 seconds| Input tokens: 4,445 | Output tokens: 118]

Openai swarm

Basic concepts explained

The main Swarm client gets initial agent (central agent to determine which sub-agent to route to) and messages

response = client.run(agent=central_agent, messages=messages)

Central agent makes decision by prompt (e.g. there are 2 scenarios, RAG agent and NL2SQL agent), then use its functions property (which contains actual agent and comment about scenario)

model_name = "gpt-4o-mini"

# Define the Central Agent
central_agent = Agent(
name="Central Agent",
instructions="Determine if the query is about general company information (RAG) or a database query (NL2SQL), and route the query accordingly."
)

# Define handoff functions
def transfer_to_nl2sql():
print("Handing off to the NL2SQL Agent.")
"""Transfer the task to the NL2SQL Agent for database queries."""
return nl2sql_agent

def transfer_to_rag():
print("Handing off to the RAG agent.")
"""Transfer the task to the RAG Agent for general queries."""
return rag_agent

# Attach the handoff functions to the central agent
central_agent.functions = [transfer_to_nl2sql, transfer_to_rag]

The sub agent just defines its name, instruction and functions

rag_agent = Agent(
name="RAG Agent",
instructions="You retrieve relevant information from the company's knowledge base and generate responses to general queries about the company.",
functions=[retrieve_and_generate]
)

The function just accepts one parameter question

def retrieve_and_generate(question):
print("Calling retrieve_and_generate")
template = """Answer the question based only on the following context:
{context}
Question: {question}
Answer: """

prompt = ChatPromptTemplate.from_template(template)

def docs2str(docs):
return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
{"context": retriever | docs2str, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)

response = rag_chain.invoke(question)
return response

The loop scenario can only handle simple loop and just merely inspects messages variable to exit loop.

The stateless nature of Swarm has its advantages and limitations. On the positive side, it keeps interactions simple and lightweight, which is ideal for scenarios where you don’t need to maintain context over time. This makes Swarm particularly well-suited for quick experiments and educational purposes, where the focus is on understanding agent coordination rather than managing complex states. However, this also means that Swarm might not be the best choice for applications that require persistent memory or long-term context, as each interaction is independent.

openai agents sdk

# https://techcommunity.microsoft.com/blog/azure-ai-services-blog/use-azure-openai-and-apim-with-the-openai-agents-sdk/4392537
# https://community.openai.com/t/agents-sdk-with-azure-hosted-models/1157781
from openai import AsyncAzureOpenAI
from agents import set_default_openai_client
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv(".env")

# Create OpenAI client using Azure OpenAI
openai_client = AsyncAzureOpenAI(
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
azure_deployment=os.getenv("AZURE_OPENAI_GPT4O_DEPLOYMENT_NAME")
)

# Set the default OpenAI client for the Agents SDK
set_default_openai_client(openai_client)
import nest_asyncio
nest_asyncio.apply()
from agents import Agent, Runner, OpenAIChatCompletionsModel, set_tracing_disabled

from openai.types.chat import ChatCompletionMessageParam

# Disable tracing since we're using Azure OpenAI
set_tracing_disabled(disabled=True)

assistant = Agent(
name="Assistant",
instructions="You are a helpful assistant.",
model=OpenAIChatCompletionsModel(
model=os.getenv("AZURE_OPENAI_GPT4O_MODEL_NAME"),
openai_client=openai_client,
) # This will use the deployment specified in your Azure OpenAI/APIM client
)

from agents import Runner
import asyncio

async def main():
result = await Runner.run(
assistant,
input="Write a haiku about recursion in programming."
)

print(f"Response: {result.final_output}")

asyncio.run(main())

The question will trip guardrail as both question seems like homework

import nest_asyncio
nest_asyncio.apply()
from agents import Agent, Runner, OpenAIChatCompletionsModel, set_tracing_disabled

from openai.types.chat import ChatCompletionMessageParam

# Disable tracing since we're using Azure OpenAI
set_tracing_disabled(disabled=True)

model=OpenAIChatCompletionsModel(
model=os.getenv("AZURE_OPENAI_GPT4O_MODEL_NAME"),
openai_client=openai_client,
) # This will use the deployment specified in your Azure OpenAI/APIM client

history_tutor_agent = Agent(
name="History Tutor",
handoff_description="Specialist agent for historical questions",
instructions="You provide assistance with historical queries. Explain important events and context clearly.",
model=model
)

math_tutor_agent = Agent(
name="Math Tutor",
handoff_description="Specialist agent for math questions",
instructions="You provide help with math problems. Explain your reasoning at each step and include examples",
model=model
)

from agents import Agent, InputGuardrail,GuardrailFunctionOutput, Runner
from pydantic import BaseModel
import asyncio

class HomeworkOutput(BaseModel):
is_homework: bool
reasoning: str

guardrail_agent = Agent(
name="Guardrail check",
instructions="Check if the user is asking about homework.",
output_type=HomeworkOutput,
model=model
)

async def homework_guardrail(ctx, agent, input_data):
result = await Runner.run(guardrail_agent, input_data, context=ctx.context)
final_output = result.final_output_as(HomeworkOutput)
return GuardrailFunctionOutput(
output_info=final_output,
tripwire_triggered=not final_output.is_homework,
)

triage_agent = Agent(
name="Triage Agent",
instructions="You determine which agent to use based on the user's homework question",
handoffs=[history_tutor_agent, math_tutor_agent],
input_guardrails=[
InputGuardrail(guardrail_function=homework_guardrail),
],
model=model
)

from agents import Runner
import asyncio

async def main():
# both will trip guardrail
result = await Runner.run(triage_agent, "who was the first president of the united states?")
print(result.final_output)

result = await Runner.run(triage_agent, "what is life")
print(result.final_output)

asyncio.run(main())

--

--

Xin Cheng
Xin Cheng

Written by Xin Cheng

Generative Agentic AI/LLM, Data, ML, Multi/Hybrid-cloud, cloud-native, IoT developer/architect, 3x Azure-certified, 3x AWS-certified, 2x GCP-certified

No responses yet