On this tutorial, we construct a production-ready agentic workflow that prioritizes reliability over best-effort era by imposing 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 programs, akin to a database, with out breaking execution. By operating all the things in a notebook-friendly, async-first setup, we display the way to transfer past fragile chatbot patterns towards strong 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, Record
from pydantic import BaseModel, Subject, field_validator
from pydantic_ai import Agent, RunContext, ModelRetry
if not os.environ.get("OPENAI_API_KEY"):
strive:
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = (userdata.get("OPENAI_API_KEY") or "").strip()
besides Exception:
cross
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 appropriately. We securely load the OpenAI API key in a Colab-friendly approach so the tutorial works with out guide 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: Record[str] = Subject(default_factory=checklist, max_length=5)
@field_validator("follow_up_questions")
@classmethod
def short_questions(cls, v):
for q in v:
if len(q) > 140:
increase ValueError("Every follow-up query should be <= 140 characters.")
return vWe outline the strict information 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, akin to database connections and tenant insurance policies, and make them injectable into the agent. We additionally outline helper capabilities that safely insert and handle ticket information 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 if you want DB state.n"
"By no means invent ticket IDs.n"
"If the person 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():
increase 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():
increase 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:
increase ModelRetry(f"Ticket {ticket_id} not discovered. Ask the person 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) -> checklist:
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:
increase 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:
increase 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):
increase 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 approach. We additionally implement output validation so the agent can self-correct at any time when 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 have been billed twice for bill 8831 and needs a refund and affirmation e mail.",
expected_outcome="Challenge 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 gadgets; 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("Record 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 all the things collectively by seeding preliminary information and operating 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 display how simply we are able to 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 get well from errors with out guide 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. General, we demonstrated how combining strict schema enforcement, dependency injection, and async execution closes the reliability hole in agentic AI and gives a strong basis for constructing reliable manufacturing programs.
Take a look at the Full Codes Right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you’ll be able to be part of us on telegram as nicely.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the newest breakthroughs, get unique updates, and join with a worldwide community of future-focused thinkers.
Unlock tomorrow’s tendencies at the moment: learn extra, subscribe to our e-newsletter, and turn into a part of the NextTech neighborhood at NextTech-news.com
