Part 6: Create the agent project
The ingestion workflow gives us a populated podcasts index. Now we create
a separate agent/ project that searches this index and answers questions
from the podcast archive.
From the workshop root:
mkdir agent
cd agent
uv init --python=3.13
Install the dependencies:
uv add pydantic-ai openai elasticsearch
uv add --dev jupyter
Create .env with your OpenAI API key:
OPENAI_API_KEY=...
Start Jupyter:
uv run jupyter notebook
Create agent.ipynb. The notebook starts with tools and a small agent, then
we move the working code into tools.py and agent.py.
Search tools
Connect to Elasticsearch:
from elasticsearch import Elasticsearch
es = Elasticsearch("http://localhost:9200")
The first tool searches titles and subtitles. Pydantic AI reads the docstring to describe the tool to the model:
def search_videos(query: str, size: int = 5) -> list[dict]:
"""
Search for videos whose titles or subtitles match a given query.
Returns highlighted match information including video IDs and snippets.
"""
body = {
"size": size,
"query": {
"multi_match": {
"query": query,
"fields": ["title^3", "subtitles"],
"type": "best_fields",
"analyzer": "english_with_stop_and_stem"
}
},
Add snippet output through the Elasticsearch highlight field:
"highlight": {
"pre_tags": ["*"],
"post_tags": ["*"],
"fields": {
"title": {"fragment_size": 150, "number_of_fragments": 1},
"subtitles": {"fragment_size": 150, "number_of_fragments": 1}
}
}
}
response = es.search(index="podcasts", body=body)
hits = response.body['hits']['hits']
Return the video ID with the snippets:
results = []
for hit in hits:
highlight = hit['highlight']
highlight['video_id'] = hit['_id']
results.append(highlight)
return results
The second tool retrieves the full transcript. The pattern is like web search: first look at snippets, then open the full document:
def get_subtitles_by_id(video_id: str) -> dict:
"""
Retrieve the full subtitle content for a specific video.
"""
result = es.get(index="podcasts", id=video_id)
return result['_source']
Test search before adding the model:
search_videos('how do I get into machine learning?')
First agent
Create a simple Pydantic AI agent:
from pydantic_ai import Agent
research_instructions = """
You're a helpful researcher agent.
""".strip()
research_agent = Agent(
name='research_agent',
instructions=research_instructions,
model='openai:gpt-4o-mini',
tools=[search_videos, get_subtitles_by_id]
)
Run it:
result = await research_agent.run(
user_prompt='how do I get rich with AI?'
)
print(result.output)
The first version can answer without using tools. That is not what we want. Look at the messages to see whether tool calls happened:
messages = result.new_messages()
for m in messages:
for p in m.parts:
kind = p.part_kind
if kind == 'user-prompt':
print('USER:', p.content)
print()
if kind == 'text':
print('ASSISTANT:', p.content)
print()
if kind == 'tool-call':
print('TOOL CALL:', p.tool_name, p.args)
Better research instructions
Give the agent a more explicit research process:
research_instructions = """
You are an autonomous research agent. Your goal is to perform deep, multi-stage research on the
given topic using the available search function.
Research process:
Stage 1: Initial Exploration
- Using your own knowledge of the topic, perform 3-5 broad search queries to understand the main topic
and identify related areas. Only use search function.
- After the initial search exploration, summarize key concepts, definitions, and major themes.
- You MUST read the full transcript to be able to provide a better answer for the user.
Stage 2: Deep Investigation
- Perform 5-6 refined queries focusing on depth.
- Look at relevant documents for specific mechanisms, case studies, and technical details.
- Gather diverse viewpoints and data to strengthen depth and accuracy.
""".strip()
Add the citation and output rules:
research_instructions += """
Rules:
1. Search queries:
- Do not include years unless explicitly requested by the user.
- Use timeless, concept-based queries.
2. Each paragraph must have at least one reference object containing:
- video_id
- timestamp
- quote
3. Do not fabricate data or sources.
4. Output format: Markdown
""".strip()
Tool-call logging
During development, print each tool call as it happens. This shows whether the agent is using the search backend:
from pydantic_ai.messages import FunctionToolCallEvent
class NamedCallback:
"""Stream handler that prints the tool calls triggered by an agent."""
def __init__(self, agent: Agent):
self.agent_name = agent.name
Handle nested streams and function-call events:
async def _print_function_calls(self, ctx, event) -> None:
if hasattr(event, "__aiter__"):
async for sub_event in event:
await self._print_function_calls(ctx, sub_event)
return
if isinstance(event, FunctionToolCallEvent):
tool_name = event.part.tool_name
args = event.part.args
print(f"TOOL CALL ({self.agent_name}): {tool_name}({args})")
async def __call__(self, ctx, event) -> None:
await self._print_function_calls(ctx, event)
Run the agent with the callback:
research_agent_callback = NamedCallback(research_agent)
result = await research_agent.run(
user_prompt='how do I get rich with AI?',
event_stream_handler=research_agent_callback
)
print(result.output)
The next problem is context size. Full podcast transcripts can be long. In Part 7: Summarize long transcripts we add a summarization tool so the research agent can read full episodes without putting every subtitle line into its own context.