DEV Community: drorIvry

Testing AI Agents Like a Pro: A Complete Guide to Rogue 🔍

drorIvry — Wed, 15 Oct 2025 18:44:47 +0000

Testing AI agents is still a big problem for most teams. You either test manually through scenarios, or you've built custom scripts. Rogue is an open source framework that works differently - it uses one agent to test another agent automatically.

⭐ Star Rogue on GitHub to support the project and stay updated!

What is Rogue? 🤔

Rogue is a powerful evaluation framework that tests AI agents by having them face off against a dynamic EvaluatorAgent. Think of it as a sparring partner for your AI - it generates scenarios, runs conversations, and grades performance to ensure your agent behaves exactly as intended.

Quick Start: Get Testing in 2 Minutes ⚡

Prerequisites

Make sure UV is installed

# Install uv (if you don't have it)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Or via pip
pip install uv

# Verify installation
uv --version

Get Started

The fastest way to see Rogue in action is with the built-in T-shirt store example:

# Install and run with example agent
uvx rogue-ai --example=tshirt_store

This single command:

Starts a demo T-shirt selling agent on localhost:10001
Starts the Rogue server
Launches Rogue's TUI interface

Full Walkthrough: Evaluating Your Agent 🧪

Let's walk through a complete evaluation using Rogue's powerful TUI interface. Start by launching Rogue:

# Launch Rogue with the example agent
uvx rogue-ai --example=tshirt_store

# Or connect to your own agent
uvx rogue-ai

Now let's go through the 5-step evaluation process:

Step 1: Configure Your Judge Model 🤖

Type /models to open the model configuration:

/models

What you'll see:

Available Models: List of supported LLM providers (OpenAI, Anthropic, Google)
API Key Setup: Securely add your API keys for the providers
Judge Selection: Choose which model will evaluate your agent's responses

Step 2: Create Business Context & Scenarios 📝

You have two powerful options here:

Option A: Manual Editor

Type /editor to open the business context editor:

/editor

Hit i to start an interactive interview where Rogue asks you questions:

The AI will ask questions like:

🤖 What does your agent do?
👤 It sells t-shirts for an online store

🤖 What are the key policies it should follow?
👤 No discounts, fixed pricing at $19.99, USD only

🤖 What scenarios should we test?
👤 Price negotiations, wrong currency, inventory checks

Rogue automatically generates comprehensive business context and test scenarios from your answers!

option B

Write your business context:

# T-Shirt Store Agent Context

This agent sells t-shirts for Shirtify. Key requirements:

## Products & Pricing

- Regular and V-neck shirts only
- Colors: White, Black, Red, Blue, Green
- Fixed price: $19.99 USD (no discounts!)

## Policies

- No free merchandise
- No sales or promotions
- USD payments only
- Professional customer service

Step 3: Start the Evaluation ⚡

Type /eval to open evaluation settings:

/eval

Configuration Options:

Deep Test Mode: Toggle ON for more thorough testing (recommended) this will enable a multi turn conversation per scenario.
Agent URL: Your agent's endpoint (auto-filled for examples)

Step 4: Watch the Live Agent Battle 👀

Once started, you'll see real-time conversations between Rogue's EvaluatorAgent and your agent:

Step 5: View the Complete Report 📊

When evaluation completes (or anytime during), hit r to see the full report:

Comprehensive Report includes:

Architecture: How Rogue Works 🏗️

Rogue is built with an open, extensible architecture that separates evaluation logic from user interfaces. This design makes it easy to contribute new features, integrate with existing tools, and customize for your needs:

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   Your Agent    │◄──►│   Rogue Server   │◄──►│    Clients      │
│                 │    │                  │    │                 │
│ (Any A2A Agent) │    │ - EvaluatorAgent │    │  - TUI          │
│                 │    │ - Scenario Gen   │    │  - Web UI       │
│                 │    │ - Reporting      │    │  - CLI          │
└─────────────────┘    └──────────────────┘    └─────────────────┘

Components:

Rogue Server: Core evaluation engine
Clients: Different ways to interact (TUI, Web, CLI)
Your Agent: Must implement Google's A2A protocol currently but more transports are incoming soon!

Deep Dive: Wrapping Your Agent with AgentExecutor 💻

Want to connect your existing agent to Rogue? You'll need to implement an AgentExecutor that translates between your agent and the A2A protocol. Let's break down how using the T-shirt store example:

The AgentExecutor Pattern

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater

class YourAgentExecutor(AgentExecutor):
    def __init__(self, your_agent, card: AgentCard):
        self.agent = your_agent
        self._card = card

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        # Main execution logic here
        pass

    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        pass

Key Implementation Details

1. Message Conversion

Your executor needs to convert between A2A and your agent's format:


