{"id":200,"date":"2025-12-29T14:04:26","date_gmt":"2025-12-29T06:04:26","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=200"},"modified":"2025-12-29T14:04:26","modified_gmt":"2025-12-29T06:04:26","slug":"how-to-build-contract-first-agentic-decision-systems-with-pydanticai-for-risk-aware-policy-compliant-enterprise-ai","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=200","title":{"rendered":"How to Build Contract-First Agentic Decision Systems with PydanticAI for Risk-Aware, Policy-Compliant Enterprise AI"},"content":{"rendered":"

In this tutorial, we demonstrate how to design a contract-first agentic decision system using PydanticAI<\/strong><\/a>, treating structured schemas as non-negotiable governance contracts rather than optional output formats. We show how we define a strict decision model that encodes policy compliance, risk assessment, confidence calibration, and actionable next steps directly into the agent\u2019s output schema. By combining Pydantic validators with PydanticAI\u2019s retry and self-correction mechanisms, we ensure that the agent cannot produce logically inconsistent or non-compliant decisions. Throughout the workflow, we focus on building an enterprise-grade decision agent that reasons under constraints, making it suitable for real-world risk, compliance, and governance scenarios rather than toy prompt-based demos. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip -q install -U pydantic-ai pydantic openai nest_asyncio\n\n\nimport os\nimport time\nimport asyncio\nimport getpass\nfrom dataclasses import dataclass\nfrom typing import List, Literal\n\n\nimport nest_asyncio\nnest_asyncio.apply()\n\n\nfrom pydantic import BaseModel, Field, field_validator\nfrom pydantic_ai import Agent\nfrom pydantic_ai.models.openai import OpenAIChatModel\nfrom pydantic_ai.providers.openai import OpenAIProvider\n\n\nOPENAI_API_KEY = os.getenv(\"OPENAI_API_KEY\")\nif not OPENAI_API_KEY:\n   try:\n       from google.colab import userdata\n       OPENAI_API_KEY = userdata.get(\"OPENAI_API_KEY\")\n   except Exception:\n       OPENAI_API_KEY = None\nif not OPENAI_API_KEY:\n   OPENAI_API_KEY = getpass.getpass(\"Enter OPENAI_API_KEY: \").strip()<\/code><\/pre>\n<\/div>\n<\/div>\n
We set up the execution environment by installing the required libraries and configuring asynchronous execution for Google Colab. We securely load the OpenAI API key and ensure the runtime is ready to handle async agent calls. This establishes a stable foundation for running the contract-first agent without environment-related issues. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class RiskItem(BaseModel):\n   risk: str = Field(..., min_length=8)\n   severity: Literal[\"low\", \"medium\", \"high\"]\n   mitigation: str = Field(..., min_length=12)\n\n\n\n\nclass DecisionOutput(BaseModel):\n   decision: Literal[\"approve\", \"approve_with_conditions\", \"reject\"]\n   confidence: float = Field(..., ge=0.0, le=1.0)\n   rationale: str = Field(..., min_length=80)\n   identified_risks: List[RiskItem] = Field(..., min_length=2)\n   compliance_passed: bool\n   conditions: List[str] = Field(default_factory=list)\n   next_steps: List[str] = Field(..., min_length=3)\n   timestamp_unix: int = Field(default_factory=lambda: int(time.time()))\n\n\n   @field_validator(\"confidence\")\n   @classmethod\n   def confidence_vs_risk(cls, v, info):\n       risks = info.data.get(\"identified_risks\") or []\n       if any(r.severity == \"high\" for r in risks) and v > 0.70:\n           raise ValueError(\"confidence too high given high-severity risks\")\n       return v\n\n\n   @field_validator(\"decision\")\n   @classmethod\n   def reject_if_non_compliant(cls, v, info):\n       if info.data.get(\"compliance_passed\") is False and v != \"reject\":\n           raise ValueError(\"non-compliant decisions must be reject\")\n       return v\n\n\n   @field_validator(\"conditions\")\n   @classmethod\n   def conditions_required_for_conditional_approval(cls, v, info):\n       d = info.data.get(\"decision\")\n       if d == \"approve_with_conditions\" and (not v or len(v) < 2):\n           raise ValueError(\"approve_with_conditions requires at least 2 conditions\")\n       if d == \"approve\" and v:\n           raise ValueError(\"approve must not include conditions\")\n       return v<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the core decision contract using strict Pydantic models that precisely describe a valid decision. We encode logical constraints such as confidence\u2013risk alignment, compliance-driven rejection, and conditional approvals directly into the schema. This ensures that any agent output must satisfy business logic, not just syntactic structure. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@dataclass\nclass DecisionContext:\n   company_policy: str\n   risk_threshold: float = 0.6\n\n\n\n\nmodel = OpenAIChatModel(\n   \"gpt-5\",\n   provider=OpenAIProvider(api_key=OPENAI_API_KEY),\n)\n\n\nagent = Agent(\n   model=model,\n   deps_type=DecisionContext,\n   output_type=DecisionOutput,\n   system_prompt=\"\"\"\nYou are a corporate decision analysis agent.\nYou must evaluate risk, compliance, and uncertainty.\nAll outputs must strictly satisfy the DecisionOutput schema.\n\"\"\"\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We inject enterprise context through a typed dependency object and initialize the OpenAI-backed PydanticAI agent. We configure the agent to produce only structured decision outputs that conform to the predefined contract. This step formalizes the separation between business context and model reasoning. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@agent.output_validator\ndef ensure_risk_quality(result: DecisionOutput) -> DecisionOutput:\n   if len(result.identified_risks) < 2:\n       raise ValueError(\"minimum two risks required\")\n   if not any(r.severity in (\"medium\", \"high\") for r in result.identified_risks):\n       raise ValueError(\"at least one medium or high risk required\")\n   return result\n\n\n\n\n@agent.output_validator\ndef enforce_policy_controls(result: DecisionOutput) -> DecisionOutput:\n   policy = CURRENT_DEPS.company_policy.lower()\n   text = (\n       result.rationale\n       + \" \".join(result.next_steps)\n       + \" \".join(result.conditions)\n   ).lower()\n   if result.compliance_passed:\n       if not any(k in text for k in [\"encryption\", \"audit\", \"logging\", \"access control\", \"key management\"]):\n           raise ValueError(\"missing concrete security controls\")\n   return result<\/code><\/pre>\n<\/div>\n<\/div>\n
We add output validators that act as governance checkpoints after the model generates a response. We force the agent to identify meaningful risks and to explicitly reference concrete security controls when claiming compliance. If these constraints are violated, we trigger automatic retries to enforce self-correction. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
async def run_decision():\n   global CURRENT_DEPS\n   CURRENT_DEPS = DecisionContext(\n       company_policy=(\n           \"No deployment of systems handling personal data or transaction metadata \"\n           \"without encryption, audit logging, and least-privilege access control.\"\n       )\n   )\n\n\n   prompt = \"\"\"\nDecision request:\nDeploy an AI-powered customer analytics dashboard using a third-party cloud vendor.\nThe system processes user behavior and transaction metadata.\nAudit logging is not implemented and customer-managed keys are uncertain.\n\"\"\"\n\n\n   result = await agent.run(prompt, deps=CURRENT_DEPS)\n   return result.output\n\n\n\n\ndecision = asyncio.run(run_decision())\n\n\nfrom pprint import pprint\npprint(decision.model_dump())<\/code><\/pre>\n<\/div>\n<\/div>\n
We run the agent on a realistic decision request and capture the validated structured output. We demonstrate how the agent evaluates risk, policy compliance, and confidence before producing a final decision. This completes the end-to-end contract-first decision workflow in a production-style setup.<\/p>\n
In conclusion, we demonstrate how to move from free-form LLM outputs to governed, reliable decision systems using PydanticAI. We show that by enforcing hard contracts at the schema level, we can automatically align decisions with policy requirements, risk severity, and confidence realism without manual prompt tuning. This approach allows us to build agents that fail safely, self-correct when constraints are violated, and produce auditable, structured outputs that downstream systems can trust. Ultimately, we demonstrate that contract-first agent design enables us to deploy agentic AI as a dependable decision layer within production and enterprise environments.<\/p>\n
\n
Check out the\u00a0FULL CODES here<\/a><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post How to Build Contract-First Agentic Decision Systems with PydanticAI for Risk-Aware, Policy-Compliant Enterprise AI<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we demonstra…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-200","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/200","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=200"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/200\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=200"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=200"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=200"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}