Beyond Hallucinations: Building Reliable Multi-Agent Systems with Robust Validation

Introduction & The Problem

The Solution Concept & Architecture

Step-by-Step Implementation

pip install openai pydantic requests

import os
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
from openai import OpenAI

# --- 1. Configuration and LLM Client Initialization ---
# For demonstration, use a placeholder or set your OpenAI API key
# client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# Or for a local Ollama instance (e.g., mistral)
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama") # Replace with your Ollama setup
LLM_MODEL = "mistral"

# --- 2. Define Output Schemas for Validation ---
class FinancialReportEntry(BaseModel):
    company_name: str = Field(description="Name of the company.")
    revenue_usd: float = Field(description="Reported annual revenue in USD.")
    profit_margin_percent: float = Field(description="Profit margin as a percentage.")
    year: int = Field(description="Fiscal year of the report.")

class DataRetrievalOutput(BaseModel):
    status: str = Field(description="Status of the data retrieval: 'SUCCESS' or 'FAILURE'.")
    data: Optional[List[FinancialReportEntry]] = Field(default=None, description="List of financial report entries if successful.")
    error_message: Optional[str] = Field(default=None, description="Error message if retrieval failed.")

# --- 3. Agent Implementations ---
class DataRetrievalAgent:
    def __init__(self, llm_client: OpenAI, model: str):
        self.llm_client = llm_client
        self.model = model

    def retrieve_financial_data(self, query: str) -> str:
        prompt = (
            f"You are a highly accurate financial data retrieval agent. Your task is to extract annual financial data "
            f"(company name, revenue in USD, profit margin percentage, and fiscal year) for the companies mentioned in the query. "
            f"If data is found, format it strictly as a JSON array of objects, each matching the FinancialReportEntry schema. "
            f"If no data is found for a company, or if you encounter any issues, provide a 'FAILURE' status with an error message."
            f"Query: {query}"
            f"Example successful output: "
            f"{{\"status\": \"SUCCESS\", \"data\": [{{\"company_name\": \"Example Inc.\", \"revenue_usd\": 1000000.0, \"profit_margin_percent\": 15.5, \"year\": 2023}}]}}"
            f"Example failure output: "
            f"{{\"status\": \"FAILURE\", \"error_message\": \"Could not find data for Example Inc.\"}}"
        )
        try:
            response = self.llm_client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                response_format={"type": "json_object"}
            )
            return response.choices[0].message.content
        except Exception as e:
            return f'{{"status": "FAILURE", "error_message": "LLM call failed: {str(e)}"}}'

class AnalysisAgent:
    def __init__(self, llm_client: OpenAI, model: str):
        self.llm_client = llm_client
        self.model = model

    def analyze_data(self, financial_data: List[FinancialReportEntry]) -> str:
        data_str = ", ".join([entry.json() for entry in financial_data])
        prompt = (
            f"You are a financial analysis agent. Analyze the following financial data and provide key insights, trends, "
            f"and potential risks. Focus on growth, profitability, and any notable changes over the years. "
            f"Data: [{data_str}]"
        )
        response = self.llm_client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

# --- 4. Validation and Self-Correction Orchestrator ---
class AgentOrchestrator:
    def __init__(self, data_agent: DataRetrievalAgent, analysis_agent: AnalysisAgent):
        self.data_agent = data_agent
        self.analysis_agent = analysis_agent

    def orchestrate_analysis(self, query: str, max_retries: int = 2) -> Dict[str, Any]:
        # Step 1: Data Retrieval
        print("\n[Orchestrator] Initiating data retrieval...")
        raw_data_output = self.data_agent.retrieve_financial_data(query)

        # Step 2: Validate Data Retrieval Output with Self-Correction
        retries = 0
        validated_data_output: Optional[DataRetrievalOutput] = None
        while retries <= max_retries:
            try:
                print(f"[Orchestrator] Attempt {retries+1} - Validating data output: {raw_data_output[:100]}...")
                validated_data_output = DataRetrievalOutput.parse_raw(raw_data_output)

                if validated_data_output.status == "FAILURE":
                    print(f"[Orchestrator] Data retrieval failed: {validated_data_output.error_message}")
                    return {"status": "ERROR", "message": validated_data_output.error_message}

                print("[Orchestrator] Data output successfully validated.")
                break # Exit loop if validation passes

            except ValidationError as e:
                print(f"[Orchestrator] Validation failed (Attempt {retries+1}): {e}")
                retries += 1
                if retries <= max_retries:
                    print("[Orchestrator] Attempting self-correction for Data Retrieval Agent...")
                    # Self-correction prompt for the Data Retrieval Agent
                    correction_prompt = (
                        f"The previous output failed validation. The error was: {e}. "
                        f"Please re-evaluate the query and provide a valid JSON output strictly conforming to the specified schema.\nQuery: {query}"
                        f"Previous invalid output: {raw_data_output}"
                    )
                    raw_data_output = self.data_agent.retrieve_financial_data(correction_prompt) # Re-run with correction prompt
                else:
                    return {"status": "ERROR", "message": f"Failed to retrieve and validate data after {max_retries+1} attempts."}
            except Exception as e:
                print(f"[Orchestrator] Unexpected error during validation: {e}")
                return {"status": "ERROR", "message": f"Unexpected error: {e}"}

        if not validated_data_output or validated_data_output.status == "FAILURE":
            return {"status": "ERROR", "message": "Failed to retrieve and validate data."}

        if not validated_data_output.data:
            print("[Orchestrator] No financial data was retrieved.")
            return {"status": "SUCCESS", "message": "No financial data found for analysis."}

        # Step 3: Analysis
        print("\n[Orchestrator] Initiating data analysis...")
        analysis_result = self.analysis_agent.analyze_data(validated_data_output.data)
        print("[Orchestrator] Analysis complete.")

        return {"status": "SUCCESS", "data": validated_data_output.data, "analysis": analysis_result}

