Validated JSON Agent 🤖

A robust, production-ready pattern for ensuring Large Language Models (LLMs) return strictly valid, schema-compliant JSON data. This project demonstrates how to bridge the gap between non-deterministic LLM outputs and deterministic software systems.

---

🚀 The Problem

LLMs often "hallucinate" JSON structures, return malformed strings, or miss required fields. In production, this leads to application crashes and data corruption.

---

🛠 The Solution

This agent implements a Self-Correction Loop:

1. Schema Injection: Automatically generates a JSON schema from Pydantic models.
2. Strict Validation: Uses Pydantic v2 to validate the LLM response.
3. Feedback Loop: If validation fails, the specific error (e.g., "missing field 'age'") is fed back to the LLM for an immediate fix.
4. Async Execution: Built with AsyncOpenAI for high-throughput performance.

---

🏗 Architecture

• Data Layer: Pydantic v2 for type safety and constraints.
• Agent Layer: Asynchronous retry logic with stateful conversation history.
• Ops Layer: GitHub Actions (CI/CD), Docker (Slim/Non-root), and Ruff/Mypy for code quality.

---

📦 Installation

1. Clone & Install:

   git clone [https://github.com/yourusername/validated-json-agent.git](https://github.com/yourusername/validated-json-agent.git)
   cd validated-json-agent
   pip install -r requirements.txt

2. Environment Setup:
   Create a .env file:

   OPENAI_API_KEY=sk-xxxx...MODEL_NAME=gpt-4o-miniMAX_RETRIES=3

---

💻 Usage

Run the asynchronous demonstration:

python main.py

Example Schema Enforcement:

# The agent ensures the LLM output matches this exactly:classUserProfile(BaseModel):
name: strage: int=Field(..., ge=0, le=120)
is_student: boolhobbies: List[str]

---

🛡 Production Features

• Security: Dockerfile runs as a non-privileged appuser.
• Observability: Structured logging tracks token usage and validation failure rates.
• Reliability: CI pipeline enforces 100% pass rate on linting (Ruff), type checking (Mypy), and unit tests (Pytest).
• Scalability: Fully asynchronous I/O for integration into FastAPI/distributed systems.

---

🧪 Testing

Run the suite to verify the retry logic and mocking:

pytest tests/

---

📂 Repository Structure

.
├── .github/
│ └── workflows/
│ ├── ci.yml # Linting, Typing, Tests
│ └── cd.yml # Docker Build & Push
├── src/
│ ├── __init__.py
│ ├── agent_async.py # Core Logic (Async)
│ ├── config.py # Env Management
│ ├── logger.py # Observability
│ └── schema.py # Data Contract
├── tests/
│ └── test_agent.py # Unit Tests (Mocked)
├── .env.example
├── .gitignore
├── Dockerfile # Production Container
├── main.py # Entry Point
├── README.md # Documentation
└── requirements.txt # Dependencies

---

📜 License

MIT