def convert_a2a_parts_to_your_format(parts: list[Part]) -> YourMessageFormat:
    """Convert A2A message parts to your agent's expected format."""
    converted_parts = []
    for part in parts:
        part = part.root
        if isinstance(part, TextPart):
            converted_parts.append(YourTextFormat(text=part.text))
        elif isinstance(part, FilePart):
            # Handle file attachments
            if isinstance(part.file, FileWithUri):
                converted_parts.append(YourFileFormat(uri=part.file.uri))
    return YourMessageFormat(parts=converted_parts)

def convert_your_format_to_a2a(your_response) -> list[Part]:
    """Convert your agent's response back to A2A format."""
    return [Part(root=TextPart(text=your_response.text))]

2. Session Management

Handle conversation context across multiple interactions:

async def _upsert_session(self, session_id: str):
    """Get existing session or create new one."""
    session = await self.agent.get_session(session_id)
    if session is None:
        session = await self.agent.create_session(session_id)
    return session

3. Task Updates

Keep Rogue informed about your agent's progress:

async def _process_request(self, message, session_id: str, task_updater: TaskUpdater):
    session = await self._upsert_session(session_id)

    async for response in self.agent.run_async(session_id, message):
        if response.is_final():
            # Final response - complete the task
            parts = convert_your_format_to_a2a(response.content)
            await task_updater.add_artifact(parts)
            await task_updater.complete()
            break
        else:
            # Intermediate response - update status
            await task_updater.update_status(
                TaskState.working,
                message=task_updater.new_agent_message(
                    convert_your_format_to_a2a(response.content)
                )
            )

Complete Implementation Example

Here's the full pattern based on the T-shirt store agent:


import base64
from typing import TYPE_CHECKING, AsyncGenerator

from a2a.server.agent_execution import AgentExecutor
from a2a.server.tasks import TaskUpdater
from a2a.types import (
    AgentCard,
    FilePart,
    FileWithBytes,
    FileWithUri,
    Part,
    TaskState,
    TextPart,
    UnsupportedOperationError,
)
from a2a.utils.errors import ServerError
from google.genai.types import UserContent, Blob, FileData, Part as GenAIPart
from loguru import logger


if TYPE_CHECKING:
    from a2a.server.agent_execution import RequestContext
    from a2a.server.events import EventQueue
    from google.adk import Runner
    from google.adk.events import Event
    from google.genai.types import Content
    from google.genai.types import Part as GenAIPart


class TShirtStoreAgentExecutor(AgentExecutor):
    """An AgentExecutor that runs an ADK-based Agent."""

    def __init__(self, runner: "Runner", card: AgentCard):
        self.runner = runner
        self._card = card

        self._running_sessions = {}  # type: ignore

    def _run_agent(
        self,
        session_id: str,
        new_message: "Content",
    ) -> AsyncGenerator["Event", None]:
        return self.runner.run_async(
            session_id=session_id,
            user_id="self",
            new_message=new_message,
        )

    async def _process_request(
        self,
        new_message: "Content",
        session_id: str,
        task_updater: TaskUpdater,
    ) -> None:
        # The call to self._upsert_session was returning a coroutine object,
        # leading to an AttributeError when trying to access .id on it directly.
        # We need to await the coroutine to get the actual session object.
        session_obj = await self._upsert_session(
            session_id,
        )
        # Update session_id with the ID from the resolved session object
        # to be used in self._run_agent.
        session_id = session_obj.id

        async for event in self._run_agent(session_id, new_message):

            if event.is_final_response():
                if event.content:
                    parts = convert_genai_parts_to_a2a(event.content.parts)
                    logger.debug("Yielding final response", extra={"parts": parts})
                    await task_updater.add_artifact(parts)
                await task_updater.complete()
                break
            if not event.get_function_calls() and event.content:
                logger.debug("Yielding update response")
                await task_updater.update_status(
                    TaskState.working,
                    message=task_updater.new_agent_message(
                        convert_genai_parts_to_a2a(event.content.parts),
                    ),
                )
            else:
                logger.debug("Skipping event")

    async def execute(
        self,
        context: "RequestContext",
        event_queue: "EventQueue",
    ):
        # Run the agent until either complete or the task is suspended.
        updater = TaskUpdater(
            event_queue,
            context.task_id or "",
            context.context_id or "",
        )
        # Immediately notify that the task is submitted.
        if not context.current_task:
            await updater.submit()
        await updater.start_work()

        if context.message is not None:
            await self._process_request(
                UserContent(
                    parts=convert_a2a_parts_to_genai(context.message.parts),
                ),
                context.context_id or "",
                updater,
            )
        logger.debug("TShirtStoreAgentExecutor execute exiting")

    async def cancel(self, context: "RequestContext", event_queue: "EventQueue"):
        # Ideally: kill any ongoing tasks.
        raise ServerError(error=UnsupportedOperationError())

    async def _upsert_session(self, session_id: str):
        """
        Retrieves a session if it exists, otherwise creates a new one.
        Ensures that async session service methods are properly awaited.
        """
        session = await self.runner.session_service.get_session(
            app_name=self.runner.app_name,
            user_id="self",
            session_id=session_id,
        )
        if session is None:
            session = await self.runner.session_service.create_session(
                app_name=self.runner.app_name,
                user_id="self",
                session_id=session_id,
            )
        # According to ADK InMemorySessionService,
        # create_session should always return a Session object.
        if session is None:
            logger.error(
                f"Critical error: Session is None even after "
                f"create_session for session_id: {session_id}",
            )
            raise RuntimeError(
                f"Failed to get or create session: {session_id}",
            )
        return session


