"""CraftPilot - AI-powered craft business assistant.
Upload a photo of your handmade craft, get a complete marketplace listing
with catalog metadata, product descriptions, social captions, and pricing.
Built for the Build Small Hackathon 2026.
Multi-agent pipeline: Vision -> Cataloger -> Copywriter + Pricer
Single model (MiniCPM-V 2.6, ~8B) running locally via llama.cpp.
No cloud APIs.
"""
import asyncio
import json
import logging
import os
import tempfile
import threading
import time
import gradio as gr
from huggingface_hub import HfApi, hf_hub_download
from PIL import Image
from agents import cataloger, copywriter, pricer
from agents.llm import LLMClient
from agents.models import AgentTrace, PipelineResult
from agents.pipeline import VISION_PROMPT
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# HF token: env var > token.txt fallback
hf_token = os.getenv("HF_TOKEN")
if not hf_token:
_token_path = os.path.join(os.path.dirname(__file__), "token.txt")
if os.path.exists(_token_path):
hf_token = open(_token_path).read().strip()
if hf_token:
os.environ["HUGGING_FACE_HUB_TOKEN"] = hf_token
TRACES_DATASET = "skamathramesh/craftpilot-traces"
def _upload_trace_async(trace_json: str):
"""Upload trace JSON to HF dataset in background thread."""
def _upload():
try:
api = HfApi(token=hf_token)
filename = f"trace_{int(time.time())}_{os.getpid()}.json"
api.upload_file(
path_or_fileobj=trace_json.encode(),
path_in_repo=f"traces/{filename}",
repo_id=TRACES_DATASET,
repo_type="dataset",
)
logger.info("Trace uploaded: %s", filename)
except Exception as e:
logger.warning("Trace upload failed (non-fatal): %s", e)
if hf_token:
threading.Thread(target=_upload, daemon=True).start()
CRAFT_TYPES = [
"Crochet",
"Embroidery",
"Painting",
"Sewing",
"Knitting",
"Other",
]
MODEL_REPO = os.getenv("MODEL_REPO", "openbmb/MiniCPM-V-2_6-gguf")
MODEL_FILE = os.getenv("MODEL_FILE", "ggml-model-Q4_K_M.gguf")
PROJ_FILE = os.getenv("PROJ_FILE", "mmproj-model-f16.gguf")
_llm_client: LLMClient | None = None
CUSTOM_HEAD = 'Raw JSON \n\n```json\n{trace_json}\n```\n']
parts.append(f'
{label}
')
parts.append('
')
for i, name in enumerate(steps):
if i > 0:
conn_cls = "step-connector done" if i <= active_step else "step-connector"
parts.append(f'
')
if i < active_step:
dot_cls, lbl_cls, icon = "step-dot done", "step-label done", "✓"
elif i == active_step:
dot_cls, lbl_cls, icon = "step-dot active", "step-label active", str(i + 1)
else:
dot_cls, lbl_cls, icon = "step-dot", "step-label", str(i + 1)
parts.append(f'
{icon}
')
parts.append(f'
{name}
')
parts.append("
{c.title} ")
parts.append(f"{c.short_desc}
")
if result.pricing:
p = result.pricing
parts.append(f"Suggested Price: ${p.suggested_price_min} \u2013 ${p.suggested_price_max} ")
parts.append("{t}" for t in cat.tags[:6])
parts.append(
f"Category: {cat.category} / {cat.sub_category}Materials: {', '.join(cat.materials)}Colors: {', '.join(cat.colors)}Complexity: {cat.complexity}
"
f"{tags}
"
)
parts.append("Description
{result.copy_data.long_desc}
")
parts.append("Instagram Captions
")
for cap in result.copy_data.captions:
parts.append(f"{cap} ")
parts.append(" ")
if result.pricing and result.pricing.reasoning:
parts.append(f"Pricing Rationale
{result.pricing.reasoning}
")
if result.pricing.cost_breakdown:
parts.append(f"Cost Breakdown: {result.pricing.cost_breakdown}
")
if result.traces:
agent_times = " \u2192 ".join(
f"{t.agent_name} ({t.duration_ms}ms)" for t in result.traces
)
parts.append(
f"Pipeline: {agent_times} "
f"| Total: {result.total_duration_ms}ms
"
)
return f'{"".join(parts)}
'
# Stores the last pipeline result for tab-select handlers.
_last_result: PipelineResult | None = None
_last_trace_path: str | None = None
_last_export_path: str | None = None
def analyze_craft(
image: Image.Image | None,
craft_type: str,
user_notes: str,
material_cost: str,
time_hours: str,
):
"""Streaming generator → gr.HTML (single output).
Uses gr.HTML to bypass Svelte 5 i18n bug that crashes gr.Markdown
during streaming. Tab content populated via tab-select handlers.
"""
global _last_result, _last_trace_path, _last_export_path
_last_result = None
_last_trace_path = None
_last_export_path = None
if image is None:
yield "Upload an image to get started.
"
return
if not craft_type:
yield "Please select a craft type.
"
return
cost = float(material_cost) if material_cost else None
hours = float(time_hours) if time_hours else None
notes = user_notes.strip() or None
try:
yield _status_html(-1, "Loading model...")
llm = get_client()
start = time.monotonic()
traces: list[AgentTrace] = []
result = PipelineResult(image_description="")
# Stage 1: Vision
yield _status_html(0, "Analyzing your craft...")
description, vision_ms = llm.describe_image(image, VISION_PROMPT)
traces.append(AgentTrace(
agent_name="vision", input_text="[image]",
output_data={"description": description},
duration_ms=vision_ms, model_id=llm.model_id,
))
result.image_description = description
result.traces = list(traces)
result.total_duration_ms = int((time.monotonic() - start) * 1000)
# Stage 2: Cataloger
yield (
_status_html(1, "Cataloging item...") +
f'Vision ({vision_ms}ms)
{description}
'
)
try:
catalog_result, catalog_trace = asyncio.run(
cataloger.run(llm, description, craft_type.lower(), notes)
)
traces.append(catalog_trace)
result.catalog = catalog_result
except Exception as e:
logger.warning("Cataloger failed: %s", e)
result.traces = list(traces)
result.total_duration_ms = int((time.monotonic() - start) * 1000)
# Stage 3: Copywriter
if result.catalog:
yield _status_html(2, "Writing copy...") + 'Error: {e}
"
def _on_vision_tab():
if not _last_result:
return ""
return _format_vision(_last_result)
def _on_catalog_tab():
if not _last_result:
return ""
return _format_catalog(_last_result)
def _on_copy_tab():
if not _last_result:
return "", gr.DownloadButton(visible=False)
return (
_format_copy(_last_result),
gr.DownloadButton(value=_last_export_path, visible=True) if _last_export_path else gr.DownloadButton(visible=False),
)
def _on_pricing_tab():
if not _last_result:
return ""
return _format_pricing(_last_result)
def _on_traces_tab():
if not _last_result:
return "", gr.DownloadButton(visible=False)
return (
_format_traces(_last_result),
gr.DownloadButton(value=_last_trace_path, visible=True) if _last_trace_path else gr.DownloadButton(visible=False),
)
def build_ui() -> gr.Blocks:
"""Build the Gradio interface."""
with gr.Blocks(
title="CraftPilot \u2014 AI Craft Business Assistant",
) as app:
# Header
gr.Markdown(
"# CraftPilot\n"
"### Photo in, marketplace listing out\n"
"Upload a photo of your handmade craft and get catalog data, "
"product copy, social captions, and fair pricing \u2014 all from a "
"single small model running locally. No cloud APIs.",
elem_id="craft-header",
)
with gr.Row(equal_height=False):
# Left: inputs
with gr.Column(scale=1, min_width=340):
with gr.Group(elem_id="input-panel"):
image_input = gr.Image(
label="Your Craft",
type="pil",
height=360,
)
craft_type = gr.Dropdown(
choices=CRAFT_TYPES,
label="Craft Type",
value="Crochet",
)
user_notes = gr.Textbox(
label="Notes (optional)",
placeholder="e.g. Made with organic cotton, took 3 evenings...",
lines=2,
)
with gr.Row():
material_cost = gr.Textbox(
label="Material Cost ($)",
placeholder="e.g. 15",
)
time_hours = gr.Textbox(
label="Time Spent (hrs)",
placeholder="e.g. 8",
)
analyze_btn = gr.Button(
"Analyze My Craft",
variant="primary",
size="lg",
elem_id="analyze-btn",
)
gr.Examples(
examples=[
["examples/crochet.png", "Crochet", "Handmade amigurumi with cotton yarn", "16", "6"],
["examples/embroidery.jpeg", "Embroidery", "Hand-stitched floral pattern on Jeans", "20", "10"],
["examples/sew_keychain.jpeg", "Sewing", "Fabric keychain with felt and thread", "5", "2"],
],
inputs=[image_input, craft_type, user_notes, material_cost, time_hours],
label="Try these examples",
)
# Right: outputs
with gr.Column(scale=2, min_width=500, elem_id="output-panel"):
with gr.Tabs():
with gr.Tab("Summary"):
# gr.HTML instead of gr.Markdown — experiment to
# bypass Svelte 5 i18n crash during streaming
summary_output = gr.HTML(
value="Your results will appear here after analysis.
",
elem_classes=["prose"],
)
with gr.Tab("Vision") as vision_tab:
vision_output = gr.Markdown(elem_classes=["prose"])
with gr.Tab("Catalog") as catalog_tab:
catalog_output = gr.Markdown(elem_classes=["prose"])
with gr.Tab("Copy") as copy_tab:
copy_output = gr.Markdown(elem_classes=["prose"])
export_download = gr.DownloadButton(
label="Export Listing (Etsy / Instagram)",
elem_id="export-btn",
visible=False,
)
with gr.Tab("Pricing") as pricing_tab:
price_output = gr.Markdown(elem_classes=["prose"])
with gr.Tab("Agent Traces") as traces_tab:
trace_output = gr.Markdown(elem_classes=["prose"])
trace_download = gr.DownloadButton(
label="Download Agent Trace JSON",
visible=False,
)
# Streaming generator → single gr.HTML output (bypasses i18n crash?)
analyze_btn.click(
fn=analyze_craft,
inputs=[
image_input,
craft_type,
user_notes,
material_cost,
time_hours,
],
outputs=[summary_output],
show_progress="minimal",
)
# Tab-select handlers: populate content + downloads on demand
vision_tab.select(fn=_on_vision_tab, outputs=[vision_output])
catalog_tab.select(fn=_on_catalog_tab, outputs=[catalog_output])
copy_tab.select(fn=_on_copy_tab, outputs=[copy_output, export_download])
pricing_tab.select(fn=_on_pricing_tab, outputs=[price_output])
traces_tab.select(fn=_on_traces_tab, outputs=[trace_output, trace_download])
# Footer
gr.Markdown(
"Built for the Build Small Hackathon 2026 \u00b7 Backyard AI track \n"
"MiniCPM-V 2.6 (~8B) via llama.cpp \u00b7 No cloud APIs \n"
"Llama Champion \u00b7 Off the Grid \u00b7 Sharing is Caring \u00b7 Field Notes",
elem_id="craft-footer",
)
return app
if __name__ == "__main__":
# Pre-load models at startup (avoids 60s request timeout on HF Spaces)
logger.info("Pre-loading models at startup...")
try:
get_client()
logger.info("Models ready!")
except Exception as e:
logger.error("Failed to load models: %s", e)
theme = gr.themes.Soft(
primary_hue="amber",
secondary_hue="orange",
neutral_hue="stone",
font=gr.themes.GoogleFont("Source Sans 3"),
font_mono=gr.themes.GoogleFont("JetBrains Mono"),
)
app = build_ui()
app.launch(
server_name="0.0.0.0",
server_port=7860,
theme=theme,
css=CUSTOM_CSS,
head=CUSTOM_HEAD,
)