working code

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())