Telemetry & Traces¶

OpenJarvis มีระบบสังเกตการณ์ (observability) ที่เสริมกันสองระบบ: telemetry สำหรับ metric ต่อการอนุมานแต่ละครั้ง และ traces สำหรับการบันทึกระดับการ โต้ตอบแบบเต็ม รวมกันแล้วให้ข้อมูลเชิงลึกที่ครอบคลุมเกี่ยวกับพฤติกรรมของระบบ และขับเคลื่อนการอัปเดตนโยบาย routing ของระบบการเรียนรู้

Telemetry¶

ระบบ telemetry บันทึก metric สำหรับการเรียก inference ทุกครั้ง -- latency, จำนวน token, ต้นทุน และการใช้พลังงาน ข้อมูลถูกเก็บใน SQLite และ query, export และรวมผลได้

TelemetryRecord¶

การเรียก inference แต่ละครั้งสร้าง TelemetryRecord ที่มีฟิลด์ดังนี้:

ฟิลด์	ชนิด	คำอธิบาย
`timestamp`	`float`	Unix timestamp ของการเรียก
`model_id`	`str`	Identifier ของโมเดล
`engine`	`str`	Engine backend ที่ใช้
`agent`	`str`	เอเจนต์ที่ใช้ (ถ้ามี)
`prompt_tokens`	`int`	Token input ที่ใช้ไป
`completion_tokens`	`int`	Token output ที่สร้างขึ้น
`total_tokens`	`int`	Token ทั้งหมด (prompt + completion)
`latency_seconds`	`float`	เวลา inference แบบ wall-clock
`ttft`	`float`	เวลาจนถึง token แรก
`cost_usd`	`float`	ต้นทุนโดยประมาณเป็น USD
`energy_joules`	`float`	การใช้พลังงานโดยประมาณ
`power_watts`	`float`	การใช้ไฟระหว่าง inference
`metadata`	`dict[str, Any]`	Metadata เพิ่มเติม

TelemetryStore¶

TelemetryStore คือฐานข้อมูล SQLite แบบ append-only ที่เก็บ telemetry record ไว้ถาวร มันผสานกับ event bus เพื่อจับ record โดยอัตโนมัติ

from openjarvis.telemetry.store import TelemetryStore
from openjarvis.core.events import EventBus

bus = EventBus()
store = TelemetryStore(db_path="~/.openjarvis/telemetry.db")
store.subscribe_to_bus(bus)

# Records are now captured automatically when TELEMETRY_RECORD events fire.
# No manual recording needed -- instrumented_generate() handles this.

store.close()

Store จะ subscribe event TELEMETRY_RECORD บน event bus เมื่อใช้ wrapper instrumented_generate() (ซึ่งเกิดขึ้นอัตโนมัติทั้งใน CLI และ SDK) telemetry record จะถูกเผยแพร่และจัดเก็บโดยไม่ต้องแทรกแซงด้วยตนเองเลย

`instrumented_generate()`¶

ฟังก์ชัน wrapper นี้เรียก engine.generate() และเผยแพร่ telemetry event โดยอัตโนมัติ:

เผยแพร่ INFERENCE_START พร้อมข้อมูลโมเดลและ engine
เรียก engine และวัด latency แบบ wall-clock
ดึงการใช้ token จากผลลัพธ์ของ engine
สร้าง TelemetryRecord จากค่าที่วัดได้
เผยแพร่ event INFERENCE_END และ TELEMETRY_RECORD

คำสั่ง CLI และ method ของ SDK ทั้งหมดใช้ wrapper นี้ ดังนั้น telemetry จึง ถูกบันทึกโดยไม่ต้องทำอะไรเพิ่ม

TelemetryAggregator¶

TelemetryAggregator มี method สำหรับ query และรวมผลแบบอ่านอย่างเดียวบน ข้อมูล telemetry ที่เก็บไว้

from openjarvis.telemetry.aggregator import TelemetryAggregator

agg = TelemetryAggregator(db_path="~/.openjarvis/telemetry.db")

# Overall summary
summary = agg.summary()
print(f"Total calls: {summary.total_calls}")
print(f"Total tokens: {summary.total_tokens}")
print(f"Total cost: ${summary.total_cost:.6f}")

