Upload folder using huggingface_hub

README.md

@@ -1,13 +1,16 @@
 ---
-title: Agentic Guardrails Retries
-emoji: 📉
-colorFrom: blue
-colorTo: gray
+title: Guardrails & Retries
+emoji: 🛡️
+colorFrom: red
+colorTo: yellow
 sdk: gradio
-sdk_version: 6.15.2
-python_version: '3.13'
+sdk_version: 4.44.0
 app_file: app.py
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Guardrails & retries
+
+Input guardrail + schema-validated output with retry-on-failure. Bring your own [OpenRouter](https://openrouter.ai/keys) key.
+
+Source & write-up: https://github.com/shenmali/agentic-ai-first/tree/main/demos/06-guardrails-retries

_core/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

_core/llm.py

@@ -0,0 +1,42 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+from openai import OpenAI
+
+
+@dataclass
+class LLMResponse:
+    content: str | None
+    tool_calls: list[dict] = field(default_factory=list)
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+
+
+class LLMClient:
+    BASE_URL = "https://openrouter.ai/api/v1"
+
+    def __init__(self, api_key: str, model: str):
+        if not api_key:
+            raise ValueError("An OpenRouter API key is required.")
+        self.model = model
+        self._client = OpenAI(api_key=api_key, base_url=self.BASE_URL)
+
+    def chat(self, messages: list[dict], tools: list[dict] | None = None) -> LLMResponse:
+        kwargs: dict[str, Any] = {"model": self.model, "messages": messages}
+        if tools:
+            kwargs["tools"] = tools
+        resp = self._client.chat.completions.create(**kwargs)
+        msg = resp.choices[0].message
+        tool_calls = []
+        if getattr(msg, "tool_calls", None):
+            for tc in msg.tool_calls:
+                tool_calls.append(
+                    {"id": tc.id, "name": tc.function.name, "arguments": tc.function.arguments}
+                )
+        usage = resp.usage
+        return LLMResponse(
+            content=msg.content,
+            tool_calls=tool_calls,
+            prompt_tokens=getattr(usage, "prompt_tokens", 0) if usage else 0,
+            completion_tokens=getattr(usage, "completion_tokens", 0) if usage else 0,
+        )

_core/models.py

@@ -0,0 +1,41 @@
+from dataclasses import dataclass
+
+# NOTE: model ids and prices reflect OpenRouter's catalog. Verify/refresh against
+# https://openrouter.ai/models before launch — ids change as providers release models.
+
+
+@dataclass(frozen=True)
+class ModelInfo:
+    id: str
+    label: str
+    prompt_cost: float       # USD per 1M prompt tokens
+    completion_cost: float   # USD per 1M completion tokens
+
+
+MODELS: list[ModelInfo] = [
+    ModelInfo("openai/gpt-4o-mini", "GPT-4o mini (cheap)", 0.15, 0.60),
+    ModelInfo("anthropic/claude-3.5-haiku", "Claude 3.5 Haiku (cheap)", 0.80, 4.00),
+    ModelInfo("google/gemini-flash-1.5", "Gemini 1.5 Flash (cheap)", 0.075, 0.30),
+    ModelInfo("deepseek/deepseek-chat", "DeepSeek Chat (cheap)", 0.14, 0.28),
+    ModelInfo("anthropic/claude-3.5-sonnet", "Claude 3.5 Sonnet (strong)", 3.00, 15.00),
+    ModelInfo("openai/gpt-4o", "GPT-4o (strong)", 2.50, 10.00),
+]
+
+DEFAULT_MODEL = "openai/gpt-4o-mini"
+
+
+def model_ids() -> list[str]:
+    return [m.id for m in MODELS]
+
+
+def get_model(model_id: str) -> ModelInfo | None:
+    return next((m for m in MODELS if m.id == model_id), None)
+
+
+def estimate_cost(model_id: str, prompt_tokens: int, completion_tokens: int) -> float:
+    m = get_model(model_id)
+    if m is None:
+        return 0.0
+    return (prompt_tokens / 1_000_000) * m.prompt_cost + (
+        completion_tokens / 1_000_000
+    ) * m.completion_cost

_core/tools.py

@@ -0,0 +1,112 @@
+import ast
+import operator
+from collections.abc import Callable
+from dataclasses import dataclass
+
+
+@dataclass
+class Tool:
+    name: str
+    description: str
+    parameters: dict
+    fn: Callable[..., str]
+
+
+class ToolRegistry:
+    def __init__(self) -> None:
+        self._tools: dict[str, Tool] = {}
+
+    def register(self, tool: Tool) -> None:
+        self._tools[tool.name] = tool
+
+    def to_openai_schema(self) -> list[dict]:
+        return [
+            {
+                "type": "function",
+                "function": {
+                    "name": t.name,
+                    "description": t.description,
+                    "parameters": t.parameters,
+                },
+            }
+            for t in self._tools.values()
+        ]
+
+    def execute(self, name: str, args: dict) -> str:
+        if name not in self._tools:
+            return f"Error: unknown tool '{name}'"
+        try:
+            return self._tools[name].fn(**args)
+        except Exception as e:  # tool failures must not crash the agent loop
+            return f"Error executing {name}: {e}"
+
+
+_SAFE_OPS = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+    ast.Pow: operator.pow,
+    ast.Mod: operator.mod,
+    ast.USub: operator.neg,
+}
+
+
+def _safe_eval(node: ast.AST) -> int | float:
+    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
+        return node.value
+    if isinstance(node, ast.BinOp) and type(node.op) in _SAFE_OPS:
+        left = _safe_eval(node.left)
+        right = _safe_eval(node.right)
+        if isinstance(node.op, ast.Pow) and abs(right) > 100:
+            raise ValueError("exponent too large")
+        return _SAFE_OPS[type(node.op)](left, right)
+    if isinstance(node, ast.UnaryOp) and type(node.op) in _SAFE_OPS:
+        return _SAFE_OPS[type(node.op)](_safe_eval(node.operand))
+    raise ValueError("unsupported expression")
+
+
+def calculate(expression: str) -> str:
+    try:
+        tree = ast.parse(expression, mode="eval")
+        return str(_safe_eval(tree.body))
+    except Exception as e:
+        return f"Error: {e}"
+
+
+def make_calculator() -> Tool:
+    return Tool(
+        name="calculator",
+        description="Evaluate a basic arithmetic expression, e.g. '2 * (3 + 4)'.",
+        parameters={
+            "type": "object",
+            "properties": {"expression": {"type": "string"}},
+            "required": ["expression"],
+        },
+        fn=lambda expression: calculate(expression),
+    )
+
+
+def web_search(query: str, max_results: int = 3) -> str:
+    try:
+        from ddgs import DDGS
+    except ImportError:  # package was renamed; support both
+        from duckduckgo_search import DDGS  # type: ignore
+    results = []
+    with DDGS() as ddgs:
+        for r in ddgs.text(query, max_results=max_results):
+            results.append(f"- {r.get('title', '')}: {r.get('body', '')}")
+    return "\n".join(results) if results else "No results found."
+
+
+def make_web_search() -> Tool:
+    return Tool(
+        name="web_search",
+        description="Search the web for current information. Returns the top results.",
+        parameters={
+            "type": "object",
+            "properties": {"query": {"type": "string"}},
+            "required": ["query"],
+        },
+        fn=lambda query: web_search(query),
+    )