# --- 5. Main Execution ---
if __name__ == "__main__":
    data_retrieval_agent = DataRetrievalAgent(client, LLM_MODEL)
    analysis_agent = AnalysisAgent(client, LLM_MODEL)
    orchestrator = AgentOrchestrator(data_retrieval_agent, analysis_agent)

    # Test Case 1: Successful Retrieval and Analysis
    print("\n--- Running Test Case 1: Successful Scenario ---")
    result1 = orchestrator.orchestrate_analysis("annual financials for Apple Inc. and Microsoft Corp. for 2023")
    print("\nFinal Result (Test Case 1):")
    print(result1)

    # Test Case 2: Intentional Failure (Simulated - actual LLM might still get it right)
    # To truly simulate a validation failure, you might need a custom mock LLM response
    # For this example, let's assume the LLM might struggle with a complex or ambiguous query.
    print("\n--- Running Test Case 2: Potential Self-Correction Scenario ---")
    # A query designed to be slightly ambiguous or complex for the LLM initially
    result2 = orchestrator.orchestrate_analysis("latest financials for a big tech company named 'Acme Innovations' and 'Global Dynamics'")
    print("\nFinal Result (Test Case 2):")
    print(result2)

• Pydantic for Schema Validation: Enforces strict output formats, crucial for inter-agent communication and data integrity.
• Orchestrator Logic: Manages the flow, calls agents, and performs validation.
• Self-Correction Loop: If validation fails, the orchestrator provides feedback to the originating agent and re-prompts it, allowing for iterative refinement.

Optimization & Best Practices

• Semantic Validation: Beyond schema, implement checks for logical consistency and factual accuracy. Integrate with knowledge graphs or trusted external APIs to verify information.
• Multi-Agent Consensus: For critical decisions, have multiple agents independently perform the same task and compare their outputs. If discrepancies arise, trigger a reconciliation agent.
• Dynamic Retries & Backoff: Implement exponential backoff for retries to prevent overwhelming upstream services or LLM APIs.
• Human-in-the-Loop Integration: For high-stakes decisions or repeated validation failures, escalate to a human reviewer. Tools like LangChain's Human-in-the-Loop components or custom UIs can facilitate this.
• Observability & Monitoring: Log all agent interactions, validation results, and self-correction attempts. Use dashboards to visualize agent performance, error rates, and identify patterns that can inform prompt engineering or agent architecture improvements.
• Cost Optimization: Be mindful of LLM API costs during self-correction. Design prompts to be precise to minimize token usage during retries. Consider using smaller, fine-tuned models for specific validation or self-correction sub-tasks.
• Version Control for Prompts & Agent Definitions: Treat your prompts and agent configurations as code, managing them in version control systems to track changes and ensure reproducibility.

Business Impact & ROI

• Reduced Operational Costs: Fewer errors mean less manual intervention, freeing up human resources from repetitive oversight tasks. Automated self-correction reduces the need for expensive post-mortem analysis and debugging.
• Increased Trust & Adoption: Predictable and reliable AI outputs build confidence among users, stakeholders, and customers, accelerating the adoption of AI-driven solutions across the organization. This is crucial for CEOs seeking high ROI from AI investments.
• Enhanced Data Quality & Compliance: Strict output validation ensures that data processed and generated by AI agents meets quality standards and, where applicable, regulatory compliance requirements. This mitigates risks associated with incorrect information.
• Faster Time-to-Market for AI Products: With built-in reliability, development teams can deploy AI agents faster, knowing they have safeguards in place, reducing the overall development lifecycle and increasing agility for agencies and solopreneurs.
• Competitive Advantage: Businesses that can consistently deliver high-quality, reliable AI services will differentiate themselves in the market, attracting and retaining customers who value trust and performance.
• Scalability: Reliable agents can operate autonomously at scale, processing vast amounts of data and handling complex workflows without human bottlenecks, enabling true SaaS scalability.

Conclusion