# Per-model breakdown
for ms in agg.per_model_stats():
    print(f"  {ms.model_id}: {ms.call_count} calls, {ms.avg_latency:.3f}s avg")

# Per-engine breakdown
for es in agg.per_engine_stats():
    print(f"  {es.engine}: {es.call_count} calls, {es.total_tokens} tokens")

# Top models by usage
top = agg.top_models(n=5)

# Export raw records
records = agg.export_records()

# Time-range filtering (Unix timestamps)
recent = agg.summary(since=1700000000.0)

# Clear all records
count = agg.clear()
print(f"Deleted {count} records")

agg.close()

Method การรวมผล¶

Method	คืนค่า	คำอธิบาย
`summary()`	`AggregatedStats`	จำนวนการเรียก, token, ต้นทุน, latency รวม + แยกตามโมเดลและ engine
`per_model_stats()`	`list[ModelStats]`	จำนวนการเรียก, token, latency, ต้นทุน จัดกลุ่มตามโมเดล
`per_engine_stats()`	`list[EngineStats]`	จำนวนการเรียก, token, latency, ต้นทุน จัดกลุ่มตาม engine
`top_models(n)`	`list[ModelStats]`	โมเดล N อันดับแรกตามจำนวนการเรียก
`export_records()`	`list[dict]`	record ทั้งหมดเป็น dictionary ธรรมดา
`record_count()`	`int`	จำนวน record ที่เก็บไว้ทั้งหมด
`clear()`	`int`	ลบ record ทั้งหมด คืนจำนวนที่ลบ

Method ทั้งหมดรับพารามิเตอร์เสริม since และ until (Unix timestamp) สำหรับ กรองตามช่วงเวลา

Data Class¶

ModelStats:

ฟิลด์	ชนิด	คำอธิบาย
`model_id`	`str`	Identifier ของโมเดล
`call_count`	`int`	จำนวนการเรียก inference ทั้งหมด
`total_tokens`	`int`	Token ที่ประมวลผลทั้งหมด
`prompt_tokens`	`int`	Token input ทั้งหมด
`completion_tokens`	`int`	Token output ทั้งหมด
`total_latency`	`float`	ผลรวม latency ทั้งหมด
`avg_latency`	`float`	Latency เฉลี่ยต่อการเรียก
`total_cost`	`float`	ต้นทุนทั้งหมดเป็น USD

EngineStats:

ฟิลด์	ชนิด	คำอธิบาย
`engine`	`str`	Identifier ของ engine
`call_count`	`int`	จำนวนการเรียก inference ทั้งหมด
`total_tokens`	`int`	Token ที่ประมวลผลทั้งหมด
`total_latency`	`float`	ผลรวม latency ทั้งหมด
`avg_latency`	`float`	Latency เฉลี่ยต่อการเรียก
`total_cost`	`float`	ต้นทุนทั้งหมดเป็น USD

AggregatedStats:

ฟิลด์	ชนิด	คำอธิบาย
`total_calls`	`int`	จำนวนการเรียก inference ทั้งหมด
`total_tokens`	`int`	Token ทั้งหมดข้ามทุกโมเดล
`total_cost`	`float`	ต้นทุนทั้งหมดเป็น USD
`total_latency`	`float`	Latency ทั้งหมดเป็นวินาที
`per_model`	`list[ModelStats]`	แยกตามโมเดล
`per_engine`	`list[EngineStats]`	แยกตาม engine

คำสั่ง CLI¶

# Show aggregated statistics
jarvis telemetry stats
jarvis telemetry stats -n 5          # Top 5 models only

# Export records
jarvis telemetry export              # JSON to stdout
jarvis telemetry export -f csv       # CSV to stdout
jarvis telemetry export -o data.json # JSON to file
jarvis telemetry export -f csv -o metrics.csv

# Clear all records
jarvis telemetry clear               # With confirmation prompt
jarvis telemetry clear --yes         # Without confirmation

Traces¶

