working code

Pulkit Verma — Mon, 21 Sep 2026 14:43:54 +0000

use this code in cmd
pip install httpx pydantic openai

this is python code
import asyncio
import json
from typing import Any, Dict, List, Optional
import httpx
from openai import AsyncOpenAI
from pydantic import BaseModel, Field, ValidationError

=====================================================================

1. Typed Schemas (Input & Output Boundary Contracts)

=====================================================================

class SearchParams(BaseModel):
"""Defensive schema defining what parameters the LLM is permitted to send."""
query: str = Field(..., description="Target search query optimized for document lookup")
top_k: int = Field(default=3, ge=1, le=5, description="Number of results (clamped between 1 and 5)")
category: Optional[str] = Field(default=None, description="Optional category filter")

class ContentChunk(BaseModel):
"""Standardized ingested payload structure."""
id: str
source_url: str
text: str
score: float

class GroundedAgentResponse(BaseModel):
"""Final answer format forcing the LLM to provide verifiable citations."""
answer: str = Field(..., description="The direct answer synthesized from context")
cited_chunk_ids: List[str] = Field(
default_factory=list,
description="The exact chunk IDs used as evidence for this answer"
)

=====================================================================

2. Defensive Network Tool

=====================================================================

async def fetch_live_content(params: SearchParams) -> List[ContentChunk]:
"""
Executes a network fetch guarded by strict timeouts, exception handling,
and structured schema validation.
"""
timeout = httpx.Timeout(4.0, connect=2.0)

# In production, point this to your actual internal API or hybrid search endpoint
url = "https://jsonplaceholder.typicode.com/posts"

async with httpx.AsyncClient(timeout=timeout) as client:
    try:
        response = await client.get(url, params={"_limit": params.top_k})
        response.raise_for_status()
        raw_data = response.json()

        # Transform and validate external payload into typed ContentChunks
        chunks: List[ContentChunk] = []
        for item in raw_data:
            chunk = ContentChunk(
                id=f"doc-{item.get('id')}",
                source_url=f"https://example.com/posts/{item.get('id')}",
                text=f"{item.get('title')}: {item.get('body')}".replace("\n", " "),
                score=1.0,
            )
            chunks.append(chunk)
        return chunks

    except (httpx.TimeoutException, httpx.HTTPStatusError, httpx.RequestError) as err:
        # Defensive logging; fall back gracefully so the agent doesn't crash
        print(f"[Tool Warning] External fetch failed: {err}. Falling back to empty set.")
        return []

=====================================================================

3. Agent Execution Engine

=====================================================================

class ProductionAgent:
def init(self, model_name: str = "gpt-4o-mini"):
self.client = AsyncOpenAI()
self.model_name = model_name

    # Tool mapping for deterministic runtime execution
    self.tools_map = {
        "fetch_live_content": fetch_live_content
    }

    # OpenAI Tool Schema declaration derived directly from Pydantic
    self.tool_definitions = [
        {
            "type": "function",
            "function": {
                "name": "fetch_live_content",
                "description": "Fetch live verified articles and real-time knowledge.",
                "parameters": SearchParams.model_json_schema()
            }
        }
    ]

