On this tutorial, we construct a production-ready agentic workflow that prioritizes reliability over best-effort technology by imposing strict, typed outputs at each step. We use PydanticAI to outline clear response schemas, wire in instruments by way of dependency injection, and make sure the agent can safely work together with exterior programs, comparable to a database, with out breaking execution. By working the whole lot in a notebook-friendly, async-first setup, we show methods to transfer past fragile chatbot patterns towards sturdy agentic programs 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, Optionally available, 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 setting 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 might 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
purpose: str = Subject(..., min_length=20, max_length=400)
confidence: Confidence
ticket: Optionally available[TicketDraft] = None
ticket_id: Optionally available[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 have to 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 imposing 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 ("crucial", "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/crucial, 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 right 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 legitimate 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 == "crucial" and never deps.coverage.get("allow_critical", True):
elevate ModelRetry("This tenant doesn't enable 'crucial'. Downgrade to 'excessive' and retry.")
return out
return agentIt incorporates 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 every 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 experiences they had been billed twice for bill 8831 and desires a refund and affirmation e-mail.",
expected_outcome="Subject a refund and ensure 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=crucial. 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 desires 'export to CSV' in analytics dashboard; precedence=medium.",
deps=deps,
)
print(json.dumps(alt_res.output.model_dump(), indent=2))We wire the whole lot collectively by seeding preliminary knowledge and working the agent asynchronously, in a notebook-safe method. We execute a number of real-world eventualities to indicate how the agent causes, calls instruments, and returns schema-valid outputs. We additionally show 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 purpose, name instruments, validate its personal outputs, and recuperate from errors with out handbook intervention. We saved the logic model-agnostic, permitting us to swap underlying LLMs whereas preserving the identical schemas and instruments, which is crucial 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 strong basis for constructing reliable manufacturing programs.
Try the Full Codes Right here. Additionally, be happy to observe us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you’ll be able to be a part of us on telegram as properly.