ในขณะที่ telemetry จับ metric ต่อการอนุมานแต่ละครั้ง ระบบ trace บันทึก ลำดับการโต้ตอบแบบครบถ้วน -- ห่วงโซ่ขั้นตอนทั้งหมดที่เอเจนต์ทำเพื่อจัดการ คำถาม Trace คือ input หลักของระบบการเรียนรู้

Trace คืออะไร?¶

Trace จับวงจรชีวิตทั้งหมดของการจัดการคำถามผู้ใช้:

ฟิลด์	ชนิด	คำอธิบาย
`trace_id`	`str`	Identifier แบบไม่ซ้ำกัน (สร้างอัตโนมัติ)
`query`	`str`	คำถามต้นฉบับของผู้ใช้
`agent`	`str`	เอเจนต์ที่จัดการคำถาม
`model`	`str`	โมเดลที่ใช้สำหรับ inference
`engine`	`str`	Engine backend ที่ใช้
`steps`	`list[TraceStep]`	รายการขั้นตอนการประมวลผลตามลำดับ
`result`	`str`	เนื้อหาคำตอบสุดท้าย
`outcome`	`str` หรือ `None`	`"success"`, `"failure"` หรือ `None` (ไม่ทราบ)
`feedback`	`float` หรือ `None`	คะแนนคุณภาพจากผู้ใช้ [0, 1]
`started_at`	`float`	Unix timestamp เมื่อเริ่มประมวลผล
`ended_at`	`float`	Unix timestamp เมื่อจบการประมวลผล
`total_tokens`	`int`	Token ทั้งหมดข้ามทุกขั้นตอน
`total_latency_seconds`	`float`	Latency ทั้งหมดข้ามทุกขั้นตอน
`metadata`	`dict[str, Any]`	Metadata เพิ่มเติม

Trace เทียบกับ Telemetry¶

แง่มุม	Telemetry	Traces
ขอบเขต	การเรียก inference ครั้งเดียว	การโต้ตอบแบบเต็ม (หลายขั้นตอน)
ความละเอียด	Metric ต่อการเรียก	ลำดับทีละขั้นตอน
จุดประสงค์	มอนิเตอร์ประสิทธิภาพ, ติดตามต้นทุน	การเรียนรู้, ปรับปรุง routing, ดีบัก
ข้อมูล	Latency, token, ต้นทุน, พลังงาน	route, retrieve, generate, tool_call, respond
การเก็บข้อมูล	ตารางเรียบของ record	ตาราง trace + ตาราง step

TraceStep¶

แต่ละขั้นตอนใน trace บันทึก action หนึ่งอย่างที่เอเจนต์ทำ

ฟิลด์	ชนิด	คำอธิบาย
`step_type`	`StepType`	ประเภทของขั้นตอน (ดูด้านล่าง)
`timestamp`	`float`	เวลาที่ขั้นตอนนี้เกิดขึ้น
`duration_seconds`	`float`	ขั้นตอนนี้ใช้เวลานานเท่าไร
`input`	`dict[str, Any]`	ข้อมูล input ของขั้นตอน
`output`	`dict[str, Any]`	ข้อมูล output จากขั้นตอน
`metadata`	`dict[str, Any]`	Metadata เพิ่มเติม

StepType¶

ประเภท	คำอธิบาย	ตัวอย่าง Input	ตัวอย่าง Output
`route`	การตัดสินใจเลือกโมเดล/เอเจนต์	`{"query_type": "math"}`	`{"model": "qwen3:8b"}`
`retrieve`	การค้นหา memory เพื่อหาบริบท	`{"query": "topic"}`	`{"num_results": 3}`
`generate`	การเรียก LLM inference	`{"model": "qwen3:8b"}`	`{"tokens": 128}`
`tool_call`	การรัน tool	`{"tool": "calculator"}`	`{"success": true}`
`respond`	คำตอบสุดท้ายให้ผู้ใช้	`{}`	`{"content": "...", "turns": 2}`

TraceCollector¶

TraceCollector ห่อ BaseAgent ใดๆ เพื่อบันทึก Trace โดยอัตโนมัติสำหรับ การเรียก run() ทุกครั้ง มัน subscribe event ของ event bus ระหว่างการรัน และแปลงเป็น object TraceStep