async def run(self, user_query: str) -> GroundedAgentResponse:
    messages = [
        {
            "role": "system",
            "content": (
                "You are a production assistant answering queries using live content. "
                "Always call `fetch_live_content` to retrieve information before answering. "
                "Ground your answers strictly on the retrieved chunks and cite their IDs."
            )
        },
        {"role": "user", "content": user_query}
    ]

    # Step 1: Initial invocation (Model decides whether to call a tool)
    first_response = await self.client.chat.completions.create(
        model=self.model_name,
        messages=messages,
        tools=self.tool_definitions,
        tool_choice="auto"
    )

    response_message = first_response.choices[0].message
    messages.append(response_message)

    retrieved_chunk_ids = set()

    # Step 2: Handle Tool Calls defensively
    if response_message.tool_calls:
        for tool_call in response_message.tool_calls:
            func_name = tool_call.function.name
            func_args_str = tool_call.function.arguments

            if func_name == "fetch_live_content":
                try:
                    # Defensive validation: parse arguments through Pydantic
                    raw_args = json.loads(func_args_str)
                    validated_params = SearchParams(**raw_args)

                    # Execute asynchronous fetch
                    chunks = await self.tools_map<a href="validated_params">func_name</a>
                    retrieved_chunk_ids.update(c.id for c in chunks)

                    # Serialize chunks for the LLM
                    tool_result_content = json.dumps([c.model_dump() for c in chunks])
                except (ValidationError, json.JSONDecodeError) as e:
                    tool_result_content = json.dumps({"error": f"Invalid tool arguments: {str(e)}"})

                # Append execution output back into context
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": func_name,
                    "content": tool_result_content
                })

    # Step 3: Final Synthesis with Structured Output
    final_completion = await self.client.beta.chat.completions.parse(
        model=self.model_name,
        messages=messages,
        response_format=GroundedAgentResponse
    )

    final_answer: GroundedAgentResponse = final_completion.choices[0].message.parsed

    # Step 4: Deterministic Grounding Verification
    self._verify_citations(final_answer, retrieved_chunk_ids)

    return final_answer

@staticmethod
def _verify_citations(response: GroundedAgentResponse, valid_ids: set) -> None:
    """
    Hardening step: ensures the agent did not invent non-existent citation IDs.
    """
    hallucinated_ids = [cid for cid in response.cited_chunk_ids if cid not in valid_ids]
    if hallucinated_ids:
        print(f"[Verification Alert] Hallucinated citation IDs detected: {hallucinated_ids}")
        # Prune out hallucinated IDs to ensure client-side trust
        response.cited_chunk_ids = [cid for cid in response.cited_chunk_ids if cid in valid_ids]

=====================================================================

4. Entrypoint

=====================================================================

async def main():
agent = ProductionAgent()
user_prompt = "What are the latest updates about project announcement posts?"

print(f"User: {user_prompt}\nRunning agent...")
result = await agent.run(user_prompt)

print("\n--- Final Grounded Output ---")
print(f"Answer:\n{result.answer}\n")
print(f"Verified Source IDs: {result.cited_chunk_ids}")

if name == "main":
asyncio.run(main())

Defensive Agent Architecture

Pulkit Verma — Mon, 21 Sep 2026 14:36:41 +0000

Schema Injection: SearchParams.model_json_schema() is passed into the LLM's tool/function calling definition.

Deterministic Invocation: The LLM outputs a JSON payload matching SearchParams.

Execution & Validation: fetch_live_content validates the arguments, dispatches the HTTP call within a strict 4-second window, and deserializes the results into ContentChunk objects.

Context Injection: The serialized chunks and their respective ids are injected back into the LLM context for synthesis and citation grounding.

This pattern addresses a critical failure mode in AI engineering: the fragility of connecting probabilistic language models to deterministic production systems.

The Problems It Solves
Hallucinated Tool Inputs: LLMs frequently drift or generate malformed parameters (e.g., requesting -10 or 1,000,000 items, or passing unparseable types). Defining the input via SearchParams forces the model to conform to strict boundaries (ge=1, le=10), catching invalid tool requests before they hit your infrastructure.

Agent Hanging and Runaway Costs: Without rigid timeouts, a lagging external API or vector index will cause the agent to hang indefinitely. This blocks server threads, degrades user experience, and inflates cloud costs. The explicit 4-second timeout enforces a deterministic upper bound on wait times.

Cascading Application Crashes: Unhandled HTTP errors (429 rate limits, 500 server crashes) crash naive agent runtimes. Swallowing these failures into a structured fallback (return []) allows the agent to handle missing data gracefully rather than throwing an uncaught exception.

DEV Community: Pulkit Verma

working code

=====================================================================

1. Typed Schemas (Input & Output Boundary Contracts)

=====================================================================

=====================================================================

2. Defensive Network Tool

=====================================================================

=====================================================================

3. Agent Execution Engine

=====================================================================

=====================================================================

4. Entrypoint

=====================================================================

Defensive Agent Architecture