diff --git a/CHANGELOG.md b/CHANGELOG.md index 0da59d4..225daff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ adheres to [Semantic Versioning](https://semver.org/). ### Added +- **Telemetry** — `tracing` spans: `agent_loop` (model), `llm_stream` per + turn (tokens in/out/cached + `cost_usd` when pricing is configured), and + `tool` per execution (name, `is_error`). Bridge to OpenTelemetry + app-side with `tracing-opentelemetry`; zero overhead with no subscriber. + New `telemetry` example and docs page. + - **Cross-provider thinking (7/7)** — `thinking_level` is now honored by every protocol: Gemini and Vertex send `thinkingConfig` (with thought summaries streamed back as `Content::Thinking`), Bedrock sends diff --git a/CLAUDE.md b/CLAUDE.md index da439f8..258b56d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -123,6 +123,10 @@ Other constructors: All unit tests use `MockProvider` (`provider/mock.rs`) to simulate LLM responses without network. Test files are in `tests/` — `agent_test.rs`, `agent_loop_test.rs`, `tools_test.rs`. Follow the existing pattern of constructing a `MockProvider` with predetermined responses. +### Telemetry + +The loop emits `tracing` spans: `agent_loop` → `llm_stream` per turn (records tokens_in/out/cached + cost_usd from `CostConfig` when configured) → `tool` per execution (records is_error). Futures are instrumented with `.instrument(span)` (never hold an entered guard across `.await`). OTel is app-side via `tracing-opentelemetry` — the library has no OTel dependency by design. + ## Key Design Conventions - Context overflow detection is centralized in `OVERFLOW_PHRASES` (`provider/traits.rs`) covering 15+ provider-specific error strings; both HTTP errors and SSE-embedded errors are classified diff --git a/Cargo.toml b/Cargo.toml index a24bdda..945a6f8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,5 +37,6 @@ openapi = ["dep:openapiv3", "dep:serde_yaml_ng", "reqwest/query"] [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } +tracing-subscriber = { version = "0.3", features = ["registry", "fmt"] } tempfile = "3" wiremock = "0.6" diff --git a/README.md b/README.md index 938c2ef..72bdcc4 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,7 @@ Everything is observable via events. Supports 7 API protocols covering 20+ LLM p - State persistence — `save_messages()` / `restore_messages()` for pause/resume workflows - Session trees — branching history with fork, checkpoints, and JSONL persistence (`Session`); edit an earlier turn and re-run without losing the original branch - Lifecycle callbacks — `before_turn`, `after_turn`, `on_error` for observability and control +- Telemetry — `tracing` spans for the loop, each LLM stream (tokens + cost fields), and each tool execution; bridge to OpenTelemetry via `tracing-opentelemetry`, zero overhead when no subscriber is installed - Full serde support — all core types implement `Serialize`/`Deserialize`/`PartialEq` - [AgentSkills](https://agentskills.io)-compatible skills — load skill directories, inject into system prompt, agent activates on demand diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 6d86fb8..384a479 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -21,6 +21,7 @@ - [State Persistence](concepts/persistence.md) - [Session Trees](concepts/session-trees.md) - [Lifecycle Callbacks](concepts/callbacks.md) +- [Telemetry](concepts/telemetry.md) # Guides diff --git a/docs/concepts/telemetry.md b/docs/concepts/telemetry.md new file mode 100644 index 0000000..97a306f --- /dev/null +++ b/docs/concepts/telemetry.md @@ -0,0 +1,62 @@ +# Telemetry + +yoagent instruments the loop with [`tracing`](https://docs.rs/tracing) **spans** +— structured, timed, nested units your observability stack can consume. With +no subscriber installed they compile to no-ops; there is zero overhead unless +you opt in. + +## The span tree + +``` +agent_loop (model) +└─ llm_stream (turn, model, tokens_in, tokens_out, tokens_cached, cost_usd) +└─ tool (tool, tool_call_id, is_error) +``` + +- `llm_stream` — one per turn, wrapping the provider call. Token counts are + recorded from real usage; `cost_usd` is recorded when the `ModelConfig` has + pricing configured (`CostConfig`). +- `tool` — one per tool execution, with the tool name and error status; + duration comes free with the span. + +## Local: print spans + +```rust +tracing_subscriber::fmt() + .with_span_events(tracing_subscriber::fmt::format::FmtSpan::CLOSE) + .init(); +``` + +Run `cargo run --example telemetry` to see it. + +## Production: OpenTelemetry + +The OTel bridge is **application-side** — yoagent needs no OTel dependency +(that's the point of `tracing`). Install the +[`tracing-opentelemetry`](https://docs.rs/tracing-opentelemetry) layer and the +same spans flow to any OTLP backend — Datadog, Grafana Tempo, Honeycomb, +Jaeger: + +```rust +use opentelemetry_otlp::WithExportConfig; +use tracing_subscriber::layer::SubscriberExt; + +let tracer = opentelemetry_otlp::new_pipeline() + .tracing() + .with_exporter(opentelemetry_otlp::new_exporter().tonic()) + .install_batch(opentelemetry_sdk::runtime::Tokio)?; + +let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(tracer)); +tracing::subscriber::set_global_default(subscriber)?; +``` + +Because these are ordinary `tracing` spans, an agent call nests inside your +app's existing request traces (e.g. an axum handler span) automatically. + +## What it buys you + +- **Cost attribution** — dollars per turn/model in your dashboards, from the + same `CostConfig` data as [`session_cost_usd`](../reference/api.md). +- **Latency diagnosis** — "40s request: 8s provider, 30s in one bash tool." +- **Audit** — which tools ran, with what outcome, per session. diff --git a/examples/telemetry.rs b/examples/telemetry.rs new file mode 100644 index 0000000..5e98f0a --- /dev/null +++ b/examples/telemetry.rs @@ -0,0 +1,31 @@ +//! Telemetry example: yoagent emits `tracing` spans for the loop, each LLM +//! stream (with token/cost fields), and each tool execution. +//! +//! Run with: cargo run --example telemetry +//! +//! Here spans are printed with the fmt subscriber; in production, install a +//! `tracing-opentelemetry` layer instead and the same spans flow to any OTLP +//! backend (Datadog, Grafana Tempo, Honeycomb, Jaeger, ...). + +use yoagent::provider::{MockProvider, ModelConfig}; +use yoagent::*; + +#[tokio::main] +async fn main() { + // Print spans with timing on close. Swap this for an OTel layer in prod. + tracing_subscriber::fmt() + .with_span_events(tracing_subscriber::fmt::format::FmtSpan::CLOSE) + .with_target(false) + .init(); + + let mut agent = Agent::from_provider( + MockProvider::text("Here is the answer."), + ModelConfig::mock(), + ); + + let mut rx = agent.prompt("What is the answer?").await; + while rx.recv().await.is_some() {} + agent.finish().await; + + println!("done — spans above show agent_loop / llm_stream timings"); +} diff --git a/src/agent_loop.rs b/src/agent_loop.rs index 3cf3775..ce348dd 100644 --- a/src/agent_loop.rs +++ b/src/agent_loop.rs @@ -208,7 +208,12 @@ pub async fn agent_loop( .ok(); } - run_loop(context, &mut new_messages, config, &tx, &cancel).await; + { + use tracing::Instrument; + run_loop(context, &mut new_messages, config, &tx, &cancel) + .instrument(tracing::info_span!("agent_loop", model = %config.model)) + .await; + } tx.send(AgentEvent::AgentEnd { messages: new_messages.clone(), @@ -241,7 +246,12 @@ pub async fn agent_loop_continue( tx.send(AgentEvent::AgentStart).ok(); tx.send(AgentEvent::TurnStart).ok(); - run_loop(context, &mut new_messages, config, &tx, &cancel).await; + { + use tracing::Instrument; + run_loop(context, &mut new_messages, config, &tx, &cancel) + .instrument(tracing::info_span!("agent_loop", model = %config.model)) + .await; + } tx.send(AgentEvent::AgentEnd { messages: new_messages.clone(), @@ -405,8 +415,33 @@ async fn run_loop( } } - // Stream assistant response - let message = stream_assistant_response(context, config, tx, cancel).await; + // Stream assistant response, under an llm_stream span that + // records tokens and (when rates are configured) dollar cost. + let llm_span = tracing::info_span!( + "llm_stream", + turn = turn_number, + model = %config.model, + tokens_in = tracing::field::Empty, + tokens_out = tracing::field::Empty, + tokens_cached = tracing::field::Empty, + cost_usd = tracing::field::Empty, + ); + let message = { + use tracing::Instrument; + stream_assistant_response(context, config, tx, cancel) + .instrument(llm_span.clone()) + .await + }; + if let Message::Assistant { usage, .. } = &message { + llm_span.record("tokens_in", usage.input); + llm_span.record("tokens_out", usage.output); + llm_span.record("tokens_cached", usage.cache_read); + if let Some(mc) = &config.model_config { + if mc.cost.is_configured() { + llm_span.record("cost_usd", mc.cost.cost_usd(usage)); + } + } + } // Tool-forcing providers (Anthropic) deliver structured output as // a forced tool call — unwrap it into plain text BEFORE tool-call // extraction, so the loop never tries to execute the synthetic tool. @@ -987,19 +1022,32 @@ async fn execute_single_tool( on_progress, }; + let tool_span = tracing::info_span!( + "tool", + tool = %name, + tool_call_id = %id, + is_error = tracing::field::Empty, + ); + use tracing::Instrument; let (result, is_error) = match tool { - Some(tool) => match tool.execute(args.clone(), ctx).await { - Ok(r) => (r, false), - Err(e) => ( - ToolResult { - content: vec![Content::Text { - text: e.to_string(), - }], - details: serde_json::Value::Null, - }, - true, - ), - }, + Some(tool) => { + let execution = tool + .execute(args.clone(), ctx) + .instrument(tool_span.clone()) + .await; + match execution { + Ok(r) => (r, false), + Err(e) => ( + ToolResult { + content: vec![Content::Text { + text: e.to_string(), + }], + details: serde_json::Value::Null, + }, + true, + ), + } + } None => ( ToolResult { content: vec![Content::Text { @@ -1011,6 +1059,8 @@ async fn execute_single_tool( ), }; + tool_span.record("is_error", is_error); + tx.send(AgentEvent::ToolExecutionEnd { tool_call_id: id.to_string(), tool_name: name.to_string(), diff --git a/src/lib.rs b/src/lib.rs index 92e7dd2..a13b111 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,6 +55,8 @@ //! compaction so long sessions keep running. //! - **Skills** ([`skills`]) — load `SKILL.md` files per the //! [AgentSkills](https://agentskills.io) standard. +//! - **Telemetry** — `tracing` spans per loop/LLM-stream/tool with token and +//! cost fields; bridge to OpenTelemetry app-side, zero overhead otherwise. //! //! The [book](https://yologdev.github.io/yoagent/) covers concepts and //! provider-specific guides. diff --git a/tests/telemetry_test.rs b/tests/telemetry_test.rs new file mode 100644 index 0000000..3d26aa3 --- /dev/null +++ b/tests/telemetry_test.rs @@ -0,0 +1,144 @@ +//! Tests that the loop emits the documented tracing spans with their fields. +//! +//! Attaches a capturing subscriber to the loop future via `with_subscriber` +//! and drives `agent_loop` directly in the current task — spans created in +//! separately-spawned tasks would not reach this scoped subscriber. + +use std::sync::{Arc, Mutex}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use tracing::instrument::WithSubscriber; +use tracing_subscriber::layer::SubscriberExt; +use yoagent::provider::mock::*; +use yoagent::provider::MockProvider; +use yoagent::*; + +/// Layer that records every new span's name. +struct SpanCollector(Arc>>); + +impl tracing_subscriber::Layer for SpanCollector +where + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, +{ + fn on_new_span( + &self, + attrs: &tracing::span::Attributes<'_>, + _id: &tracing::span::Id, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + self.0 + .lock() + .unwrap() + .push(attrs.metadata().name().to_string()); + } +} + +struct EchoTool; + +#[async_trait::async_trait] +impl AgentTool for EchoTool { + fn name(&self) -> &str { + "echo" + } + fn label(&self) -> &str { + "Echo" + } + fn description(&self) -> &str { + "echoes" + } + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({"type": "object"}) + } + async fn execute( + &self, + _params: serde_json::Value, + _ctx: ToolContext, + ) -> Result { + Ok(ToolResult { + content: vec![Content::Text { text: "ok".into() }], + details: serde_json::Value::Null, + }) + } +} + +fn loop_config(provider: MockProvider) -> yoagent::agent_loop::AgentLoopConfig { + yoagent::agent_loop::AgentLoopConfig { + provider: std::sync::Arc::new(provider), + model: "mock".into(), + api_key: "test".into(), + thinking_level: ThinkingLevel::Off, + max_tokens: None, + temperature: None, + model_config: None, + convert_to_llm: None, + transform_context: None, + get_steering_messages: None, + get_follow_up_messages: None, + context_config: None, + compaction_strategy: None, + execution_limits: None, + cache_config: CacheConfig::default(), + tool_execution: ToolExecutionStrategy::default(), + tool_middleware: vec![], + output_schema: None, + retry_config: yoagent::RetryConfig::none(), + before_turn: None, + after_turn: None, + on_error: None, + input_filters: vec![], + turn_delay: None, + } +} + +#[tokio::test] +async fn loop_emits_agent_llm_and_tool_spans() { + let spans = Arc::new(Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::registry().with(SpanCollector(spans.clone())); + + let provider = MockProvider::new(vec![ + MockResponse::ToolCalls(vec![MockToolCall { + provider_metadata: None, + name: "echo".into(), + arguments: serde_json::json!({}), + }]), + MockResponse::Text("done".into()), + ]); + let config = loop_config(provider); + + let mut context = AgentContext { + system_prompt: "test".into(), + messages: Vec::new(), + tools: vec![Box::new(EchoTool)], + }; + let (tx, _rx) = mpsc::unbounded_channel(); + + // with_subscriber attaches the subscriber across every poll of the + // future (a plain `with_default(sub, || fut).await` would only cover + // creating the future, not running it). + agent_loop( + vec![AgentMessage::Llm(Message::user("go"))], + &mut context, + &config, + tx, + CancellationToken::new(), + ) + .with_subscriber(subscriber) + .await; + + let names = spans.lock().unwrap().clone(); + assert!( + names.contains(&"agent_loop".to_string()), + "expected agent_loop span, got: {names:?}" + ); + // Two turns → two llm_stream spans; one tool execution → one tool span. + assert_eq!( + names.iter().filter(|n| *n == "llm_stream").count(), + 2, + "got: {names:?}" + ); + assert_eq!( + names.iter().filter(|n| *n == "tool").count(), + 1, + "got: {names:?}" + ); +}