from openjarvis.agents.orchestrator import OrchestratorAgent
from openjarvis.traces.collector import TraceCollector
from openjarvis.traces.store import TraceStore
from openjarvis.core.events import EventBus

bus = EventBus()
store = TraceStore(db_path="./traces.db")

agent = OrchestratorAgent(engine, model, tools=tools, bus=bus)
collector = TraceCollector(agent, store=store, bus=bus)

# The trace is recorded automatically
result = collector.run("What is 2+2?")
print(result.content)
# Trace is now saved to the store and published on the bus

Collector ทำงานอย่างไร:

Subscribe event INFERENCE_START, INFERENCE_END, TOOL_CALL_START, TOOL_CALL_END และ MEMORY_RETRIEVE
รัน method run() ของเอเจนต์ที่ห่อไว้
แปลง event ที่จับได้เป็น object TraceStep พร้อมข้อมูลเวลา
เพิ่มขั้นตอน RESPOND สุดท้ายพร้อมผลลัพธ์
สร้าง object Trace แบบสมบูรณ์และบันทึกลง TraceStore
เผยแพร่ event TRACE_COMPLETE บน bus
Unsubscribe จาก event หลังการรันเสร็จสิ้น

TraceStore¶

TraceStore คือฐานข้อมูลที่ใช้ SQLite สำหรับเก็บ trace แบบสมบูรณ์พร้อม ขั้นตอนของมันไว้ถาวร

from openjarvis.traces.store import TraceStore

store = TraceStore(db_path="./traces.db")

# Save a trace
store.save(trace)

# Get a specific trace
trace = store.get("abc123def456")

# List traces with filters
traces = store.list_traces(
    agent="orchestrator",
    model="qwen3:8b",
    outcome="success",
    since=1700000000.0,
    limit=50,
)

# Count total traces
count = store.count()

# Subscribe to event bus for automatic saving
store.subscribe_to_bus(bus)

store.close()

ตัวเลือกการกรอง¶

พารามิเตอร์	ชนิด	คำอธิบาย
`agent`	`str`	กรองตาม agent ID
`model`	`str`	กรองตาม model ID
`outcome`	`str`	กรองตามผลลัพธ์ (`"success"`, `"failure"`)
`since`	`float`	จุดเริ่มต้นของช่วงเวลา (Unix timestamp)
`until`	`float`	จุดสิ้นสุดของช่วงเวลา (Unix timestamp)
`limit`	`int`	จำนวน trace สูงสุดที่จะคืน (ค่าเริ่มต้น: 100)

TraceAnalyzer¶

TraceAnalyzer มีสถิติแบบรวมผลแบบอ่านอย่างเดียวบน trace ที่เก็บไว้ สถิติ เหล่านี้ถูกใช้โดยระบบการเรียนรู้เพื่ออัปเดตนโยบาย routing

from openjarvis.traces.analyzer import TraceAnalyzer

analyzer = TraceAnalyzer(store=trace_store)

# Overall summary
summary = analyzer.summary()
print(f"Total traces: {summary.total_traces}")
print(f"Total steps: {summary.total_steps}")
print(f"Avg steps/trace: {summary.avg_steps_per_trace:.1f}")
print(f"Avg latency: {summary.avg_latency:.3f}s")
print(f"Success rate: {summary.success_rate:.1%}")
print(f"Step distribution: {summary.step_type_distribution}")

# Per-route statistics (model + agent combinations)
for rs in analyzer.per_route_stats():
    print(f"  {rs.model}/{rs.agent}: {rs.count} traces, "
          f"{rs.avg_latency:.3f}s avg, {rs.success_rate:.1%} success")

# Per-tool statistics
for ts in analyzer.per_tool_stats():
    print(f"  {ts.tool_name}: {ts.call_count} calls, "
          f"{ts.avg_latency:.3f}s avg, {ts.success_rate:.1%} success")

# Find traces matching query characteristics
code_traces = analyzer.traces_for_query_type(has_code=True)
short_traces = analyzer.traces_for_query_type(max_length=100)

# Export traces as plain dicts
exported = analyzer.export_traces(limit=500)

Method การวิเคราะห์¶