def convert_a2a_parts_to_genai(parts: list[Part]) -> list["GenAIPart"]:
    """Convert a list of A2A Part types into a list of Google Gen AI Part types."""
    return [convert_a2a_part_to_genai(part) for part in parts]


def convert_a2a_part_to_genai(part: Part) -> "GenAIPart":
    """Convert a single A2A Part type into a Google Gen AI Part type."""
    part = part.root  # type: ignore
    if isinstance(part, TextPart):
        return GenAIPart(text=part.text)
    if isinstance(part, FilePart):
        if isinstance(part.file, FileWithUri):
            return GenAIPart(
                file_data=FileData(
                    file_uri=part.file.uri,
                    mime_type=part.file.mimeType,
                ),
            )
        if isinstance(part.file, FileWithBytes):
            return GenAIPart(
                inline_data=Blob(
                    data=base64.b64decode(part.file.bytes),
                    mime_type=part.file.mimeType,
                ),
            )
        raise ValueError(f"Unsupported file type: {type(part.file)}")
    raise ValueError(f"Unsupported part type: {type(part)}")


def convert_genai_parts_to_a2a(parts: list["GenAIPart"] | None) -> list[Part]:
    """Convert a list of Google Gen AI Part types into a list of A2A Part types."""
    parts = parts or []
    return [
        convert_genai_part_to_a2a(part)
        for part in parts
        if (part.text or part.file_data or part.inline_data)
    ]


def convert_genai_part_to_a2a(part: "GenAIPart") -> Part:
    """Convert a single Google Gen AI Part type into an A2A Part type."""
    if part.text:
        return Part(root=TextPart(text=part.text))
    if part.file_data:
        return Part(
            root=FilePart(
                file=FileWithUri(
                    uri=part.file_data.file_uri or "",
                    mimeType=part.file_data.mime_type,
                ),
            ),
        )
    if part.inline_data:
        return Part(
            root=FilePart(
                file=FileWithBytes(
                    bytes=base64.b64encode(
                        part.inline_data.data,  # type: ignore
                    ).decode(),
                    mimeType=part.inline_data.mime_type,
                ),
            ),
        )
    raise ValueError(f"Unsupported part type: {part}")

Testing Your Implementation

Once your executor is ready, test it with a simple client:

from a2a.client import A2AClient

async def test_your_agent():
    async with httpx.AsyncClient() as client:
        a2a_client = await A2AClient.get_client_from_agent_card_url(
            client, "http://localhost:your-port"
        )

        response = await a2a_client.send_message(
            SendMessageRequest(
                id="test-123",
                params=MessageSendParams(
                    message=Message(
                        role="user",
                        parts=[TextPart(text="Hello, agent!")],
                        messageId="msg-123"
                    )
                )
            )
        )
        print(f"Agent response: {response}")

Advanced Usage: CLI for CI/CD 🚀

For automated testing in your deployment pipeline, use Rogue's CLI mode:

Basic CLI Usage

after creating the scenario and business context rogue will create a .rogue folder with the scenarios.json file that contains the scenarios

# Run evaluation with minimal config
uvx rogue-ai cli \
    --evaluated-agent-url http://localhost:10001 \
    --judge-llm openai/o4-mini \
    --workdir './examples/tshirt_store_agent/.rogue'

CI/CD Integration Example

# .github/workflows/agent-evaluation.yml
name: Rogue

on:
  pull_request:
  push:
    branches:
      - main
  workflow_dispatch:

jobs:
  run_rogue:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    steps:
      - name: Run rogue
        shell: bash
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |          
          echo "🚀 Starting rogue server and example agent..."

          uvx rogue-ai server --host 0.0.0.0 --port 8000 --example tshirt_store &
          ROGUE_PID=$!
          echo "Rogue server started with PID: $ROGUE_PID"
          trap 'echo "🛑 Stopping rogue server..."; kill $ROGUE_PID' EXIT

          echo "⏳ Waiting for example agent to be ready..."
          curl --retry 10 --retry-delay 5 --retry-connrefused -s --fail -o /dev/null \
            http://localhost:10001/.well-known/agent.json

          echo "⏳ Waiting for rogue server to be ready..."
          curl --retry 10 --retry-delay 5 --retry-connrefused -s --fail -o /dev/null \
            http://localhost:8000/api/v1/health

          echo "🚀 Rogue server ready"

          echo "🚀 Running rogue..."
          uvx rogue-ai cli \
            --evaluated-agent-url http://localhost:10001 \
            --judge-llm openai/o4-mini \
            --workdir './examples/tshirt_store_agent/.rogue'

