On this tutorial, we construct a production-ready agentic workflow that prioritizes reliability over best-effort era by implementing strict, typed outputs at each step. We use PydanticAI to outline clear response schemas, wire in instruments through dependency injection, and make sure the agent can safely work together with exterior methods, comparable to a database, with out breaking execution. By operating every thing in a notebook-friendly, async-first setup, we display the best way to transfer past fragile chatbot patterns towards sturdy agentic methods appropriate for actual enterprise workflows.
!pip -q set up "pydantic-ai-slim[openai]" pydantic
import os, json, sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Literal, Elective, Checklist
from pydantic import BaseModel, Subject, field_validator
from pydantic_ai import Agent, RunContext, ModelRetry
if not os.environ.get("OPENAI_API_KEY"):
attempt:
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = (userdata.get("OPENAI_API_KEY") or "").strip()
besides Exception:
go
if not os.environ.get("OPENAI_API_KEY"):
import getpass
os.environ["OPENAI_API_KEY"] = getpass.getpass("Paste your OPENAI_API_KEY: ").strip()
assert os.environ.get("OPENAI_API_KEY"), "OPENAI_API_KEY is required."We arrange the execution surroundings and guarantee all required libraries can be found for the agent to run accurately. We securely load the OpenAI API key in a Colab-friendly means so the tutorial works with out handbook configuration adjustments. We additionally import all core dependencies that will likely be shared throughout schemas, instruments, and agent logic.
Precedence = Literal["low", "medium", "high", "critical"]
ActionType = Literal["create_ticket", "update_ticket", "query_ticket", "list_open_tickets", "no_action"]
Confidence = Literal["low", "medium", "high"]
class TicketDraft(BaseModel):
title: str = Subject(..., min_length=8, max_length=120)
buyer: str = Subject(..., min_length=2, max_length=60)
precedence: Precedence
class: Literal["billing", "bug", "feature_request", "security", "account", "other"]
description: str = Subject(..., min_length=20, max_length=1000)
expected_outcome: str = Subject(..., min_length=10, max_length=250)
class AgentDecision(BaseModel):
motion: ActionType
cause: str = Subject(..., min_length=20, max_length=400)
confidence: Confidence
ticket: Elective[TicketDraft] = None
ticket_id: Elective[int] = None
follow_up_questions: Checklist[str] = Subject(default_factory=listing, max_length=5)
@field_validator("follow_up_questions")
@classmethod
def short_questions(cls, v):
for q in v:
if len(q) > 140:
elevate ValueError("Every follow-up query should be <= 140 characters.")
return vWe outline the strict knowledge fashions that act because the contract between the agent and the remainder of the system. We use typed fields and validation guidelines to ensure that each agent response follows a predictable construction. By implementing these schemas, we forestall malformed outputs from silently propagating by means of the workflow.
@dataclass
class SupportDeps:
db: sqlite3.Connection
tenant: str
coverage: dict
def utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def init_db() -> sqlite3.Connection:
conn = sqlite3.join(":reminiscence:", check_same_thread=False)
conn.execute("""
CREATE TABLE tickets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tenant TEXT NOT NULL,
title TEXT NOT NULL,
buyer TEXT NOT NULL,
precedence TEXT NOT NULL,
class TEXT NOT NULL,
description TEXT NOT NULL,
expected_outcome TEXT NOT NULL,
standing TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
""")
conn.commit()
return conn
def seed_ticket(db: sqlite3.Connection, tenant: str, ticket: TicketDraft, standing: str = "open") -> int:
now = utc_now_iso()
cur = db.execute(
"""
INSERT INTO tickets
(tenant, title, buyer, precedence, class, description, expected_outcome, standing, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
tenant,
ticket.title,
ticket.buyer,
ticket.precedence,
ticket.class,
ticket.description,
ticket.expected_outcome,
standing,
now,
now,
),
)
db.commit()
return int(cur.lastrowid)We assemble the dependency layer and initialize a light-weight SQLite database for persistence. We mannequin real-world runtime dependencies, comparable to database connections and tenant insurance policies, and make them injectable into the agent. We additionally outline helper features that safely insert and handle ticket knowledge throughout execution.
def build_agent(model_name: str) -> Agent[SupportDeps, AgentDecision]:
agent = Agent(
f"openai:{model_name}",
output_type=AgentDecision,
output_retries=2,
directions=(
"You're a manufacturing help triage agent.n"
"Return an output that matches the AgentDecision schema.n"
"Use instruments once you want DB state.n"
"By no means invent ticket IDs.n"
"If the consumer intent is unclear, ask concise follow-up questions.n"
),
)
@agent.software
def create_ticket(ctx: RunContext[SupportDeps], ticket: TicketDraft) -> int:
deps = ctx.deps
if ticket.precedence in ("important", "excessive") and deps.coverage.get("require_security_phrase_for_critical", False):
if ticket.class == "safety" and "incident" not in ticket.description.decrease():
elevate ModelRetry("For safety excessive/important, embrace the phrase 'incident' in description and retry.")
return seed_ticket(deps.db, deps.tenant, ticket, standing="open")
@agent.software
def update_ticket_status(
ctx: RunContext[SupportDeps],
ticket_id: int,
standing: Literal["open", "in_progress", "resolved", "closed"],
) -> dict:
deps = ctx.deps
now = utc_now_iso()
cur = deps.db.execute("SELECT id FROM tickets WHERE tenant=? AND id=?", (deps.tenant, ticket_id))
if not cur.fetchone():
elevate ModelRetry(f"Ticket {ticket_id} not discovered for this tenant. Ask for the proper ticket_id.")
deps.db.execute(
"UPDATE tickets SET standing=?, updated_at=? WHERE tenant=? AND id=?",
(standing, now, deps.tenant, ticket_id),
)
deps.db.commit()
return {"ticket_id": ticket_id, "standing": standing, "updated_at": now}
@agent.software
def query_ticket(ctx: RunContext[SupportDeps], ticket_id: int) -> dict:
deps = ctx.deps
cur = deps.db.execute(
"""
SELECT id, title, buyer, precedence, class, standing, created_at, updated_at
FROM tickets WHERE tenant=? AND id=?
""",
(deps.tenant, ticket_id),
)
row = cur.fetchone()
if not row:
elevate ModelRetry(f"Ticket {ticket_id} not discovered. Ask the consumer for a sound ticket_id.")
keys = ["id", "title", "customer", "priority", "category", "status", "created_at", "updated_at"]
return dict(zip(keys, row))
@agent.software
def list_open_tickets(ctx: RunContext[SupportDeps], restrict: int = 5) -> listing:
deps = ctx.deps
restrict = max(1, min(int(restrict), 20))
cur = deps.db.execute(
"""
SELECT id, title, precedence, class, standing, updated_at
FROM tickets
WHERE tenant=? AND standing IN ('open','in_progress')
ORDER BY updated_at DESC
LIMIT ?
""",
(deps.tenant, restrict),
)
rows = cur.fetchall()
return [
{"id": r[0], "title": r[1], "precedence": r[2], "class": r[3], "standing": r[4], "updated_at": r[5]}
for r in rows
]
@agent.output_validator
def validate_decision(ctx: RunContext[SupportDeps], out: AgentDecision) -> AgentDecision:
deps = ctx.deps
if out.motion == "create_ticket" and out.ticket is None:
elevate ModelRetry("You selected create_ticket however didn't present ticket. Present ticket fields and retry.")
if out.motion in ("update_ticket", "query_ticket") and out.ticket_id is None:
elevate ModelRetry("You selected replace/question however didn't present ticket_id. Ask for ticket_id and retry.")
if out.ticket and out.ticket.precedence == "important" and never deps.coverage.get("allow_critical", True):
elevate ModelRetry("This tenant doesn't enable 'important'. Downgrade to 'excessive' and retry.")
return out
return agentIt accommodates the core agent logic for assembling a model-agnostic PydanticAI agent. We register typed instruments for creating, querying, updating, and itemizing tickets, permitting the agent to work together with exterior state in a managed means. We additionally implement output validation so the agent can self-correct each time its choices violate enterprise guidelines.
db = init_db()
deps = SupportDeps(
db=db,
tenant="acme_corp",
coverage={"allow_critical": True, "require_security_phrase_for_critical": True},
)
seed_ticket(
db,
deps.tenant,
TicketDraft(
title="Double-charged on bill 8831",
buyer="Riya",
precedence="excessive",
class="billing",
description="Buyer stories they have been billed twice for bill 8831 and needs a refund and affirmation e mail.",
expected_outcome="Difficulty a refund and make sure decision to buyer.",
),
)
seed_ticket(
db,
deps.tenant,
TicketDraft(
title="App crashes on login after replace",
buyer="Sam",
precedence="excessive",
class="bug",
description="After newest replace, the app crashes instantly on login. Reproducible on two units; wants investigation.",
expected_outcome="Present a repair or workaround and restore profitable logins.",
),
)
agent = build_agent("gpt-4o-mini")
async def run_case(immediate: str):
res = await agent.run(immediate, deps=deps)
out = res.output
print(json.dumps(out.model_dump(), indent=2))
return out
case_a = await run_case(
"We suspect account takeover: a number of password reset emails and unauthorized logins. "
"Buyer=Leila. Precedence=important. Open a safety ticket."
)
case_b = await run_case("Checklist our open tickets and summarize what to sort out first.")
case_c = await run_case("What's the standing of ticket 1? If it is open, transfer it to in_progress.")
agent_alt = build_agent("gpt-4o")
alt_res = await agent_alt.run(
"Create a characteristic request ticket: buyer=Noah needs 'export to CSV' in analytics dashboard; precedence=medium.",
deps=deps,
)
print(json.dumps(alt_res.output.model_dump(), indent=2))We wire every thing collectively by seeding preliminary knowledge and operating the agent asynchronously, in a notebook-safe method. We execute a number of real-world eventualities to point out how the agent causes, calls instruments, and returns schema-valid outputs. We additionally display how simply we will swap the underlying mannequin whereas maintaining the identical workflows and ensures intact.
In conclusion, we confirmed how a type-safe agent can cause, name instruments, validate its personal outputs, and get well from errors with out handbook intervention. We stored the logic model-agnostic, permitting us to swap underlying LLMs whereas preserving the identical schemas and instruments, which is important for long-term maintainability. Total, we demonstrated how combining strict schema enforcement, dependency injection, and async execution closes the reliability hole in agentic AI and supplies a stable basis for constructing reliable manufacturing methods.
Take a look at the Full Codes Here. Additionally, be happy to observe us on Twitter and don’t neglect to hitch our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
