Upload folder using huggingface_hub

README.md

@@ -1,13 +1,17 @@
 ---
-title: Agentic React From Scratch
-emoji: 🚀
-colorFrom: pink
-colorTo: red
+title: ReAct From Scratch
+emoji: 🧠
+colorFrom: blue
+colorTo: indigo
 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
+# ReAct from scratch
+
+A Reasoning + Acting agent built from first principles, with a transparent
+step-by-step trace. Bring your own [OpenRouter](https://openrouter.ai/keys) key.
+
+Source & write-up: https://github.com/shenmali/agentic-ai-first/tree/main/demos/01-react-from-scratch

_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,81 @@
+import json
+import time
+from collections.abc import Iterator
+
+from _core.llm import LLMClient
+from _core.models import estimate_cost
+from _core.tools import ToolRegistry
+from _core.tracer import Step
+
+SYSTEM_PROMPT = (
+    "You are a ReAct agent. Reason step by step about the user's question. "
+    "Use a tool when you need external information or computation. "
+    "When you can answer directly, respond with the final answer and do NOT call a tool."
+)
+
+
+class ReActAgent:
+    def __init__(self, llm: LLMClient, tools: ToolRegistry, max_steps: int = 6):
+        self.llm = llm
+        self.tools = tools
+        self.max_steps = max_steps
+
+    def run(self, question: str) -> Iterator[Step]:
+        messages: list[dict] = [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": question},
+        ]
+        for _ in range(self.max_steps):
+            start = time.monotonic()
+            resp = self.llm.chat(messages, tools=self.tools.to_openai_schema())
+            latency = int((time.monotonic() - start) * 1000)
+            cost = estimate_cost(self.llm.model, resp.prompt_tokens, resp.completion_tokens)
+            step_tokens = resp.prompt_tokens + resp.completion_tokens
+
+            if resp.tool_calls:
+                messages.append(
+                    {
+                        "role": "assistant",
+                        "content": resp.content or "",
+                        "tool_calls": [
+                            {
+                                "id": tc["id"],
+                                "type": "function",
+                                "function": {"name": tc["name"], "arguments": tc["arguments"]},
+                            }
+                            for tc in resp.tool_calls
+                        ],
+                    }
+                )
+                if resp.content:
+                    yield Step(
+                        kind="thought",
+                        content=resp.content,
+                        tokens=step_tokens,
+                        cost_usd=cost,
+                        latency_ms=latency,
+                    )
+                for tc in resp.tool_calls:
+                    yield Step(kind="action", content=f"{tc['name']}({tc['arguments']})")
+                    try:
+                        args = json.loads(tc["arguments"]) if tc["arguments"] else {}
+                    except json.JSONDecodeError as e:
+                        # Thread the error back as an observation so the model can
+                        # recover, rather than aborting the whole run.
+                        result = f"Error: invalid tool arguments (not valid JSON): {e}"
+                    else:
+                        result = self.tools.execute(tc["name"], args)
+                    yield Step(kind="observation", content=result)
+                    messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})
+                continue
+
+            yield Step(
+                kind="final",
+                content=resp.content or "",
+                tokens=step_tokens,
+                cost_usd=cost,
+                latency_ms=latency,
+            )
+            return
+
+        yield Step(kind="final", content="Reached max steps without a final answer.")

app.py

@@ -0,0 +1,28 @@
+from _core.llm import LLMClient
+from _core.tools import ToolRegistry, make_calculator, make_web_search
+from _core.ui import build_agent_app
+from agent import ReActAgent
+
+
+def run_fn(llm: LLMClient, user_input: str):
+    tools = ToolRegistry()
+    tools.register(make_calculator())
+    tools.register(make_web_search())
+    return ReActAgent(llm=llm, tools=tools).run(user_input)
+
+
+demo = build_agent_app(
+    title="ReAct from scratch",
+    description=(
+        "The Reasoning + Acting loop, built from first principles. The agent "
+        "interleaves thinking, tool calls, and observations. Bring your own "
+        "OpenRouter key — it stays in your browser session."
+    ),
+    input_label="Question",
+    input_placeholder="What is 24 * 17? And who won the 2022 FIFA World Cup?",
+    run_fn=run_fn,
+    example="What is 24 * 17?",
+)
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

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