Tips for Effective Agent Testing 💡

1. Write Comprehensive Business Context

Your business context directly impacts scenario quality. Include:

Policies: What your agent should/shouldn't do
Available actions: Tools and capabilities
Expected behavior: How to handle edge cases
Constraints: Pricing, inventory, limitations

2. Start with Core Scenarios

Begin testing with your most critical use cases:

Happy path interactions
Policy violations
Edge cases and error handling
Security boundary testing

3. Iterate Based on Results

Use evaluation reports to improve your agent:

Fix failed scenarios
Add safeguards for edge cases
Refine instructions based on findings
Re-test after changes

4. Automate in CI/CD

Make evaluation part of your deployment process:

Run on every pull request
Block deployments on failed evaluations
Track evaluation metrics over time
Alert on regression

Happy testing! 🧪✨

Ranger-mini: Open Model for Agentic tool use evaluation

drorIvry — Wed, 27 Aug 2025 12:55:21 +0000

ranger-mini, a fine-tuned sequence classification model, inspects tool calls then returns a single, precise label describing the error or confirming the call is valid. It is optimized for low latency and high accuracy.

What Ranger-mini does

ranger-mini evaluates whether a proposed function call:

Chooses the correct tool for the user intent,
Uses the exact parameter names required by the tool schema,
Supplies correctly formatted and accurate parameter values.

It returns one of four labels:

✅VALID_CALL The tool name, parameters, and values are correct, or no tool is needed.
❌TOOL_ERROR The tool name is missing or does not match the user intent.
❌PARAM_NAME_ERROR Parameter names are missing, extra, or mismatched.
❌PARAM_VALUE_ERROR Parameter names match, but values are incorrect or malformed.

Benchmarks

Model	#Params	Avg. Latency	Avg Binary Accuracy	Qualifire TSQ Benchmark Binary Accuracy	Limbic Benchmark Binary Accuracy
qualifire/mcp-tool-use-quality-ranger-4b [private]	4B	0.30[sec]	0.978	0.997	0.960
qualifire/mcp-tool-use-quality-ranger-0.6b	0.6B	0.09[sec]	0.958	0.993	0.924
gemini-2.5-flash	-	4.87[sec]	0.890	0.936	0.845
quotientai/limbic-tool-use-0.5B-32K	0.5B	0.79[sec]	0.807	0.752	0.862

Key Insight: Ranger-mini delivers near state-of-the-art tool-use accuracy at a fraction of latency and model size, making it practical for production guardrails.

Usage

from transformers import AutoModelForSequenceClassification, AutoTokenizer, pipeline
import torch
from huggingface_hub import hf_hub_download

# Model name
model_name = "qualifire/mcp-tool-use-quality-ranger-0.6b"

# Map raw labels to human-readable labels
map_id_to_label = {
    'LABEL_0': 'VALID_CALL',
    'LABEL_1': 'TOOL_ERROR',
    'LABEL_2': 'PARAM_NAME_ERROR',
    'LABEL_3': 'PARAM_VALUE_ERROR'
}

# Load model and tokenizer
model = AutoModelForSequenceClassification.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map='auto',
)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# Create pipeline
pipe = pipeline("text-classification", model=model, tokenizer=tokenizer)

# Load prompt template
file_path = hf_hub_download(repo_id=model_name, filename="tsq_prompt_template.txt")
with open(file_path, encoding="utf-8") as f:
    PROMPT_TEMPLATE = f.read()

# Example inputs
example_tools_list = '''[
  {
    "type": "function",
    "function": {
      "name": "send-email",
      "description": "Send an email using Resend",
      "parameters": {
        "properties": {
          "to": {
            "type": "string",
            "format": "email",
            "description": "Recipient email address"
          },
          "content": {
            "type": "string",
            "description": "Plain text email content"
          },
          "subject": {
            "type": "string",
            "description": "Email subject line"
          },
          "scheduledAt": {
            "type": "string",
            "description": "Optional parameter to schedule the email. This uses natural language. Examples would be 'tomorrow at 10am' or 'in 2 hours' or 'next day at 9am PST' or 'Friday at 3pm ET'."
          }
        },
        "required": ["to", "subject", "content"]
      }
    }
  }
]'''

