File size: 3,462 Bytes
fc8507a | 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 | #!/usr/bin/env python3
"""Export a Dukaan Saathi agent trace to JSONL for the Hub (Sharing-is-Caring badge).
Runs a few real shopkeeper turns through the deepagents loop against the configured
LLM (point DUKAAN_LLM_BASE_URL at Modal) and writes every message — the human
prompt, the model's tool calls, each tool result, and the final Hindi reply — to a
JSONL trace you can upload to a Hugging Face dataset and link from the README.
Usage
-----
uv run python scripts/export_trace.py # uses config endpoints
uv run python scripts/export_trace.py https://<ws>--dukaan-llm-serve.modal.run
# writes: logs/agent_trace.jsonl (one JSON object per message)
"""
from __future__ import annotations
import json
import os
import sys
import uuid
from pathlib import Path
# A short, demo-worthy script that exercises read + diagnostic + the visible
# multi-tool reasoning that makes the "Best Agent" case.
TURNS = [
"aaj kitni bikri hui?",
"sabse zyada udhaar kiska hai? top 3 batao",
"Munna Yadav ka kitna udhaar baaki hai?",
"Parle-G kyun nahi bik raha?",
]
def _route_to(base: str) -> None:
base = base.rstrip("/")
if base.endswith("/v1"):
base = base[:-3]
os.environ["DUKAAN_LLM_BASE_URL"] = base + "/v1"
def _msg_to_dict(m) -> dict:
"""Serialize a LangChain message (Human / AI / Tool) to a plain JSON dict."""
role = getattr(m, "type", None) or getattr(m, "role", None) or m.__class__.__name__
content = getattr(m, "content", "")
if not isinstance(content, str):
try:
from dukaan.agent import _content_to_text
content = _content_to_text(content)
except Exception:
content = str(content)
out: dict = {"role": role, "content": content}
tcs = getattr(m, "tool_calls", None) or []
if tcs:
out["tool_calls"] = [
{"name": (tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)),
"args": (tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", None))}
for tc in tcs
]
name = getattr(m, "name", None)
if name:
out["name"] = name # tool name on a ToolMessage
return out
def main() -> int:
if len(sys.argv) > 1:
_route_to(sys.argv[1])
from dukaan import agent, config
out_path = Path("logs/agent_trace.jsonl")
out_path.parent.mkdir(parents=True, exist_ok=True)
print(f"Exporting agent trace via {config.LLM_BASE_URL}")
thread = uuid.uuid4().hex
records: list[dict] = []
seen = 0
for turn in TURNS:
print(f" turn: {turn!r}")
res = agent.run_agent(turn, thread_id=thread)
if res.get("error"):
print(f" !! error: {res['error']}")
msgs = res.get("messages", [])
for m in msgs[seen:]: # only this turn's new messages (history accumulates)
records.append({"turn": turn, **_msg_to_dict(m)})
seen = len(msgs)
print(f" tools: {res.get('tool_calls')} intent: {res.get('intent')}")
with open(out_path, "w", encoding="utf-8") as fh:
for r in records:
fh.write(json.dumps(r, ensure_ascii=False) + "\n")
print(f"\nWrote {len(records)} messages -> {out_path}")
print("Upload it to a Hub dataset/repo and link it from the README for the Sharing-is-Caring badge.")
return 0 if records else 1
if __name__ == "__main__":
raise SystemExit(main())
|