_core/tracer.py

@@ -0,0 +1,27 @@
+from dataclasses import dataclass, field
+from typing import Literal
+
+StepKind = Literal["thought", "action", "observation", "final"]
+
+
+@dataclass
+class Step:
+    kind: StepKind
+    content: str
+    tokens: int = 0
+    cost_usd: float = 0.0
+    latency_ms: int = 0
+
+
+@dataclass
+class Trace:
+    steps: list[Step] = field(default_factory=list)
+
+    def add(self, step: Step) -> None:
+        self.steps.append(step)
+
+    def total_tokens(self) -> int:
+        return sum(s.tokens for s in self.steps)
+
+    def total_cost(self) -> float:
+        return sum(s.cost_usd for s in self.steps)

_core/ui.py

@@ -0,0 +1,116 @@
+from collections.abc import Callable, Iterator
+
+import gradio as gr
+
+from _core.llm import LLMClient
+from _core.models import DEFAULT_MODEL, model_ids
+from _core.tracer import Step, Trace
+
+_KIND_ICON = {"thought": "💭", "action": "🔧", "observation": "👁️", "final": "✅"}
+
+
+def api_key_input() -> gr.Textbox:
+    return gr.Textbox(
+        label="OpenRouter API key",
+        type="password",
+        placeholder="sk-or-...",
+        info=(
+            "Get a key at https://openrouter.ai/keys — it stays in your browser"
+            " session and is never stored."
+        ),
+    )
+
+
+def model_selector() -> gr.Dropdown:
+    return gr.Dropdown(choices=model_ids(), value=DEFAULT_MODEL, label="Model")
+
+
+def render_step_markdown(step: Step) -> str:
+    icon = _KIND_ICON.get(step.kind, "•")
+    meta = ""
+    if step.tokens or step.cost_usd or step.latency_ms:
+        meta = f"  \n<sub>{step.tokens} tok · ${step.cost_usd:.4f} · {step.latency_ms} ms</sub>"
+    return f"**{icon} {step.kind.title()}**  \n{step.content}{meta}"
+
+
+def render_trace_markdown(trace: Trace) -> str:
+    return "\n\n---\n\n".join(render_step_markdown(s) for s in trace.steps)
+
+
+def metrics_summary(trace: Trace) -> str:
+    return (
+        f"**Total:** {trace.total_tokens()} tokens · "
+        f"${trace.total_cost():.4f} · {len(trace.steps)} steps"
+    )
+
+
+def build_agent_app(
+    *,
+    title: str,
+    description: str,
+    input_label: str,
+    input_placeholder: str,
+    run_fn: Callable[[LLMClient, str], Iterator[Step]],
+    example: str = "",
+) -> gr.Blocks:
+    """Build a standard single-input agent demo.
+
+    run_fn(llm, user_input) yields Steps. Key validation, LLM construction,
+    trace accumulation, rendering and error handling are handled here.
+    """
+
+    def _handler(api_key: str, model_id: str, user_input: str):
+        if not api_key:
+            yield "⚠️ Please enter your OpenRouter API key.", ""
+            return
+        try:
+            llm = LLMClient(api_key=api_key, model=model_id)
+        except Exception as e:
+            yield f"⚠️ {e}", ""
+            return
+        trace = Trace()
+        try:
+            for step in run_fn(llm, user_input):
+                trace.add(step)
+                yield render_trace_markdown(trace), metrics_summary(trace)
+        except Exception as e:
+            yield render_trace_markdown(trace) + f"\n\n⚠️ **Error:** {e}", metrics_summary(trace)
+
+    with gr.Blocks(title=title) as demo:
+        gr.Markdown(f"# {title}\n\n{description}")
+        with gr.Row():
+            key = api_key_input()
+            model = model_selector()
+        inp = gr.Textbox(label=input_label, placeholder=input_placeholder, value=example)
+        btn = gr.Button("Run agent", variant="primary")
+        trace_out = gr.Markdown()
+        metrics_out = gr.Markdown()
+        btn.click(_handler, inputs=[key, model, inp], outputs=[trace_out, metrics_out])
+    return demo
+
+
+def _truncate(text: str, limit: int = 60) -> str:
+    text = text.replace("\n", " ")
+    return text if len(text) <= limit else text[: limit - 1] + "…"
+
+
+def render_trace_table(trace: Trace) -> str:
+    header = (
+        "| # | Step | Tokens | Cost | Latency | Content |\n"
+        "|---|------|-------|------|---------|---------|"
+    )
+    rows = [
+        f"| {i + 1} | {s.kind} | {s.tokens} | ${s.cost_usd:.4f}"
+        f" | {s.latency_ms} ms | {_truncate(s.content)} |"
+        for i, s in enumerate(trace.steps)
+    ]
+    return "\n".join([header, *rows])
+
+
+def cost_breakdown(trace: Trace) -> str:
+    by_kind: dict[str, float] = {}
+    for s in trace.steps:
+        by_kind[s.kind] = by_kind.get(s.kind, 0.0) + s.cost_usd
+    lines = [f"- **{kind}**: ${cost:.4f}" for kind, cost in by_kind.items()]
+    lines.append(f"- **total**: ${trace.total_cost():.4f}")
+    return "\n".join(lines)