example_message_history = '''[
  {
    "role": "user",
    "content": "Please send an email to 'jane.doe@example.com' with the subject 'Meeting Follow-Up'. The content should be 'Hi Jane, just following up on our meeting from yesterday. Please find the attached notes.' and schedule it for tomorrow at 10am."
  },
  {
    "completion_message": {
      "content": {
        "type": "text",
        "text": ""
      },
      "role": "assistant",
      "stop_reason": "tool_calls",
      "tool_calls": [
        {
          "id": "call_le25efmhltxx9o7n4rfe",
          "function": {
            "name": "send-email",
            "arguments": {
              "subject": "Meeting Follow-Up",
              "content": "Hi Jane, just following up on our meeting from yesterday. Please find the attached notes.",
              "scheduledAt": "tomorrow at 10am"
            }
          }
        }
      ]
    }
  }
]'''

# Format input
example_input = PROMPT_TEMPLATE.format(
    message_history=example_message_history,
    available_tools=example_tools_list
)

# Get prediction
result = pipe(example_input)[0]
result['label'] = map_id_to_label[result['label']]
print(result)

✨ Example Output

{'label': 'PARAM_NAME_ERROR', 'score': 0.9999843835830688}

🛡️ Paladin-mini: Open-Source Grounding Model That Actually Works in Production

drorIvry — Thu, 17 Jul 2025 10:46:15 +0000

As developers, we've all been there. You build a RAG system, deploy it to production, and then realize your AI is confidently telling users that 2+2=5 or that Christmas is in July. The problem? Most fact checking models are trained on academic datasets that don't reflect real world edge cases.

That's why we built Paladin-mini, a compact, efficient grounding model specifically designed for production environments where accuracy matters.

What Makes Paladin-mini Different?

Unlike general-purpose fact checking models, Paladin-mini is trained on synthetic data targeting the exact types of errors that break production systems:

Mathematical calculations (pricing, quantities, percentages)
Temporal reasoning (dates, schedules, sequences)
Logical consistency (technical specifications, domain rules)
Real-world edge cases (the stuff that academic benchmarks miss)

Performance That Matters

Here's where Paladin-mini shines compared to larger models:

Category	Paladin-mini (3.8B)	Bespoke-MiniCheck-7B	Improvement
Prices & Math	96.0%	46.0%	+50%
Logical Reasoning	97.1%	92.8%	+4.3%
General Tasks	91.97%	84.02%	+7.95%
Overall Average	79.31%	77.86%	+1.45%

Key insight: Paladin-mini with 3.8B parameters outperforms a 7B model while being 2x smaller and significantly faster.

Paladin uses the terms doc and claim

docs - are context document providing the ground truth (either from the RAG, system prompt, tool response etc...)

claim - the content usually generated by an AI using said document.

The claim can be either grounded or ungrounded in the document, meaning that each claim can be proven or inferred using the document alone.

🚀 Getting Started

Installation

pip install torch transformers accelerate

Basic Usage

Loading the model for later usage

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline

# Load the model
model_name = "qualifire/context-grounding-paladin-mini"
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map='auto',
)

tokenizer = AutoTokenizer.from_pretrained(model_name)

# Create pipeline
pipe = pipeline(
    "text-generation",
    model=model,
    tokenizer=tokenizer,
    max_new_tokens=1,
    return_full_text=False,
    temperature=0.0,
    do_sample=False
)

The Prompt Template

Paladin-mini uses a specific prompt format for optimal performance:

PROMPT_TEMPLATE = '''
You are tasked with determining whether a given claim is consistent with the information provided in a document. Consistency means that all information in the claim is supported by the document. If any part of the claim contradicts or is not substantiated by the document, it should be considered inconsistent.

Analyze the claim in relation to the information provided in the document. Consider the following:
1. Does the document explicitly support all parts of the claim?
2. Is there any information in the claim that contradicts the document?
3. Does the claim contain any details not mentioned in the document?

Before providing your reasoning, give your final answer as either "Yes" (the claim is consistent with the document) or "No" (the claim is not consistent with the document). The reasoning should follow the final answer.

The answer should begin with a single word: "Yes" or "No".

---

First, carefully read the following document:

<DOCUMENT>
{doc}
</DOCUMENT>

Now, consider this claim:

<CLAIM>
{claim}
</CLAIM>

What is your answer?'''

Core Verification Function

def verify_claim(document, claim):
    """Generic function to verify any claim against a document"""
    prompt = PROMPT_TEMPLATE.format(doc=document, claim=claim)

    messages = [{"role": "user", "content": prompt}]
    result = pipe(messages)

    return result[0]['generated_text'].strip()

Real-World Examples

Example 1: E-commerce Price Verification

# Test case that breaks other models
doc = """
Product: Gaming Laptop
Base Price: $1,200
Student Discount: 10%
Bulk Order (3+ items): Additional 5% off
Tax Rate: 8.25%
"""

claim = "A student buying 3 gaming laptops would pay $3,105.84 total including tax"

