Spaces:
Runtime error
Runtime error
File size: 3,841 Bytes
44de0e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | 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)
|