agent.py

@@ -0,0 +1,92 @@
+import json
+import time
+from collections.abc import Iterator
+
+from _core.llm import LLMClient
+from _core.models import estimate_cost
+from _core.tracer import Step
+
+TARGET_SCHEMA = '{"name": string, "age": integer, "skills": [string]}'
+GEN_PROMPT = (
+    f"Extract the person's info as JSON matching this schema: {TARGET_SCHEMA}. "
+    "Output ONLY the JSON object."
+)
+BANNED_PHRASES = ["ignore previous", "system prompt", "disregard instructions"]
+MAX_INPUT_CHARS = 2000
+
+
+def input_guardrail(text: str) -> str | None:
+    if len(text) > MAX_INPUT_CHARS:
+        return f"Input too long (max {MAX_INPUT_CHARS} characters)."
+    low = text.lower()
+    for phrase in BANNED_PHRASES:
+        if phrase in low:
+            return f"Input rejected: contains disallowed phrase '{phrase}'."
+    return None
+
+
+def validate(payload: dict) -> list[str]:
+    if not isinstance(payload, dict):
+        return ["output must be a JSON object"]
+    errors: list[str] = []
+    if not isinstance(payload.get("name"), str):
+        errors.append("'name' must be a string")
+    if not isinstance(payload.get("age"), int) or isinstance(payload.get("age"), bool):
+        errors.append("'age' must be an integer")
+    skills = payload.get("skills")
+    if not isinstance(skills, list) or not all(isinstance(s, str) for s in skills):
+        errors.append("'skills' must be a list of strings")
+    return errors
+
+
+class GuardedAgent:
+    def __init__(self, llm: LLMClient, max_retries: int = 2):
+        self.llm = llm
+        self.max_retries = max_retries
+
+    def run(self, user_input: str) -> Iterator[Step]:
+        blocked = input_guardrail(user_input)
+        if blocked:
+            yield Step(kind="final", content=f"⛔ {blocked}")
+            return
+
+        messages: list[dict] = [
+            {"role": "system", "content": GEN_PROMPT},
+            {"role": "user", "content": user_input},
+        ]
+        for attempt in range(self.max_retries + 1):
+            start = time.monotonic()
+            resp = self.llm.chat(messages)
+            latency = int((time.monotonic() - start) * 1000)
+            cost = estimate_cost(self.llm.model, resp.prompt_tokens, resp.completion_tokens)
+            yield Step(
+                kind="thought",
+                content=f"Attempt {attempt + 1}: {resp.content}",
+                tokens=resp.prompt_tokens + resp.completion_tokens,
+                cost_usd=cost,
+                latency_ms=latency,
+            )
+            try:
+                payload = json.loads(resp.content or "{}")
+                errors = validate(payload)
+            except json.JSONDecodeError as e:
+                payload = None
+                errors = [f"invalid JSON: {e}"]
+
+            if not errors:
+                yield Step(kind="observation", content="✅ Output passed validation.")
+                yield Step(kind="final", content=json.dumps(payload, indent=2))
+                return
+
+            yield Step(kind="observation", content="❌ Validation failed: " + "; ".join(errors))
+            messages.append({"role": "assistant", "content": resp.content or ""})
+            messages.append(
+                {
+                    "role": "user",
+                    "content": "Your output was invalid: "
+                    + "; ".join(errors)
+                    + ". Return corrected JSON only.",
+                }
+            )
+
+        yield Step(kind="final", content="Failed validation after all retries.")

app.py

@@ -0,0 +1,24 @@
+from _core.llm import LLMClient
+from _core.ui import build_agent_app
+from agent import GuardedAgent
+
+
+def run_fn(llm: LLMClient, user_input: str):
+    return GuardedAgent(llm=llm).run(user_input)
+
+
+demo = build_agent_app(
+    title="Guardrails & retries",
+    description=(
+        "Reliability patterns: an input guardrail blocks disallowed prompts, and the "
+        "agent validates its JSON output against a schema — retrying with the error "
+        "fed back until it's valid. Bring your own OpenRouter key."
+    ),
+    input_label="Describe a person (name, age, skills)",
+    input_placeholder="Ada Lovelace, 36, skilled in mathematics and programming.",
+    run_fn=run_fn,
+    example="Ada Lovelace, 36, skilled in mathematics and programming.",
+)
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,3 @@
+gradio>=4.44,<5
+huggingface_hub>=0.25,<1.0
+openai>=1.40