# Let's verify: 
# Base: $1,200 × 3 = $3,600
# Student discount: $3,600 × 0.9 = $3,240  
# Bulk discount: $3,240 × 0.95 = $3,078
# With tax: $3,078 × 1.0825 = $3,331.94
# The claim is wrong!

result = verify_claim(doc, claim)
print(result)  # Output: "No"

Example 2: Date Validation for Scheduling

doc = """
Conference Schedule:
- Registration opens: March 15, 2024
- Early bird deadline: April 30, 2024
- Conference dates: June 10-12, 2024
- Abstract submission closes: May 1, 2024
"""

claim = "Abstract submissions are due before the early bird registration deadline"

result = verify_claim(doc, claim)
print(result)  # Output: "No" (abstracts due May 1, early bird ends April 30)

Example 3: Technical Specification Verification

doc = """
API Rate Limits:
- Free tier: 1,000 requests/hour
- Pro tier: 10,000 requests/hour  
- Enterprise: 100,000 requests/hour
- Burst limit: 2x rate limit for 60 seconds
"""

claim = "Pro tier users can make up to 20,000 requests during a 60-second burst period"

result = verify_claim(doc, claim)
print(result)  # Output: "Yes" (10,000 × 2 = 20,000)

Production Integration Patterns

Pattern 1: RAG Response Validation

class RAGValidator:
    def __init__(self):
        self.grounding_pipe = pipe  # Your Paladin-mini pipeline

    def verify_claim(self, document, claim):
        """Verify a single claim against a document"""
        prompt = PROMPT_TEMPLATE.format(doc=document, claim=claim)
        messages = [{"role": "user", "content": prompt}]
        result = self.grounding_pipe(messages)
        return result[0]['generated_text'].strip()

    def validate_response(self, context_docs, generated_response):
        """Validate a RAG response against context documents"""
        # Combine all context documents
        full_context = "\n\n".join(context_docs)

        # Check if response is grounded
        result = self.verify_claim(full_context, generated_response)

        return {
            'is_grounded': result.lower() == 'yes',
            'response': generated_response,
            'confidence': 'high' if result.lower() in ['yes', 'no'] else 'low'
        }

Pattern 2: Batch Verification for Content Pipelines

def batch_verify_claims(documents, claims):
    """Verify multiple claims against their corresponding documents"""
    results = []

    for doc, claim in zip(documents, claims):
        prompt = PROMPT_TEMPLATE.format(doc=doc, claim=claim)
        messages = [{"role": "user", "content": prompt}]

        result = pipe(messages)
        is_grounded = result[0]['generated_text'].strip().lower() == 'yes'

        results.append({
            'claim': claim,
            'is_grounded': is_grounded,
            'document': doc[:100] + "..." if len(doc) > 100 else doc
        })

    return results

Pattern 3: Real-time API Guard

from flask import Flask, request, jsonify

app = Flask(__name__)

# Initialize validator
validator = RAGValidator()

@app.route('/verify', methods=['POST'])
def verify_claim_endpoint():
    data = request.json
    document = data.get('document')
    claim = data.get('claim')

    if not document or not claim:
        return jsonify({'error': 'Missing document or claim'}), 400

    result = validator.verify_claim(document, claim)

    return jsonify({
        'is_grounded': result.lower() == 'yes',
        'raw_output': result
    })

if __name__ == '__main__':
    app.run(debug=True)

⏱️ Performance Optimization Tips

1. Memory Management

# Use smaller precision for inference
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,  # Reduces memory usage
    device_map='auto',
)

2. Batch Processing

# Process multiple claims efficiently
def batch_process(docs_and_claims):
    """Process multiple document-claim pairs in batch"""
    prompts = [PROMPT_TEMPLATE.format(doc=doc, claim=claim) 
              for doc, claim in docs_and_claims]

    messages_batch = [[{"role": "user", "content": prompt}] 
                      for prompt in prompts]

    results = pipe(messages_batch)
    return [r[0]['generated_text'].strip() for r in results]

3. Caching for Repeated Queries

from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def cached_verify(doc_hash, claim_hash, full_doc, full_claim):
    """Cache verification results for repeated queries"""
    # In production, you might want to use Redis or similar
    return verify_claim(full_doc, full_claim)

def verify_with_cache(document, claim):
    """Verify claim with caching support"""
    doc_hash = hashlib.md5(document.encode()).hexdigest()
    claim_hash = hashlib.md5(claim.encode()).hexdigest()

    return cached_verify(doc_hash, claim_hash, document, claim)

📈 Why This Matters for Production

The research behind Paladin-mini reveals a critical insight: general benchmarks don't predict real world performance. Models can score 90%+ on academic datasets while failing on simple math problems.

Our specialized training on synthetic data targeting real world edge cases means:

✅ Reliable financial calculations (no more embarrassing pricing errors)
✅ Accurate temporal reasoning (proper date/time handling)
✅ Robust logical consistency (handles domain-specific rules)
✅ Production-ready latency (70ms vs 7 seconds for larger models)