Method	คืนค่า	คำอธิบาย
`summary()`	`TraceSummary`	สถิติโดยรวม: จำนวน, ค่าเฉลี่ย, การกระจาย
`per_route_stats()`	`list[RouteStats]`	สถิติจัดกลุ่มตามคู่ (model, agent)
`per_tool_stats()`	`list[ToolStats]`	สถิติจัดกลุ่มตามชื่อ tool
`traces_for_query_type()`	`list[Trace]`	กรอง trace ตามลักษณะของคำถาม
`export_traces()`	`list[dict]`	Export trace เป็น dictionary ที่ serialize ได้

Method การวิเคราะห์ทั้งหมดรับพารามิเตอร์เสริม since และ until สำหรับกรอง ตามช่วงเวลา

Data Class¶

TraceSummary:

ฟิลด์	ชนิด	คำอธิบาย
`total_traces`	`int`	จำนวน trace ทั้งหมด
`total_steps`	`int`	จำนวนขั้นตอนทั้งหมดข้ามทุก trace
`avg_steps_per_trace`	`float`	จำนวนขั้นตอนเฉลี่ยต่อ trace
`avg_latency`	`float`	Latency รวมเฉลี่ยต่อ trace
`avg_tokens`	`float`	Token เฉลี่ยต่อ trace
`success_rate`	`float`	สัดส่วนของ trace ที่ประเมินแล้วสำเร็จ
`step_type_distribution`	`dict[str, int]`	จำนวนของแต่ละประเภทขั้นตอน

RouteStats:

ฟิลด์	ชนิด	คำอธิบาย
`model`	`str`	Identifier ของโมเดล
`agent`	`str`	Identifier ของเอเจนต์
`count`	`int`	จำนวน trace สำหรับ route นี้
`avg_latency`	`float`	Latency เฉลี่ยสำหรับ route นี้
`avg_tokens`	`float`	Token เฉลี่ยสำหรับ route นี้
`success_rate`	`float`	อัตราความสำเร็จสำหรับ route นี้
`avg_feedback`	`float` หรือ `None`	Feedback เฉลี่ยจากผู้ใช้ (ถ้ามี)

ToolStats:

ฟิลด์	ชนิด	คำอธิบาย
`tool_name`	`str`	Identifier ของ tool
`call_count`	`int`	จำนวนครั้งที่ tool ถูกเรียก
`avg_latency`	`float`	Latency เฉลี่ยของการรัน
`success_rate`	`float`	สัดส่วนของการรันที่สำเร็จ

การไหลของข้อมูล¶

แผนภาพต่อไปนี้แสดงการไหลของข้อมูล telemetry และ trace ผ่านระบบ:

User Query
    |
    v
Agent.run()  -->  EventBus  -->  TraceCollector (captures steps)
    |                   |
    v                   v
Engine.generate()  TelemetryStore (captures per-call metrics)
    |
    v
instrumented_generate()
    |
    +---> INFERENCE_START event
    +---> INFERENCE_END event
    +---> TELEMETRY_RECORD event
    |
    v
TraceCollector
    |
    +---> Builds Trace with TraceSteps
    +---> Saves to TraceStore
    +---> Publishes TRACE_COMPLETE event
    |
    v
TraceAnalyzer / TelemetryAggregator  -->  Learning System

ทั้งสองระบบทำงานโดยไม่ต้องแทรกแซง -- ไม่ต้องเขียน instrumentation ด้วยตนเอง เมื่อใช้ CLI หรือ SDK เพราะทั้งสองตั้งค่า event bus และ telemetry store ให้ โดยอัตโนมัติ

Telemetry & Traces¶

Telemetry¶

TelemetryRecord¶

TelemetryStore¶

instrumented_generate()¶

TelemetryAggregator¶

Method การรวมผล¶

Data Class¶

คำสั่ง CLI¶

Traces¶

Trace คืออะไร?¶

Trace เทียบกับ Telemetry¶

TraceStep¶

StepType¶

TraceCollector¶

TraceStore¶

ตัวเลือกการกรอง¶

TraceAnalyzer¶

Method การวิเคราะห์¶

Data Class¶

การไหลของข้อมูล¶

`instrumented_generate()`¶