diff --git a/docs/opentelemetry-pipeline.md b/docs/opentelemetry-pipeline.md index 0c844547..0d0208ce 100644 --- a/docs/opentelemetry-pipeline.md +++ b/docs/opentelemetry-pipeline.md @@ -15,22 +15,39 @@ receivers: http: endpoint: 0.0.0.0:4318 exporters: - otlphttp/agentevals: + otlp_http/agentevals: endpoint: http://agentevals:4318 service: pipelines: traces: receivers: [otlp] - exporters: [otlphttp/agentevals] + exporters: [otlp_http/agentevals] logs: receivers: [otlp] - exporters: [otlphttp/agentevals] + exporters: [otlp_http/agentevals] ``` -The `otlphttp` exporter uses gzip by default, which agentevals accepts. Send logs as well as traces: several instrumentations put message content in log events. +The `otlp_http` exporter uses gzip by default, which agentevals accepts. Send logs as well as traces: several instrumentations put message content in log events. + +`otlp_http` and `otlp_grpc` are the current exporter names. The older `otlphttp` and `otlp` still work, but the Collector logs them as deprecated. Send to one agentevals instance. Live sessions are kept in the memory of the process that receives them, so behind a load balancer with several replicas a session is split between them. The Helm chart keeps `replicaCount: 1` for this reason. +## Drop high volume spans that are not GenAI + +agentevals keeps the first 10,000 spans of a trace. A producer that writes a span for every streamed chunk or storage call can push a long turn past that, and the turn loses its agent span. Drop those spans in the Collector with a `filter` processor on the agentevals branch. For kagent, whose TaskStore gRPC calls make up most of a turn's spans: + +```yaml +processors: + filter/kagent_taskstore: + error_mode: ignore + traces: + span: + - IsMatch(span.attributes["rpc.method"], "^kagent\\.api\\.v1alpha1\\.TaskStoreService/") +``` + +Add it to the `processors` of the traces pipeline that exports to agentevals. Turns, tool calls and tokens come out the same; only the RPC spans go. + ## Production: backend and agentevals side by side One Collector, two branches: @@ -76,7 +93,7 @@ processors: log_record: - resource.attributes["service.name"] == "agentevals" connectors: - signaltometrics: + signal_to_metrics: logs: - name: agentevals.evaluation.score description: Scores from gen_ai.evaluation.result events @@ -91,11 +108,11 @@ connectors: buckets: [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] value: Double(log.attributes["gen_ai.evaluation.score.value"]) exporters: - otlp/backend: + otlp_grpc/backend: endpoint: ${env:BACKEND_OTLP_ENDPOINT} sending_queue: enabled: true - otlphttp/agentevals: + otlp_http/agentevals: endpoint: http://agentevals:4318 sending_queue: enabled: true @@ -104,22 +121,22 @@ service: traces/backend: receivers: [otlp] processors: [memory_limiter, filter/drop_genai_content_events, transform/drop_genai_content] - exporters: [otlp/backend] + exporters: [otlp_grpc/backend] traces/agentevals: receivers: [otlp] processors: [memory_limiter] - exporters: [otlphttp/agentevals] + exporters: [otlp_http/agentevals] logs/backend: receivers: [otlp] processors: [memory_limiter, filter/drop_genai_content_events, transform/drop_genai_content] - exporters: [otlp/backend, signaltometrics] + exporters: [otlp_grpc/backend, signal_to_metrics] logs/agentevals: receivers: [otlp] processors: [memory_limiter, filter/not_from_agentevals] - exporters: [otlphttp/agentevals] + exporters: [otlp_http/agentevals] metrics: - receivers: [signaltometrics] - exporters: [otlp/backend] + receivers: [signal_to_metrics] + exporters: [otlp_grpc/backend] ``` Notes: diff --git a/examples/kubernetes/README.md b/examples/kubernetes/README.md index 9dab019b..275b3be8 100644 --- a/examples/kubernetes/README.md +++ b/examples/kubernetes/README.md @@ -10,6 +10,13 @@ kagent agents --> OTel Collector --> agentevals (UI on :8001) Tested with kagent `46fdd3d7` (`v1.0.0-alpha9` plus a UI only commit, `v1alpha3` agents on Substrate), Claude Code 2.1.285, Collector contrib 0.162.0, and `claude-haiku-4-5` for both agents. +What you get depends on the kagent version: + +| kagent | Go ADK | Claude Code harness | Codex harness | +|---|---|---|---| +| `v1.0.0-alpha9` (tested) | Model and tool calls with arguments, results and tokens | Native Claude Code spans only. The `transform/claude_code` recipe turns them into tool names and model calls with tokens | Not tested | +| With runtime GenAI spans ([kagent#3121](https://github.com/kagent-dev/kagent/pull/3121), not released yet) | Same | The runtime writes `chat` and `execute_tool` spans with tool names, call ids and tokens, but no tool arguments. The recipe steps aside by itself | Runtime spans, not tested yet | + ## Set it up You need a cluster with kagent 1.0 already running. @@ -61,6 +68,42 @@ Open the kagent UI at http://localhost:8080. For `adk-go-test`, and then `claude Use a new chat after the agents got their new revision. Older chats stay on the old one and send nothing. +### Or from the command line + +To repeat a run without clicking through the UI, talk to the kagent controller: + +```bash +kubectl port-forward -n kagent svc/kagent-controller 8083:8083 +``` + +With the kagent CLI, which uses `http://localhost:8083` by default: + +```bash +for agent in adk-go-test claude-test; do + session=$(kagent agent session create --agent "$agent" -o json | jq -r .session.id) + kagent agent invoke --session "$session" -t "Use your tools to list the namespaces in this cluster, then tell me how many pods are running in the kagent namespace. Answer in two short sentences." + kagent agent invoke --session "$session" -t "Now use your tools to list the services in the agentevals namespace. One sentence." +done +``` + +Without the CLI, send the same prompts over A2A. The first message starts a session; pass its `contextId` to continue it: + +```bash +send() { + curl -sS --max-time 300 "http://localhost:8083/agents/kagent/$1" \ + -H 'Content-Type: application/json' -H 'X-User-Id: admin@kagent.dev' \ + -d "$(jq -n --arg t "$2" --arg c "${3:-}" --arg id "$(uuidgen)" \ + '{jsonrpc: "2.0", id: "1", method: "SendMessage", + params: {message: ({messageId: $id, role: "ROLE_USER", parts: [{text: $t}]} + + (if $c == "" then {} else {contextId: $c} end))}}')" +} + +ctx=$(send adk-go-test "Use your tools to list the namespaces in this cluster, then tell me how many pods are running in the kagent namespace. Answer in two short sentences." | jq -r .result.task.contextId) +send adk-go-test "Now use your tools to list the services in the agentevals namespace. One sentence." "$ctx" +``` + +Each session (one `contextId`) shows up in agentevals as one session. + Now open agentevals at http://localhost:8001 and click **Local Development** in the sidebar. Each chat shows up as one session with two turns, and is marked complete a few seconds after the last answer. If the page stops loading after agentevals restarts, restart its port forward. ## Score them @@ -124,6 +167,10 @@ The trace and span id point at the turn's `invoke_agent` span, so your backend s Events carry no prompts or answers. Set `AGENTEVALS_EVALUATION_EVENTS_EXPLANATION=true` to add the evaluator's explanation, which can contain content. To forward events to a real backend and turn scores into a histogram metric, see [OpenTelemetry pipelines](../../docs/opentelemetry-pipeline.md). Outside Kubernetes, `agentevals run ... --emit-otel` and `agentevals serve --emit-otel` do the same. +## Long turns + +kagent writes a TaskStore gRPC span for every streamed chunk, so a long turn can have thousands of them. agentevals keeps the first 10,000 spans of a trace, and a turn that passes that loses its `invoke_agent` span. The `filter/kagent_taskstore` processor in `otel-collector.yaml` drops those RPC spans before they reach agentevals. Turns, tool calls and tokens come out the same. + ## Claude Code workaround Claude Code exports its own `claude_code.*` spans instead of GenAI ones. kagent wraps each turn in a GenAI `invoke_agent` span, so out of the box you only get the message and the answer. diff --git a/examples/kubernetes/otel-collector.yaml b/examples/kubernetes/otel-collector.yaml index f196d4ce..27f2a388 100644 --- a/examples/kubernetes/otel-collector.yaml +++ b/examples/kubernetes/otel-collector.yaml @@ -3,6 +3,10 @@ # traces and logs from kagent go to agentevals; evaluation results that agentevals sends back # are printed by the debug exporter (kubectl logs deploy/otel-collector). # +# filter/kagent_taskstore drops kagent's TaskStore gRPC spans: one per streamed chunk, up to +# 99% of a turn's spans. Without it a long turn passes the 10,000 spans per trace that +# agentevals keeps, and the turn loses its invoke_agent span. +# # transform/claude_code is a workaround for Claude Code native telemetry, checked with # Claude Code 2.1.285 (kagent Claude harness). It skips processes whose resource has # kagent.genai.producer, where the runtime writes its own chat and execute_tool spans. @@ -27,6 +31,12 @@ data: limit_percentage: 80 spike_limit_percentage: 20 + filter/kagent_taskstore: + error_mode: ignore + traces: + span: + - IsMatch(span.attributes["rpc.method"], "^kagent\\.api\\.v1alpha1\\.TaskStoreService/") + transform/claude_code: error_mode: ignore trace_statements: @@ -58,7 +68,7 @@ data: - resource.attributes["service.name"] != "agentevals" exporters: - otlphttp/agentevals: + otlp_http/agentevals: endpoint: http://agentevals.agentevals.svc.cluster.local:4318 debug: verbosity: detailed @@ -67,12 +77,12 @@ data: pipelines: traces: receivers: [otlp] - processors: [memory_limiter, transform/claude_code] - exporters: [otlphttp/agentevals] + processors: [memory_limiter, filter/kagent_taskstore, transform/claude_code] + exporters: [otlp_http/agentevals] logs/agentevals: receivers: [otlp] processors: [memory_limiter, filter/not_from_agentevals] - exporters: [otlphttp/agentevals] + exporters: [otlp_http/agentevals] logs/results: receivers: [otlp] processors: [memory_limiter, filter/only_from_agentevals] @@ -94,7 +104,7 @@ spec: app: otel-collector annotations: # bump to restart the Collector after editing the ConfigMap - config-revision: "1" + config-revision: "2" spec: containers: - name: otelcol diff --git a/examples/zero-code-examples/openai-agents/requirements.txt b/examples/zero-code-examples/openai-agents/requirements.txt index 7b968e9a..aedf192e 100644 --- a/examples/zero-code-examples/openai-agents/requirements.txt +++ b/examples/zero-code-examples/openai-agents/requirements.txt @@ -1,5 +1,5 @@ openai>=2.30.0 -openai-agents>=0.13.0 +openai-agents>=0.19.0 opentelemetry-instrumentation-openai-agents-v2>=0.1.0 opentelemetry-sdk>=1.36.0 diff --git a/pyproject.toml b/pyproject.toml index 29576e95..f10682b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -94,7 +94,7 @@ e2e = [ "langchain>=1.2.0", "langchain-openai>=1.1.10", "openai>=2.30.0", - "openai-agents>=0.13.0", + "openai-agents>=0.19.0", "opentelemetry-sdk>=1.36.0", "opentelemetry-exporter-otlp-proto-http>=1.36.0", "opentelemetry-instrumentation-openai-v2", diff --git a/src/agentevals/api/app.py b/src/agentevals/api/app.py index bf72c030..1dceaa2a 100644 --- a/src/agentevals/api/app.py +++ b/src/agentevals/api/app.py @@ -24,6 +24,7 @@ from ..storage.postgres.migrator import Migrator, discover_migrations from ..utils.log_buffer import log_buffer from .debug_routes import debug_router +from .models import OtlpReceiverPorts from .routes import router from .runs_routes import runs_router @@ -145,8 +146,10 @@ def create_app( trace_manager: LiveManager | None = None, enable_streaming: bool = False, static_dir: Path | None = None, + otlp_ports: OtlpReceiverPorts | None = None, ) -> FastAPI: - """Create the main agentevals API app.""" + """Create the main agentevals API app. ``otlp_ports`` are the receivers started next to it, + reported by ``/api/config`` so the UI can say where to send telemetry.""" from ..evaluator.resolver import require_index_membership require_index_membership() @@ -185,6 +188,7 @@ def create_app( if trace_manager is not None: app.state.trace_manager = trace_manager + app.state.otlp_ports = otlp_ports if enable_streaming: if trace_manager is None: diff --git a/src/agentevals/api/models.py b/src/agentevals/api/models.py index f00336b9..c81df17d 100644 --- a/src/agentevals/api/models.py +++ b/src/agentevals/api/models.py @@ -44,8 +44,14 @@ class ApiKeyStatus(CamelModel): openai: bool +class OtlpReceiverPorts(CamelModel): + http_port: int + grpc_port: int + + class ConfigData(CamelModel): api_keys: ApiKeyStatus + otlp: OtlpReceiverPorts | None = None class MetricInfo(CamelModel): diff --git a/src/agentevals/api/routes.py b/src/agentevals/api/routes.py index 80156577..17e670f2 100644 --- a/src/agentevals/api/routes.py +++ b/src/agentevals/api/routes.py @@ -221,14 +221,15 @@ async def health_check(): @router.get("/config", response_model=StandardResponse[ConfigData]) -async def get_config(): +async def get_config(request: Request): return StandardResponse( data=ConfigData( api_keys=ApiKeyStatus( google=bool(os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")), anthropic=bool(os.environ.get("ANTHROPIC_API_KEY")), openai=bool(os.environ.get("OPENAI_API_KEY")), - ) + ), + otlp=getattr(request.app.state, "otlp_ports", None), ) ) diff --git a/src/agentevals/cli.py b/src/agentevals/cli.py index 556c2771..448f14a3 100644 --- a/src/agentevals/cli.py +++ b/src/agentevals/cli.py @@ -547,6 +547,7 @@ async def _run_servers( import uvicorn from .api.app import create_app + from .api.models import OtlpReceiverPorts from .api.otlp_app import create_otlp_app from .streaming.manager import LiveManager @@ -558,7 +559,11 @@ async def _run_servers( from . import evaluation_events as emit mgr = LiveManager(emitter_instance_id=emit.instance_id() if emit.enabled() else None) - main_app = create_app(trace_manager=mgr, enable_streaming=True) + main_app = create_app( + trace_manager=mgr, + enable_streaming=True, + otlp_ports=OtlpReceiverPorts(http_port=otlp_http_port, grpc_port=otlp_grpc_port), + ) otlp_app = create_otlp_app(trace_manager=mgr) main_server = uvicorn.Server(uvicorn.Config(main_app, port=port, **shared_kwargs)) diff --git a/src/agentevals/genai/extract.py b/src/agentevals/genai/extract.py index 5fb92c86..40fa9767 100644 --- a/src/agentevals/genai/extract.py +++ b/src/agentevals/genai/extract.py @@ -190,7 +190,8 @@ def analyze(trace: Trace) -> TraceAnalysis: if span_id in agentic_anchors: view = info[span_id].view has_messages = view.messages(sc.INPUT_MESSAGES) or view.messages(sc.OUTPUT_MESSAGES) - if INFERENCE in kinds or TOOL in kinds or has_messages: + failed = _status(view.span) == "error" or attr_str(view.attrs, sc.ERROR_TYPE) is not None + if INFERENCE in kinds or TOOL in kinds or has_messages or failed: anchors.append(span_id) else: warnings.append(f"agent span {span_id} has no inference, tools or messages; not a turn") diff --git a/src/agentevals/streaming/manager.py b/src/agentevals/streaming/manager.py index 5ea4ba81..dfd801e2 100644 --- a/src/agentevals/streaming/manager.py +++ b/src/agentevals/streaming/manager.py @@ -114,6 +114,10 @@ def invocation_dict(turn: Turn, spans: dict[tuple[str, str], Span]) -> dict[str, {"name": t.name, "response": t.result, "id": t.call_id} for t in turn.tool_calls if t.result is not None ], "modelInfo": model_info(turn, spans), + "status": turn.status, + "errorType": turn.error_type, + "contentCaptured": turn.content_captured, + "warnings": turn.warnings, "externalEvaluations": [ { "name": e.name, diff --git a/tests/genai/test_failed_turns.py b/tests/genai/test_failed_turns.py new file mode 100644 index 00000000..b9fcc80e --- /dev/null +++ b/tests/genai/test_failed_turns.py @@ -0,0 +1,73 @@ +"""An agent span that failed before any model call is still a turn, so failures count.""" + +from __future__ import annotations + +from agentevals.genai import semconv as sc +from agentevals.genai.extract import analyze, extract_conversation +from agentevals.otel.decode import decode_json_document +from agentevals.otel.model import build_traces +from agentevals.streaming.manager import invocation_dict +from otel.builders import agent, chat, request, sid, span + +STATUS_ERROR = 2 + + +def _traces(*spans: dict): + traces, _ = build_traces(decode_json_document(request(list(spans)), strict=True).spans) + return traces + + +def _rpc_child(span_id: str = "rpc") -> dict: + return span(span_id=span_id, parent="a1", name="TaskStoreService/UpdateTask", attrs={"rpc.method": "x/UpdateTask"}) + + +def _failed_agent(*, status: bool = True, error_type: str | None = None) -> dict: + data = agent() + if status: + data["status"] = {"code": STATUS_ERROR} + if error_type: + data["attributes"].append({"key": "error.type", "value": {"stringValue": error_type}}) + return data + + +def test_agent_span_with_error_status_and_no_work_is_a_failed_turn(): + conversation = extract_conversation(_traces(_failed_agent(error_type="runtime_error"), _rpc_child())) + + [turn] = conversation.turns + assert turn.ref.span_id == sid("a1") + assert (turn.status, turn.error_type) == ("error", "runtime_error") + assert turn.llm_calls == [] and turn.tool_calls == [] + assert turn.content_captured is False + + +def test_error_status_without_error_type_gets_the_fallback_type(): + [turn] = extract_conversation(_traces(_failed_agent())).turns + assert (turn.status, turn.error_type) == ("error", sc.ERROR_TYPE_OTHER) + + +def test_error_type_alone_marks_the_anchor_as_failed(): + [turn] = extract_conversation(_traces(_failed_agent(status=False, error_type="timeout"))).turns + assert turn.error_type == "timeout" + + +def test_empty_agent_span_without_an_error_is_still_not_a_turn(): + [trace] = _traces(agent(), _rpc_child()) + analysis = analyze(trace) + + assert analysis.anchors == [] + assert any("not a turn" in w for w in analysis.warnings) + + +def test_failed_agent_with_a_model_call_keeps_the_call(): + [turn] = extract_conversation(_traces(_failed_agent(error_type="runtime_error"), chat(parent="a1"))).turns + assert turn.status == "error" and len(turn.llm_calls) == 1 + + +def test_session_payload_reports_status_and_content_capture(): + [turn] = extract_conversation(_traces(_failed_agent(error_type="runtime_error"))).turns + payload = invocation_dict(turn, {}) + + assert payload["status"] == "error" + assert payload["errorType"] == "runtime_error" + assert payload["contentCaptured"] is False + assert payload["warnings"] == [] diff --git a/tests/test_api.py b/tests/test_api.py index 43815591..dd712b9e 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -425,6 +425,18 @@ def test_config_with_keys(self): assert keys["anthropic"] is True assert keys["openai"] is False + def test_config_without_receivers_reports_no_otlp_ports(self): + body = _assert_envelope(self.client.get("/api/config")) + assert body["data"]["otlp"] is None + + def test_config_reports_the_receiver_ports_the_server_started(self): + from agentevals.api.app import create_app + from agentevals.api.models import OtlpReceiverPorts + + app = create_app(otlp_ports=OtlpReceiverPorts(http_port=14318, grpc_port=14317)) + body = _assert_envelope(TestClient(app).get("/api/config")) + assert body["data"]["otlp"] == {"httpPort": 14318, "grpcPort": 14317} + # --------------------------------------------------------------------------- # GET /api/metrics diff --git a/ui/src/api/client.ts b/ui/src/api/client.ts index b08ae754..4931ebf9 100644 --- a/ui/src/api/client.ts +++ b/ui/src/api/client.ts @@ -278,7 +278,10 @@ export async function validateEvalSet(evalSetFile: File): Promise<{ valid: boole } } -export async function getConfig(): Promise<{ apiKeys: { google: boolean; anthropic: boolean; openai: boolean } }> { +export async function getConfig(): Promise<{ + apiKeys: { google: boolean; anthropic: boolean; openai: boolean }; + otlp?: { httpPort: number; grpcPort: number } | null; +}> { const response = await fetch(`${API_BASE_URL}/config`); if (!response.ok) { throw new Error(`Failed to fetch config: ${response.statusText}`); diff --git a/ui/src/components/inspector/InspectorView.tsx b/ui/src/components/inspector/InspectorView.tsx index 6eef5fbf..9986ece6 100644 --- a/ui/src/components/inspector/InspectorView.tsx +++ b/ui/src/components/inspector/InspectorView.tsx @@ -172,33 +172,10 @@ export const InspectorView: React.FC = () => { ? invocations.find(inv => inv.invocationId === selectedInvocationId) : invocations[0]; - // Match with expected invocation - const matchExpectedInvocation = (actual: Invocation): Invocation | null => { - if (expectedInvocations.length === 0) return null; - - // If only one expected invocation, use it - if (expectedInvocations.length === 1) { - return expectedInvocations[0]; - } - - // Try to match by user content text - const actualUserText = actual.userContent.parts - .filter(p => p.text) - .map(p => p.text) - .join(' ') - .toLowerCase() - .trim(); - - return expectedInvocations.find(exp => { - const expUserText = exp.userContent.parts - .filter(p => p.text) - .map(p => p.text) - .join(' ') - .toLowerCase() - .trim(); - return expUserText === actualUserText; - }) || null; - }; + // Same pairing as scoring: turn N against expected turn N. Matching by user text breaks when + // content was not captured, since every text is then empty. + const matchExpectedInvocation = (actual: Invocation): Invocation | null => + expectedInvocations[invocations.indexOf(actual)] ?? null; const expectedInvocation = selectedInvocation ? matchExpectedInvocation(selectedInvocation) : null; diff --git a/ui/src/components/streaming/LiveConversationPanel.tsx b/ui/src/components/streaming/LiveConversationPanel.tsx index c0bc0bae..be2f5f11 100644 --- a/ui/src/components/streaming/LiveConversationPanel.tsx +++ b/ui/src/components/streaming/LiveConversationPanel.tsx @@ -1,5 +1,5 @@ import { useRef, useEffect, useState, useMemo } from 'react'; -import { UserMessage, ToolCallMessage, ToolResultMessage, AgentMessage } from './LiveMessage'; +import { UserMessage, ToolCallMessage, ToolResultMessage, AgentMessage, TurnErrorMessage } from './LiveMessage'; import type { ConversationElement } from '../../lib/types'; export type { ConversationElement }; @@ -113,6 +113,8 @@ export function LiveConversationPanel({ elements, isActive }: LiveConversationPa timestamp={element.timestamp} isStreaming={false} />; + case 'turn_error': + return ; default: return null; } diff --git a/ui/src/components/streaming/LiveMessage.tsx b/ui/src/components/streaming/LiveMessage.tsx index 6c452258..150fa0d3 100644 --- a/ui/src/components/streaming/LiveMessage.tsx +++ b/ui/src/components/streaming/LiveMessage.tsx @@ -144,3 +144,21 @@ export function AgentMessage({ text, isStreaming }: AgentMessageProps) { ); } + +interface TurnErrorMessageProps { + errorType?: string | null; + timestamp: number; +} + +export function TurnErrorMessage({ errorType }: TurnErrorMessageProps) { + return ( +
+ Turn failed{errorType ? `: ${errorType}` : ''} +
+ ); +} diff --git a/ui/src/components/streaming/LiveStreamingView.tsx b/ui/src/components/streaming/LiveStreamingView.tsx index 9d024c3d..b609804f 100644 --- a/ui/src/components/streaming/LiveStreamingView.tsx +++ b/ui/src/components/streaming/LiveStreamingView.tsx @@ -2,6 +2,8 @@ import { useEffect, useRef, useState } from 'react'; import { useTraceContext } from '../../context/TraceContext'; import { SessionCard } from './SessionCard'; import { config } from '../../config'; +import { getConfig } from '../../api/client'; +import { isFailedTurn, pairToolResponses } from './turnStatus'; import type { ConversationElement, LiveSession, StreamingInvocation } from '../../lib/types'; function invocationsToElements(invocations: StreamingInvocation[]): ConversationElement[] { @@ -18,14 +20,16 @@ function invocationsToElements(invocations: StreamingInvocation[]): Conversation data: { text: inv.userText }, }); } - for (const tc of inv.toolCalls || []) { + const toolCalls = inv.toolCalls || []; + const responses = pairToolResponses(toolCalls, inv.toolResponses || []); + toolCalls.forEach((tc, i) => { elements.push({ type: 'tool_call', timestamp: base + seq++, invocationId: inv.invocationId, data: { toolCall: tc }, }); - const tr = inv.toolResponses?.find(r => r.id === tc.id || r.name === tc.name); + const tr = responses[i]; if (tr) { elements.push({ type: 'tool_result', @@ -34,7 +38,7 @@ function invocationsToElements(invocations: StreamingInvocation[]): Conversation data: { response: tr.response, isError: !!tr.response?.isError }, }); } - } + }); if (inv.agentText) { elements.push({ type: 'agent_response', @@ -43,6 +47,14 @@ function invocationsToElements(invocations: StreamingInvocation[]): Conversation data: { text: inv.agentText }, }); } + if (isFailedTurn(inv)) { + elements.push({ + type: 'turn_error', + timestamp: base + seq++, + invocationId: inv.invocationId, + data: { errorType: inv.errorType }, + }); + } return elements; }); } @@ -54,6 +66,11 @@ export function LiveStreamingView() { const [connectionStatus, setConnectionStatus] = useState<'connecting' | 'connected' | 'disconnected'>('connecting'); const [selectedGoldenId, setSelectedGoldenId] = useState(null); const [isPreparingEvaluation, setIsPreparingEvaluation] = useState(false); + const [otlpPorts, setOtlpPorts] = useState<{ httpPort: number; grpcPort: number } | null>(null); + + useEffect(() => { + getConfig().then(cfg => setOtlpPorts(cfg.otlp ?? null)).catch(() => {}); + }, []); const totalQueuedSessions = state.annotationQueues.reduce((sum, q) => sum + q.items.length, 0); @@ -502,13 +519,16 @@ export function LiveStreamingView() { } }; + const startedAtMs = (s: LiveSession) => Date.parse(s.startedAt) || 0; + const newestFirst = (a: LiveSession, b: LiveSession) => + startedAtMs(b) - startedAtMs(a) || b.sessionId.localeCompare(a.sessionId); const sessions = Array.from(activeSessions.values()); const activeLiveSessions = sessions .filter(s => s.status === 'active') - .sort((a, b) => b.sessionId.localeCompare(a.sessionId)); + .sort(newestFirst); const completedSessions = sessions .filter(s => s.status === 'complete') - .sort((a, b) => b.sessionId.localeCompare(a.sessionId)); + .sort(newestFirst); const allSessions = [...activeLiveSessions, ...completedSessions]; return ( @@ -763,12 +783,45 @@ export function LiveStreamingView() {

- Run your agent with streaming enabled to see traces appear here in real-time + Send OpenTelemetry traces and logs from your agent, or from your Collector, to agentevals.

+
+ OTLP HTTP + + http://{window.location.hostname}:{otlpPorts?.httpPort ?? 4318} + + OTLP gRPC + + {window.location.hostname}:{otlpPorts?.grpcPort ?? 4317} + +
+
    +
  1. + Set OTEL_EXPORTER_OTLP_ENDPOINT (or your Collector's exporter) to one of the endpoints above. +
    2. +
  2. + Turn on message content capture in your instrumentation. Without it, turns arrive with no text and cannot become an EvalSet. +
    3. +
) : (
diff --git a/ui/src/components/streaming/SessionCard.tsx b/ui/src/components/streaming/SessionCard.tsx index 7319bcba..bf3c9c48 100644 --- a/ui/src/components/streaming/SessionCard.tsx +++ b/ui/src/components/streaming/SessionCard.tsx @@ -2,27 +2,8 @@ import { useState, useEffect, useRef } from 'react'; import { LiveConversationPanel } from './LiveConversationPanel'; import type { ConversationElement } from './LiveConversationPanel'; import { SessionMetadata } from './SessionMetadata'; -import type { AnnotationQueue } from '../../lib/types'; - -interface Invocation { - invocationId: string; - userText: string; - agentText: string; - toolCalls: Array<{ name: string; args: any }>; - modelInfo?: { - models?: string[]; - inputTokens?: number; - outputTokens?: number; - provider?: string; - responseModels?: string[]; - finishReasons?: string[]; - cacheCreationTokens?: number; - cacheReadTokens?: number; - temperature?: number; - maxTokens?: number; - errorTypes?: string[]; - }; -} +import { isFailedTurn, lacksContent } from './turnStatus'; +import type { AnnotationQueue, StreamingInvocation } from '../../lib/types'; interface SessionCardProps { session: { @@ -32,7 +13,7 @@ interface SessionCardProps { spans: any[]; status: 'active' | 'complete'; metadata: Record; - invocations?: Invocation[]; + invocations?: StreamingInvocation[]; liveElements?: ConversationElement[]; liveStats?: { totalInputTokens: number; @@ -105,6 +86,11 @@ export function SessionCard({ session, isSelected, onSelect, onRemove, evaluatio const totalCacheCreation = session.invocations?.reduce((sum, inv) => sum + (inv.modelInfo?.cacheCreationTokens || 0), 0) || 0; const totalCacheRead = session.invocations?.reduce((sum, inv) => sum + (inv.modelInfo?.cacheReadTokens || 0), 0) || 0; + const failedTurns = session.invocations?.filter(isFailedTurn).length ?? 0; + const noContent = lacksContent(session.invocations); + const warnings = Array.from(new Set(session.invocations?.flatMap(inv => inv.warnings ?? []) ?? [])); + const evalSetBlocked = noContent && !isSelected; + return (
)} + {failedTurns > 0 && ( + + {failedTurns} failed turn{failedTurns !== 1 ? 's' : ''} + + )} + + {noContent && ( + + content off + + )} + + {warnings.length > 0 && ( + + {warnings.length} warning{warnings.length !== 1 ? 's' : ''} + + )} + {queueNames && queueNames.length > 0 && queueNames.map(name => (
+ {evalSetBlocked && ( +

+ This session has no message content, so it cannot be set as the EvalSet. Turn on content capture in your instrumentation and run the agent again. +

+ )} + {expanded && session.startedAt && (totalTokens > 0 || Object.keys(session.metadata).length > 0) && ( [number]; + +export function isFailedTurn(inv: Pick): boolean { + return inv.status === 'error' || !!inv.errorType; +} + +/** True only when the server reported every turn as captured without message content. */ +export function lacksContent(invocations: StreamingInvocation[] | undefined): boolean { + return !!invocations?.length && invocations.every(inv => inv.contentCaptured === false); +} + +/** + * The response for each call, or undefined. Ids decide when both sides have one; otherwise the + * next unused response with the same name, so repeated calls to one tool keep their own results. + */ +export function pairToolResponses(calls: ToolCall[], responses: ToolResponse[]): Array { + const used = new Set(); + const take = (match: (r: ToolResponse) => boolean) => { + const i = responses.findIndex((r, j) => !used.has(j) && match(r)); + if (i < 0) return undefined; + used.add(i); + return responses[i]; + }; + return calls.map(tc => + (tc.id ? take(r => r.id === tc.id) : undefined) + ?? take(r => r.name === tc.name && !(tc.id && r.id)), + ); +} diff --git a/ui/src/context/TraceProvider.tsx b/ui/src/context/TraceProvider.tsx index 51e350e0..6738f6aa 100644 --- a/ui/src/context/TraceProvider.tsx +++ b/ui/src/context/TraceProvider.tsx @@ -333,7 +333,7 @@ export const TraceProvider: React.FC = ({ children }) => { const newItem = { sessionId: session.sessionId, traceId: session.traceId, - agentName: session.metadata?.agentName, + agentName: session.metadata?.['gen_ai.agent.name'] || session.metadata?.['service.name'], startTime: session.startedAt, model: session.liveStats?.model || session.metadata?.model, totalTokens: totalTokens > 0 ? totalTokens : undefined, diff --git a/ui/src/lib/types.ts b/ui/src/lib/types.ts index 7f0a6ddc..843031fe 100644 --- a/ui/src/lib/types.ts +++ b/ui/src/lib/types.ts @@ -478,7 +478,7 @@ export const AVAILABLE_METRICS: MetricMetadata[] = [ // Streaming / Live session types export interface ConversationElement { - type: 'user_input' | 'tool_call' | 'tool_result' | 'agent_response'; + type: 'user_input' | 'tool_call' | 'tool_result' | 'agent_response' | 'turn_error'; timestamp: number; invocationId: string; data: any; @@ -502,6 +502,10 @@ export interface StreamingInvocation { toolCalls: Array<{ name: string; args: any; id?: string }>; toolResponses?: Array<{ name: string; response: any; id?: string }>; modelInfo?: Record; + status?: 'unset' | 'ok' | 'error'; + errorType?: string | null; + contentCaptured?: boolean; + warnings?: string[]; /** ``gen_ai.evaluation.result`` events found in the telemetry (from other evaluators). */ externalEvaluations?: ExternalEvaluation[]; } diff --git a/uv.lock b/uv.lock index 83a9c091..7124698f 100644 --- a/uv.lock +++ b/uv.lock @@ -138,7 +138,7 @@ e2e = [ { name = "langchain", specifier = ">=1.2.0" }, { name = "langchain-openai", specifier = ">=1.1.10" }, { name = "openai", specifier = ">=2.30.0" }, - { name = "openai-agents", specifier = ">=0.13.0" }, + { name = "openai-agents", specifier = ">=0.19.0" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.36.0" }, { name = "opentelemetry-instrumentation-openai-agents-v2", specifier = ">=0.1.0" }, { name = "opentelemetry-instrumentation-openai-v2" }, @@ -2197,7 +2197,7 @@ wheels = [ [[package]] name = "openai-agents" -version = "0.13.6" +version = "0.20.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "griffelib" }, @@ -2205,12 +2205,12 @@ dependencies = [ { name = "openai" }, { name = "pydantic" }, { name = "requests" }, - { name = "types-requests" }, { name = "typing-extensions" }, + { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/4f/e8/a3bc1a91af9c71d2934f8e2f3eee2954540fa95d47b0e3f155d348d91b38/openai_agents-0.13.6.tar.gz", hash = "sha256:de7b3add7933ae704a5ee6e531f650d8aabb3ebaa1631f458ba39684a5ed966e", size = 2704270, upload-time = "2026-04-09T04:10:51.581Z" } +sdist = { url = "https://files.pythonhosted.org/packages/57/18/1204bde976436ad85498b759d5297843ebd2e0a214f729eb2dc4423cdfc7/openai_agents-0.20.0.tar.gz", hash = "sha256:a6b37954877868ac233876a27dab1e2a461c53ecbd5fcf6e19e2d276d3e848cd", size = 6146452, upload-time = "2026-08-11T03:12:49.556Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/1c/83/a991b2ad389abadabf13f6c4228bd88ac8dc363e4b50fcae8c5ea966bd41/openai_agents-0.13.6-py3-none-any.whl", hash = "sha256:8decb9eb0cc5dbe7749858e97a7d8316f9439526ca4e539e3bd105e0eb41115e", size = 471763, upload-time = "2026-04-09T04:10:49.81Z" }, + { url = "https://files.pythonhosted.org/packages/9f/8b/18528f00cb1a5d0b8e00f457f7dc88e7b17439bc520861aae2b638b88710/openai_agents-0.20.0-py3-none-any.whl", hash = "sha256:aaff662b802fa90762ad539e131b9ea387e12e3664b87bc75157ad1b3fc88850", size = 1064338, upload-time = "2026-08-11T03:12:47.615Z" }, ] [[package]] @@ -3721,18 +3721,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a7/24/5480c20380dfd18cf33d14784096dca45a24eae6102e91d49a718d3b6855/typer_slim-0.24.0-py3-none-any.whl", hash = "sha256:d5d7ee1ee2834d5020c7c616ed5e0d0f29b9a4b1dd283bdebae198ec09778d0e", size = 3394, upload-time = "2026-02-16T22:08:49.92Z" }, ] -[[package]] -name = "types-requests" -version = "2.33.0.20260408" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "urllib3" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/69/6a/749dc53a54a3f35842c1f8197b3ca6b54af6d7458a1bfc75f6629b6da666/types_requests-2.33.0.20260408.tar.gz", hash = "sha256:95b9a86376807a216b2fb412b47617b202091c3ea7c078f47cc358d5528ccb7b", size = 23882, upload-time = "2026-04-08T04:34:49.33Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/90/b8/78fd6c037de4788c040fdd323b3369804400351b7827473920f6c1d03c10/types_requests-2.33.0.20260408-py3-none-any.whl", hash = "sha256:81f31d5ea4acb39f03be7bc8bed569ba6d5a9c5d97e89f45ac43d819b68ca50f", size = 20739, upload-time = "2026-04-08T04:34:48.325Z" }, -] - [[package]] name = "typing-extensions" version = "4.15.0"