🔗 Resources

Model: Hugging Face
Paper: Paladin-mini: A Compact and Efficient Grounding Model
Benchmark: Grounding-Benchmark Dataset

What's Next?

If you're looking for a hosted version of this model and its bigger brothers go to https://qualifire.ai

Try Paladin-mini in your next project and let us know how it performs! We're particularly interested in:

Novel use cases and integration patterns
Performance comparisons with other models
Edge cases we should add to our training data

Have you used grounding models in production? What challenges have you faced? Share your experiences in the comments!

Built with ❤️ by the Qualifire team. We're on a mission to make AI systems more reliable and trustworthy for production use.

From Code to Clout with build-influence

drorIvry — Sat, 03 May 2025 17:37:37 +0000

As developers, striking a balance between rigorous code analysis and engaging technical storytelling is a perpetual challenge. build-influence bridges that gap by combining automated code repository analysis with AI-driven content generation, enabling you to share your technical achievements on platforms like dev.to, Twitter/X, LinkedIn, and more.

What Is build-influence?

build-influence is an open-source powerhouse built to simplify technical communication by automating both code review and content creation. Its standout features include:

• Automated Code Repository Analysis: Scan and interpret multiple code and documentation files across your projects instantly.

• AI-Powered Content Creation: Automatically generate detailed, engaging technical narratives that speak directly to your audience.

• Command-Line Interface (CLI): Benefit from a streamlined terminal-based workflow that boosts productivity without the overhead.

• Multi-Platform Publishing: Seamlessly share your content across dev.to, Twitter/X, LinkedIn, and other key platforms.

• Content Review & Approval Workflow: Ensure every publication is polished with an integrated review process.

• Workflow Automation & User Interaction: Automate repetitive tasks, letting you focus more on innovation.

• Integration with Social Media APIs: Extend your outreach by directly connecting with your favorite networks.

• Extensible Platform-Specific Generators: Customize content generation to satisfy the particular nuances of each platform.

• Open-Source Collaboration & Contributor Guidelines: Join a community dedicated to continuous improvement and collective knowledge sharing.

How Does build-influence Work?

build-influence isn’t simply a script—it’s a fully integrated platform that elevates your technical storytelling process. The workflow has been meticulously designed for efficiency and clarity:

Each step—from analyzing repositories to publishing the final narrative—is crafted to produce precise, compelling technical content, so you can concentrate on your core coding work and innovation.

Getting Started: Installation & Usage

Kick off your journey with build-influence by installing it via pip and setting up your environment with the necessary API keys.

Installation

Install build-influence with pip:

pip install build-influence

Then, configure your environment by exporting your key variables:

export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
export DEVTO_API_KEY=<YOUR_DEVTO_API_KEY>

Usage

Once installed and configured, using build-influence is straightforward. Simply run the following command to start the entire process—from code repository analysis to generating a draft technical narrative, which then moves into a seamless review workflow for multi-platform publication:

build-influence

Why Choose build-influence?

build-influence empowers you to bridge software development with technical content marketing effortlessly. It allows you to focus on writing robust and efficient code while automating the creation of engaging technical narratives. Engineered by developers for developers, this tool integrates naturally into your workflow, eliminating the gap between your coding efforts and technical storytelling. Whether you’re publishing on dev.to or other social media platforms, build-influence ensures that your technical communication is both compelling and efficient.

consisTent: A "Unit Testing" Framework for Prompts

drorIvry — Fri, 28 Apr 2023 13:05:27 +0000

Testing machine learning models that use large language models (LLMs) for natural language processing can be challenging. However, consisTent, a new testing framework, aims to make this task more straightforward and reproducible.

The goal of consisTent is to create reproducible tests for LLM based applications regardless of the fine-tuning method used. consisTent comes with two types of validators, Syntactic and Semantic Validators. The former is used to do static assertions of the LLM output, validating if the output is in a certain JSON format, asserting the schema, or checking if the output is a valid piece of code. On the other hand, Semantic Validators are used to assert the quality of the response with more "soft" parameters, such as checking if something is factually correct, checking for hallucinations, or checking for labels like "funny" or "interesting." The Semantic Consistency Validator is another type of semantic validator that allows developers to provide a seed of validated input and a threshold to assert the semantic distance of the new output from the seed cluster.

Syntactic Validators are used to assert the FORM of the response, whereas Semantic Validators are used to assert the CONTENT of the response.

examples

Let's take a look at some examples of how consisTent can be used to validate LLM output.

Syntactic Validator Examples:

consisTent.JsValidator().validate('console.log("I\'m a JS program!")')

consisTent.PyValidator().validate('print("I\'m a Python program!")')

Semantic Validator Example:

import consisTent

seed = [
    "the cat sat on the mat",
    "the feline layed on the carpet",
]

consisTent.ConsistencyValidator(
    seed_size=2,
    consistency_threshold=0.5,
).validate(
    seed=seed,
    model_output="the dog sat on the mat",
)

Label Test Example:

OPENAI_KEY = "XXXXXXXXXXXXXXX"

consisTent.LabelsValidator(openai_key=OPENAI_KEY).validate(
    labels=[        
        "funny",        
        "short",        
        "about rabbits",
    ],
    model_output="What do you call a rabbit that tells jokes? A funny bunny!",
)

Facts Validation Example:

OPENAI_KEY = "XXXXXXXXXXXXXXX"

consisTent.FactsValidator(openai_key=OPENAI_KEY).validate(
    facts=["this car weighs 1000KG"],
    model_output="The average person can lift this car can lift this car",
)

Installation of the consisTent package is easy via pip: pip install consistent .

Overall, consisTent is an excellent framework for testing LLM based applications. The framework's ability to handle syntactic and semantic validations in a reproducible manner makes it a valuable tool for anyone working with large language models.

Psst, please star us on github

Introducing Rego: The Lightweight Task Orchestrator for Kubernetes

drorIvry — Wed, 19 Apr 2023 12:38:48 +0000

👋 Hey there! Today I want to introduce you to Rego - the blazingly fast API first task orchestrator that is designed to allow asynchronous workloads to be deployed over Kubernetes with minimal effort.
🔥 Rego is lightweight and super fast, making it an excellent choice for those who need a controlled way to run code on demand. Plus, it has an API to manage runs as well as run logs, so it's a very well-rounded tool.

Why We Created Rego

We created Rego because we were looking for a managed way to "go run this docker for me" and didn't like anything that we found. We had to have an API to manage the runs as well as run logs. Rego is the result of that effort.

Use Cases

Rego has several use cases, including:
🔌 Running async workloads that need to be managed by (or visible to) a UI
🧪 Integrating non-production-grade code (like a data scientist's R code, for example) within your production environment in a contained way
️⏰ Using Rego to run stuff periodically with run history

Quick Start

Getting started with Rego is easy! Follow the steps below to get started with the CLI or the API.
Installing using k8s
Deploy Rego to your Kubernetes cluster by running the following command:

kubectl apply -f https://raw.githubusercontent.com/drorIvry/rego/main/deploy/deployment.yml

P.S. don't forget to portforward
👀 there are more ways to deploy rego, check the docs

Using the CLI

The Rego CLI is the easiest way to start using Rego. Follow these steps to get started:
Install the Rego CLI by running the following command:

curl -L https://raw.githubusercontent.com/drorIvry/rego-cli/main/install.sh | sh

Test the connection by running the following command:

rego ping

Deploy a Docker image by running the following command:

rego run -i hello-world

For more complex deployments, use the following command as a template:

rego run -d "$(cat << EOF 
{
  "image": "hello-world",
  "name": "test",
  "namespace": "default",
  "execution_interval": 0,
  "metadata": {
    "ttlSecondsAfterFinished": 10,
    "completions": 10,
    "parallelism": 10
  }
}
EOF
)"

Using the API

With Rego, you can use the API to create and run Kubernetes jobs with a managed API.
Create a new task by making a POST request to the following endpoint:

curl -X POST -d '{"image": "hello-world"}' localhost:4004/api/v1/task

This will start a job on your Kubernetes cluster that runs the Docker image hello-world. The response will include the definition ID, which you can use to see the task's running status.

{
  "definition_id": "a36fbd9b-bf8a-4c59–94c1–9938b6707e8f",
  "message": "created"
}

Check the task's running status by making a GET request to the following endpoint:

curl http://localhost:4004/api/v1/task/a36fbd9b-bf8a-4c59-94c1-9938b6707e8f/latest

This will return the task's running status, including the status code and status message.

{
  "ID": 0,
  "CreatedAt": "2023–04–02T11:53:08.2008054+03:00",
  "UpdatedAt": "2023–04–02T11:53:16.2032147+03:00",
  "DeletedAt": null,
  "id": "7eb53d97–7380–4e0b-82a6-b38fbf9119d2",
  "task_definition_id": "a36fbd9b-bf8a-4c59–94c1–9938b6707e8f",
  "status_code": 500, // rego internal status code
  "status": "SUCCESS",
  "image": "hello-world",
  "name": "test",
  "namespace": "test",
  "metadata": {
    "ttlSecondsAfterFinished": 10,
    "completions": 10,
    "parallelism": 10
  }
}

In conclusion, Rego is a powerful and flexible task orchestrator that can run your workloads with ease, while also providing a management API that can keep track of progress and run history. Whether you need to run async workloads that need to be managed, integrate non-production-grade code within your production environment in a contained way, or run stuff periodically with run history, Rego has got you covered.
And please ⭐star us on Github