From 3efb90981b85038d93c9f220d3a94353b175fb2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=A0=84=EC=9D=B8=EC=88=98?= Date: Thu, 15 Oct 2026 11:40:40 +0900 Subject: [PATCH 1/3] =?UTF-8?q?feat(observability):=20LLM=20=ED=98=B8?= =?UTF-8?q?=EC=B6=9C=C2=B7=EB=8F=84=EA=B5=AC=20=EC=8B=A4=ED=96=89=20OpenTe?= =?UTF-8?q?lemetry=20span?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 배경 dev Tempo 의 Geny 턴 trace 에 `chat` span 이 하나도 없었다. LLM 의 `POST /v1/chat/completions` 5건과 도구가 부른 HTTP·MCP 호출이 에이전트 턴이 아니라 워크플로우 span 바로 아래에 흩어져, 어느 호출이 병목인지 trace 에서 읽을 수 없었다. 호스트는 턴 끝의 사용량 합계와 도구 사건만 받으므로 호출 구간도 부모 관계도 되살릴 수 없다. 그래서 호출과 실행이 실제로 일어나는 런타임이 span 을 연다. ## 변경 - `observability/` 패키지 추가 (`spans` 수명 / `attributes` 키 / `genai` 계측) - `APIStage._traced_call`: Stage 6 호출 1건(재시도 포함) → `chat {model}` - `RegistryRouter.route`: 도구 실행 1건 → `execute_tool {tool}` (모든 실행기와 CLI 도구 표면이 지나는 한 곳) - 4.91.0, CHANGELOG ## 동작 규칙 - span 은 호출 시점의 현재 context 의 자식이다. 호스트가 턴 span 을 현재 context 로 두고 `run_turn`/`stream_turn` 을 부르면 그 아래에 달린다. - 토큰(`gen_ai.usage.*`)은 호출 span 에만 둔다. 호스트는 `spans_enabled()` 가 참이면 자기 쪽 도구 span 과 턴 span 의 토큰을 끈다 (중복 span · 이중 합산 방지). - 프롬프트·응답·도구 입출력 본문은 남기지 않는다. 취소는 오류로 표시하지 않는다. - `opentelemetry-api` 는 선택 의존성이다. 없거나 `XGEN_AGENT_RUNTIME_OTEL=0` 이면 no-op, span 조작 실패는 실행을 깨지 않는다. ## 테스트 - `tests/test_otel_spans.py` 6건: chat/execute_tool 이 호스트 턴 아래에 달림, HTTP span 이 각 호출 아래에 중첩, 실패 도구의 오류 표시와 본문 미기록, 스위치 off, tracer 고장 시 턴 유지 - 단위 테스트 전체: 변경 전후 실패 목록 동일 (Windows 에서 셸·서브프로세스를 실행하는 기존 실패) - 실제 Tempo trace 는 워크플로우 쪽 변경과 함께 배포한 뒤 확인 필요 ## 함께 필요한 변경 xgen-workflow `fix/geny-otel-trace`: 턴을 노드 실행 시작에서 열고 턴 context 를 전파한다. 이 PR 의 릴리스 후 워크플로우의 런타임 핀(`pyproject.toml`, `requirements.lock`)을 v4.91.0 으로 올린다. 워크플로우 변경은 옛 런타임에서도 기존 방식으로 동작한다. --- CHANGELOG.md | 26 +++ pyproject.toml | 2 +- .../observability/__init__.py | 48 +++++ .../observability/attributes.py | 35 ++++ src/xgen_agent_runtime/observability/genai.py | 121 +++++++++++ src/xgen_agent_runtime/observability/spans.py | 140 +++++++++++++ .../stages/s06_api/artifact/default/stage.py | 44 +++- .../s10_tool/artifact/default/routers.py | 12 ++ tests/test_otel_spans.py | 195 ++++++++++++++++++ 9 files changed, 621 insertions(+), 2 deletions(-) create mode 100644 src/xgen_agent_runtime/observability/__init__.py create mode 100644 src/xgen_agent_runtime/observability/attributes.py create mode 100644 src/xgen_agent_runtime/observability/genai.py create mode 100644 src/xgen_agent_runtime/observability/spans.py create mode 100644 tests/test_otel_spans.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f01626c..c89c21b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,32 @@ All notable changes to `xgen-agent-runtime` are recorded here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/). +## [4.91.0] - 2026-10-15 + +### Added: LLM 호출·도구 실행 OpenTelemetry span (`xgen_agent_runtime.observability`) + +dev Tempo 의 Geny 턴 trace(2026-10-15)에는 `chat` span 이 하나도 없었다. LLM 의 `POST /v1/chat/completions` +다섯 건과 도구가 부른 HTTP·MCP 호출은 에이전트 턴이 아니라 워크플로우 span 바로 아래에 흩어져, 어느 호출이 병목인지 +trace 에서 읽을 수 없었다. 호스트는 턴 끝의 사용량 합계와 도구 사건만 받으므로 호출 구간도 부모 관계도 되살릴 수 없다. +그래서 호출과 실행이 일어나는 런타임이 span 을 연다. + +- **`chat {model}`**: Stage 6 의 호출 1건(재시도 포함)마다. 현재 span 으로 열어 벤더 SDK 의 HTTP 자동 계측 span 이 그 + 아래에 달린다. 속성은 GenAI semantic conventions — `gen_ai.provider.name`·`gen_ai.request.model`· + `gen_ai.response.model`·`gen_ai.response.finish_reasons`·`gen_ai.usage.*`(캐시 토큰은 0 이면 생략), 그리고 + `xgen.llm.stream`·`message_count`·`has_tools`·`output_length`·`tool_calls`. `APIError` 는 오류 코드(없으면 분류)를 + `error.type` 으로 남긴다. +- **`execute_tool {tool}`**: `RegistryRouter.route` 에서 도구 실행 1건마다. 모든 실행기(순차·병렬·분할·스트리밍)와 + CLI 도구 표면이 여기를 지나므로 한 곳에서 연다. 도구가 부르는 HTTP·MCP 호출이 이 span 아래에 달린다. + `xgen.tool.is_error`·`xgen.tool.result_size` 를 남기고, 실패면 `error.type=tool_error`. +- 프롬프트·응답·도구 입출력 **본문은 남기지 않는다**(길이·개수만). 취소는 오류로 표시하지 않는다. +- span 의 부모는 호출 시점의 현재 context 다. 턴을 도는 전용 이벤트 루프의 Task 가 호출 스레드의 context 를 + 이어받으므로, 호스트가 턴 span 을 현재 context 로 두고 `run_turn`/`stream_turn` 을 부르면 그 아래에 달린다. +- **선택 의존성**: `opentelemetry-api` 가 없거나 `XGEN_AGENT_RUNTIME_OTEL=0` 이면 전 경로가 no-op 이다. 호스트는 + `observability.spans_enabled()` 가 참이면 자기 쪽 같은 span(사건으로 만든 도구 span)과 턴 span 의 `gen_ai.usage.*` 를 + 끈다 — 둘 다 내면 같은 호출이 두 번 나오고 토큰이 두 번 더해진다. span 조작 실패는 실행을 깨지 않는다. +- 구성: `observability.spans`(span 수명 — 켜짐 판정·tracer·현재 span 으로 열고 닫기), `observability.attributes` + (속성 키 — xgen-workflow 와 같은 이름), `observability.genai`(`chat`/`execute_tool` span 과 속성 기록). + ## [4.90.0] - 2026-10-13 ### Added: CLI 구독 한도 신호와 한도 소진 분류 diff --git a/pyproject.toml b/pyproject.toml index 1d15b27..3288cdd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "xgen-agent-runtime" -version = "4.90.0" +version = "4.91.0" description = "Harness-engineered agent pipeline library with 21-stage dual-abstraction architecture, built on the Anthropic API" readme = "README.md" license = "Apache-2.0" diff --git a/src/xgen_agent_runtime/observability/__init__.py b/src/xgen_agent_runtime/observability/__init__.py new file mode 100644 index 0000000..80a134e --- /dev/null +++ b/src/xgen_agent_runtime/observability/__init__.py @@ -0,0 +1,48 @@ +"""OpenTelemetry 계측 — 런타임이 **직접 하는 일**(LLM 호출 1건, 도구 실행 1건)을 span 으로 남긴다. + +왜 런타임이 내는가 +------------------ +호스트(xgen-workflow)는 턴이 끝난 뒤 받는 사용량 합계와 도구 시작/끝 사건으로만 span 을 +만들 수 있었다. 그 방식으로는 두 가지가 끝내 안 된다. + +* **LLM 호출 1건의 구간.** 사건은 큐를 지나 늦게 닿아 실제 호출 구간이 아니다. 그래서 trace 에 + ``chat`` span 이 없고 ``POST /v1/chat/completions`` 같은 HTTP span 만 남아 병목을 읽을 수 없었다. +* **부모 관계.** 사건으로 연 span 은 현재 context 가 아니라서, 도구가 부르는 HTTP·MCP 호출이 + 그 도구 아래로 가지 못하고 턴 밖에 흩어졌다. + +호출과 실행이 일어나는 자리가 여기이므로 span 도 여기서 연다. span 은 현재 context 의 자식이 되고, +그 안의 HTTP 자동 계측 span 은 이 span 의 자식이 된다. + +구성 +---- +* ``spans`` — span 수명(켜짐 판정·tracer·현재 span 으로 열고 닫기). GenAI 를 모른다. +* ``attributes`` — 속성 키. OpenTelemetry GenAI semantic conventions + ``xgen.*`` 확장. +* ``genai`` — ``chat {model}`` / ``execute_tool {tool}`` span 과 응답·결과의 속성 기록. + +규약 +---- +* 토큰(``gen_ai.usage.*``)은 **호출 span 에만** 붙인다. 호스트는 :func:`spans_enabled` 가 참이면 + 턴 합계를 ``gen_ai.usage.*`` 로 다시 쓰지 않는다 — 더할 때 두 번 세지 않게. +* 프롬프트·응답·도구 입출력 **본문은 남기지 않는다.** 길이·개수만. +* ``opentelemetry-api`` 는 선택 의존성이다. 없거나 ``XGEN_AGENT_RUNTIME_OTEL=0`` 이면 전 경로가 + no-op 이다. 호스트가 SDK 를 설정하지 않았으면 API 가 no-op span 을 준다. +* 관측이 실행을 깨선 안 된다 — span 조작 실패는 삼킨다. +""" + +from xgen_agent_runtime.observability.genai import ( + llm_call_span, + record_llm_response, + record_tool_result, + tool_span, +) +from xgen_agent_runtime.observability.spans import ENV_SWITCH, SpanHandle, spans_enabled + +__all__ = [ + "ENV_SWITCH", + "SpanHandle", + "llm_call_span", + "record_llm_response", + "record_tool_result", + "spans_enabled", + "tool_span", +] diff --git a/src/xgen_agent_runtime/observability/attributes.py b/src/xgen_agent_runtime/observability/attributes.py new file mode 100644 index 0000000..f185c8e --- /dev/null +++ b/src/xgen_agent_runtime/observability/attributes.py @@ -0,0 +1,35 @@ +"""span 속성 키 — OpenTelemetry GenAI semantic conventions + ``xgen.*`` 확장. + +호스트(xgen-workflow ``service/observability/attributes.py``)와 **같은 키**를 쓴다. 대시보드·알림이 +키 이름으로 읽으므로, 여기서 이름을 바꾸면 호스트 쪽 조회가 조용히 비게 된다. +""" + +from __future__ import annotations + +# ── GenAI semantic conventions ───────────────────────────────────────────── +GEN_AI_OPERATION_NAME = "gen_ai.operation.name" +GEN_AI_PROVIDER_NAME = "gen_ai.provider.name" +GEN_AI_REQUEST_MODEL = "gen_ai.request.model" +GEN_AI_RESPONSE_MODEL = "gen_ai.response.model" +GEN_AI_RESPONSE_ID = "gen_ai.response.id" +GEN_AI_RESPONSE_FINISH_REASONS = "gen_ai.response.finish_reasons" +GEN_AI_USAGE_INPUT_TOKENS = "gen_ai.usage.input_tokens" +GEN_AI_USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens" +GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = "gen_ai.usage.cache_read.input_tokens" +GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS = "gen_ai.usage.cache_creation.input_tokens" +GEN_AI_TOOL_NAME = "gen_ai.tool.name" +GEN_AI_TOOL_TYPE = "gen_ai.tool.type" +ERROR_TYPE = "error.type" + +OP_CHAT = "chat" +OP_EXECUTE_TOOL = "execute_tool" +TOOL_TYPE_FUNCTION = "function" + +# ── xgen 확장 ────────────────────────────────────────────────────────────── +XGEN_LLM_STREAM = "xgen.llm.stream" +XGEN_LLM_MESSAGE_COUNT = "xgen.llm.message_count" +XGEN_LLM_HAS_TOOLS = "xgen.llm.has_tools" +XGEN_LLM_OUTPUT_LENGTH = "xgen.llm.output_length" +XGEN_LLM_TOOL_CALLS = "xgen.llm.tool_calls" +XGEN_TOOL_IS_ERROR = "xgen.tool.is_error" +XGEN_TOOL_RESULT_SIZE = "xgen.tool.result_size" diff --git a/src/xgen_agent_runtime/observability/genai.py b/src/xgen_agent_runtime/observability/genai.py new file mode 100644 index 0000000..3593009 --- /dev/null +++ b/src/xgen_agent_runtime/observability/genai.py @@ -0,0 +1,121 @@ +"""GenAI span — LLM 호출 1건(``chat {model}``)과 도구 실행 1건(``execute_tool {tool}``). + +이름 ``{...}`` 자리에는 저카디널리티 값(모델·도구 이름)만 들어간다 — 규약이 허용하는 예외이고, +trace UI 의 operation 목록에서 바로 읽힌다. span 수명은 ``spans`` 에 맡기고, 여기서는 무엇을 +어떤 키로 남길지만 정한다. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Any, Dict, Iterator, Optional + +from xgen_agent_runtime.observability import attributes as attrs +from xgen_agent_runtime.observability.spans import ( + SpanHandle, + client_kind, + current_span, + internal_kind, +) + + +def _span_name(operation: str, target: str) -> str: + target = (target or "").strip() + return f"{operation} {target}" if target else operation + + +# ── LLM 호출 ──────────────────────────────────────────────────────────────── + + +@contextmanager +def llm_call_span( + *, + model: str, + provider: str = "", + stream: bool = False, + message_count: Optional[int] = None, + has_tools: Optional[bool] = None, +) -> Iterator[SpanHandle]: + """LLM 호출 1건(재시도 포함). 응답은 :func:`record_llm_response` 로 남긴다.""" + attributes: Dict[str, Any] = { + attrs.GEN_AI_OPERATION_NAME: attrs.OP_CHAT, + attrs.GEN_AI_PROVIDER_NAME: provider, + attrs.GEN_AI_REQUEST_MODEL: model, + attrs.XGEN_LLM_STREAM: bool(stream), + attrs.XGEN_LLM_MESSAGE_COUNT: message_count, + attrs.XGEN_LLM_HAS_TOOLS: has_tools, + } + with current_span(_span_name(attrs.OP_CHAT, model), client_kind(), attributes) as span: + yield span + + +def _usage_attributes(usage: Any) -> Dict[str, Any]: + """호출 1건의 토큰. 캐시 토큰은 0 이면 생략한다 — 캐시를 안 쓰는 모델이 대부분이라 잡음이다.""" + payload: Dict[str, Any] = { + attrs.GEN_AI_USAGE_INPUT_TOKENS: int(getattr(usage, "input_tokens", 0) or 0), + attrs.GEN_AI_USAGE_OUTPUT_TOKENS: int(getattr(usage, "output_tokens", 0) or 0), + } + cache_read = int(getattr(usage, "cache_read_input_tokens", 0) or 0) + cache_write = int(getattr(usage, "cache_creation_input_tokens", 0) or 0) + if cache_read: + payload[attrs.GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS] = cache_read + if cache_write: + payload[attrs.GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS] = cache_write + return payload + + +def record_llm_response(span: SpanHandle, response: Any) -> None: + """응답의 모델·종료 사유·토큰·크기를 호출 span 에 남긴다. 본문은 남기지 않는다.""" + if not span.recording or response is None: + return + payload: Dict[str, Any] = { + attrs.GEN_AI_RESPONSE_MODEL: getattr(response, "model", None), + attrs.GEN_AI_RESPONSE_ID: getattr(response, "message_id", None), + attrs.XGEN_LLM_OUTPUT_LENGTH: len(getattr(response, "text", "") or ""), + attrs.XGEN_LLM_TOOL_CALLS: len(getattr(response, "tool_calls", None) or []), + } + stop_reason = getattr(response, "stop_reason", "") + if stop_reason: + payload[attrs.GEN_AI_RESPONSE_FINISH_REASONS] = [str(stop_reason)] + usage = getattr(response, "usage", None) + if usage is not None: + payload.update(_usage_attributes(usage)) + span.set(payload) + + +# ── 도구 실행 ─────────────────────────────────────────────────────────────── + + +@contextmanager +def tool_span(*, tool_name: str) -> Iterator[SpanHandle]: + """도구 실행 1건. 결과는 :func:`record_tool_result` 로 남긴다.""" + attributes: Dict[str, Any] = { + attrs.GEN_AI_OPERATION_NAME: attrs.OP_EXECUTE_TOOL, + attrs.GEN_AI_TOOL_NAME: tool_name, + attrs.GEN_AI_TOOL_TYPE: attrs.TOOL_TYPE_FUNCTION, + } + with current_span(_span_name(attrs.OP_EXECUTE_TOOL, tool_name), internal_kind(), attributes) as span: + yield span + + +def _result_size(result: Any) -> Optional[int]: + content = getattr(result, "display_text", None) or getattr(result, "content", None) + if isinstance(content, (str, bytes, list, dict)): + return len(content) + return None + + +def record_tool_result(span: SpanHandle, result: Any) -> None: + """결과의 성패와 크기만 남긴다. 본문은 남기지 않는다.""" + if not span.recording or result is None: + return + is_error = bool(getattr(result, "is_error", False)) + span.set({ + attrs.XGEN_TOOL_IS_ERROR: is_error, + attrs.XGEN_TOOL_RESULT_SIZE: _result_size(result), + }) + if is_error: + span.fail("tool_error") + + +__all__ = ["llm_call_span", "record_llm_response", "record_tool_result", "tool_span"] diff --git a/src/xgen_agent_runtime/observability/spans.py b/src/xgen_agent_runtime/observability/spans.py new file mode 100644 index 0000000..c87c7d1 --- /dev/null +++ b/src/xgen_agent_runtime/observability/spans.py @@ -0,0 +1,140 @@ +"""span 수명 — 켜짐 판정, tracer, 현재 span 으로 열고 닫기. + +무엇을 계측하는지(GenAI 이름·속성)는 모른다 — 그건 ``genai`` 의 몫이다. 여기서는 "span 하나를 +안전하게 열고 닫는다" 만 책임진다: OTel 이 없거나 꺼졌거나 고장 나도 실행은 그대로 간다. +""" + +from __future__ import annotations + +import logging +import os +from contextlib import contextmanager +from typing import Any, Iterator, Mapping + +from xgen_agent_runtime.observability.attributes import ERROR_TYPE + +logger = logging.getLogger(__name__) + +try: # pragma: no cover - 설치 여부에 따라 한쪽만 탄다 + from opentelemetry import trace as _otel_trace + from opentelemetry.trace import SpanKind + from opentelemetry.trace import Status as _Status + from opentelemetry.trace import StatusCode as _StatusCode +except ImportError: # pragma: no cover + _otel_trace = None # type: ignore[assignment] + SpanKind = None # type: ignore[assignment,misc] + _Status = None # type: ignore[assignment] + _StatusCode = None # type: ignore[assignment] + +#: 끄는 스위치. 호스트가 같은 일을 자기 방식으로 계측하고 있으면 0 으로 둔다. +ENV_SWITCH = "XGEN_AGENT_RUNTIME_OTEL" + +TRACER_NAME = "xgen_agent_runtime" + +_OFF_VALUES = frozenset({"0", "false", "no", "off"}) + + +def spans_enabled() -> bool: + """이 프로세스에서 런타임이 span 을 내는가. + + 호스트는 이 값으로 자기 쪽 같은 span(사건으로 만든 도구 span, 턴 span 의 토큰)을 끈다. + 둘 다 내면 같은 호출이 trace 에 두 번 나온다. + """ + if _otel_trace is None: + return False + return os.environ.get(ENV_SWITCH, "1").strip().lower() not in _OFF_VALUES + + +def get_tracer() -> Any: + """전역 provider 의 tracer. 테스트가 이 함수만 바꿔 끼운다.""" + return _otel_trace.get_tracer(TRACER_NAME) + + +def client_kind() -> Any: + return SpanKind.CLIENT if SpanKind is not None else None + + +def internal_kind() -> Any: + return SpanKind.INTERNAL if SpanKind is not None else None + + +class SpanHandle: + """열린 span 하나. 비활성이면 ``span`` 이 None 이라 모든 메서드가 no-op 이다.""" + + __slots__ = ("span",) + + def __init__(self, span: Any = None) -> None: + self.span = span + + @property + def recording(self) -> bool: + return self.span is not None + + def set(self, attributes: Mapping[str, Any]) -> None: + """None·빈 문자열은 건너뛴다 — "모름" 을 값으로 남기지 않는다.""" + if self.span is None: + return + try: + for key, value in attributes.items(): + if value is None or value == "": + continue + self.span.set_attribute(key, value) + except Exception: # noqa: BLE001 — 관측이 실행을 깨선 안 된다 + logger.debug("span attribute 기록 실패(무시)", exc_info=True) + + def fail(self, error_type: str) -> None: + """오류로 표시한다. 분류 코드만 — 자유 텍스트는 PII 경로다.""" + if self.span is None: + return + try: + self.span.set_attribute(ERROR_TYPE, error_type or "error") + self.span.set_status(_Status(_StatusCode.ERROR, error_type or "error")) + except Exception: # noqa: BLE001 + logger.debug("span 오류 표시 실패(무시)", exc_info=True) + + +@contextmanager +def current_span(name: str, kind: Any, attributes: Mapping[str, Any]) -> Iterator[SpanHandle]: + """``name`` span 을 현재 span 으로 열고 블록이 끝나면 닫는다. + + 블록 안의 예외는 그대로 올라가고 span 에는 오류로 남는다. 취소(``CancelledError``)와 제너레이터 + 종료는 오류가 아니다 — 사용자가 멈춘 턴이 빨갛게 보이면 안 된다. + """ + if not spans_enabled(): + yield SpanHandle() + return + try: + cm = get_tracer().start_as_current_span( + name, kind=kind, record_exception=False, set_status_on_exception=False + ) + raw = cm.__enter__() + except Exception: # noqa: BLE001 — span 을 못 열어도 실행은 간다 + logger.debug("span 시작 실패(무시): %s", name, exc_info=True) + yield SpanHandle() + return + handle = SpanHandle(raw) + handle.set(attributes) + exc_info: tuple = (None, None, None) + try: + yield handle + except BaseException as exc: + exc_info = (type(exc), exc, exc.__traceback__) + if isinstance(exc, Exception): + handle.fail(type(exc).__name__) + raise + finally: + try: + cm.__exit__(*exc_info) + except Exception: # noqa: BLE001 + logger.debug("span 종료 실패(무시): %s", name, exc_info=True) + + +__all__ = [ + "ENV_SWITCH", + "SpanHandle", + "client_kind", + "current_span", + "get_tracer", + "internal_kind", + "spans_enabled", +] diff --git a/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py b/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py index d1659c9..fdf0510 100644 --- a/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py +++ b/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py @@ -18,7 +18,7 @@ import asyncio import time -from typing import Any, AsyncIterator, Dict, List, Optional, Union +from typing import Any, AsyncIterator, Awaitable, Callable, Dict, List, Optional, Union from xgen_agent_runtime.core.compaction import note_api_call from xgen_agent_runtime.core.errors import APIError, ErrorCategory, ExecutorErrorCode @@ -33,6 +33,7 @@ from xgen_agent_runtime.core.stage import Stage from xgen_agent_runtime.core.state import PipelineState from xgen_agent_runtime.core.config import ModelConfig +from xgen_agent_runtime import observability from xgen_agent_runtime.llm_client import BaseClient, ClientCapabilities, ClientRegistry from xgen_agent_runtime.stages.s06_api.interface import ( APIProvider, @@ -498,6 +499,16 @@ async def call_once( state.shared.pop("_api_ttft_emitted", None) async def _invoke() -> APIResponse: + return await self._traced_call( + _call_provider, + client=client, + cfg=cfg, + stream=use_stream, + message_count=len(state.messages) + len(extra_messages or []), + has_tools=bool(state.tools), + ) + + async def _call_provider() -> APIResponse: if use_stream: return await self._call_streaming_with_retry( client, cfg, state, extra_messages=extra_messages @@ -669,6 +680,37 @@ async def _recover_overflow(self, state: PipelineState, error: APIError) -> bool self._signal_stream_restart(state) return shrunk + @staticmethod + async def _traced_call( + call: Callable[[], Awaitable[APIResponse]], + *, + client: BaseClient, + cfg: ModelConfig, + stream: bool, + message_count: int, + has_tools: bool, + ) -> APIResponse: + """호출 1건(재시도 포함)을 ``chat {model}`` span 으로 감싼다. + + 현재 span 으로 열어 벤더 SDK 의 HTTP 자동 계측 span 이 이 호출 아래에 달린다. 무엇을 + 어떤 키로 남기는지는 ``xgen_agent_runtime.observability`` 가 정하고, 여기서는 오류를 + 분류 코드로 옮기는 일만 한다 — 자유 텍스트(메시지)는 남기지 않는다. + """ + with observability.llm_call_span( + model=cfg.model, + provider=getattr(client, "provider", ""), + stream=stream, + message_count=message_count, + has_tools=has_tools, + ) as span: + try: + response = await call() + except APIError as e: + span.fail(e.code.value if e.code is not None else e.category.value) + raise + observability.record_llm_response(span, response) + return response + def _build_request(self, state: PipelineState) -> APIRequest: """Assemble a canonical :class:`APIRequest` from state. diff --git a/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py b/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py index 42b825b..114f07a 100644 --- a/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py +++ b/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py @@ -10,6 +10,7 @@ import jsonschema +from xgen_agent_runtime import observability from xgen_agent_runtime.hooks.events import HookEvent, HookEventPayload from xgen_agent_runtime.permission.matrix import evaluate_permission from xgen_agent_runtime.permission.types import ( @@ -207,6 +208,17 @@ def description(self) -> str: async def route( self, tool_name: str, tool_input: Dict[str, Any], context: ToolContext + ) -> ToolResult: + # 도구 실행 1건 = ``execute_tool {name}`` span. 모든 실행기(순차·병렬·분할·스트리밍)와 + # CLI 도구 표면이 여기를 지나므로 한 곳에서 연다. 현재 span 이라 도구가 부르는 + # HTTP·MCP 호출이 이 도구 아래에 달린다 — ``xgen_agent_runtime.observability``. + with observability.tool_span(tool_name=tool_name) as span: + result = await self._route(tool_name, tool_input, context) + observability.record_tool_result(span, result) + return result + + async def _route( + self, tool_name: str, tool_input: Dict[str, Any], context: ToolContext ) -> ToolResult: tool = self._registry.get(tool_name) if tool is None: diff --git a/tests/test_otel_spans.py b/tests/test_otel_spans.py new file mode 100644 index 0000000..a6ea175 --- /dev/null +++ b/tests/test_otel_spans.py @@ -0,0 +1,195 @@ +"""런타임이 LLM 호출·도구 실행마다 OTel span 을 내고, 그 span 이 호스트 턴 아래에 달린다. + +실측(2026-10-15, dev Tempo): Geny 턴 trace 에 ``chat`` span 이 하나도 없었고, LLM 의 +``POST /v1/chat/completions`` 와 도구가 부른 HTTP·MCP 호출이 에이전트 턴 밖(워크플로우 바로 +아래)에 흩어져 어느 호출이 병목인지 읽을 수 없었다. 호스트는 사건만 받아서는 호출 구간도 +부모 관계도 되살릴 수 없다 — 그래서 일이 일어나는 런타임이 span 을 연다. +""" + +from __future__ import annotations + +from typing import Any, Dict, List + +import pytest + +pytest.importorskip("opentelemetry.sdk") + +from opentelemetry import trace as otel_trace # noqa: E402 +from opentelemetry.sdk.trace import TracerProvider # noqa: E402 +from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: E402 +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter # noqa: E402 +from opentelemetry.trace import StatusCode # noqa: E402 + +from xgen_agent_runtime import observability # noqa: E402 +from xgen_agent_runtime.observability import spans as obs_spans # noqa: E402 +from xgen_agent_runtime.core.state import PipelineState # noqa: E402 +from xgen_agent_runtime.host import runner # noqa: E402 +from xgen_agent_runtime.llm_client.base import BaseClient, ClientCapabilities # noqa: E402 +from xgen_agent_runtime.llm_client.types import APIResponse, ContentBlock, TokenUsage # noqa: E402 +from xgen_agent_runtime.tools.base import Tool, ToolCapabilities, ToolContext, ToolResult # noqa: E402 +from xgen_agent_runtime.tools.registry import ToolRegistry # noqa: E402 + +_exporter = InMemorySpanExporter() +_provider = TracerProvider() +_provider.add_span_processor(SimpleSpanProcessor(_exporter)) +_tracer = _provider.get_tracer("test") + + +@pytest.fixture(autouse=True) +def _spans(monkeypatch): + """전역 provider 는 건드리지 않는다 — OTel 은 첫 설정만 받는다.""" + monkeypatch.setattr(obs_spans, "get_tracer", lambda: _tracer) + monkeypatch.delenv(observability.ENV_SWITCH, raising=False) + _exporter.clear() + yield + _exporter.clear() + + +class _Fetch(Tool): + """도구 안에서 바깥 호출을 하는 도구 — 그 호출이 도구 span 아래에 달려야 한다.""" + + def __init__(self, fail: bool = False) -> None: + self.fail = fail + + @property + def name(self) -> str: + return "fetch" + + @property + def description(self) -> str: + return "fetch" + + @property + def input_schema(self) -> Dict[str, Any]: + return {"type": "object"} + + def capabilities(self, input: Dict[str, Any]) -> ToolCapabilities: + return ToolCapabilities(concurrency_safe=True) + + async def execute(self, input: Dict[str, Any], context: ToolContext) -> ToolResult: + with _tracer.start_as_current_span("POST"): # HTTP 자동 계측 span 자리 + pass + if self.fail: + return ToolResult(content="boom", is_error=True) + return ToolResult(content="12345") + + +class _Client(BaseClient): + """1번째 응답: 도구 호출. 2번째: 최종 답. 호출마다 HTTP span 을 하나 낸다.""" + + provider = "fake" + capabilities = ClientCapabilities() + + def __init__(self, **kw: Any) -> None: + super().__init__(**kw) + self.requests: List[Any] = [] + + async def _send(self, request: Any, *, purpose: str = "") -> APIResponse: + self.requests.append(request) + with _tracer.start_as_current_span("POST /v1/chat/completions"): + pass + if len(self.requests) == 1: + return APIResponse( + content=[ContentBlock(type="tool_use", tool_use_id="t1", tool_name="fetch", tool_input={})], + stop_reason="tool_use", + usage=TokenUsage(input_tokens=100, output_tokens=7, cache_read_input_tokens=40), + model="m-snapshot", + ) + return APIResponse( + content=[ContentBlock(type="text", text="done")], + stop_reason="end_turn", + usage=TokenUsage(input_tokens=130, output_tokens=3), + model="m-snapshot", + ) + + +def _run_under_host_turn(tool: _Fetch) -> str: + reg = ToolRegistry() + reg.register(tool, core=True) + pipe = runner.build_pipeline( + name="t", provider="openai", model="m", api_key="k", llm_client=_Client(api_key="k"), + stream=False, enable_compaction=False, registry=reg, + ) + # 호스트가 연 턴 span 을 현재 span 으로 두고 턴을 돈다 — 런타임은 전용 이벤트 루프에서 돌지만 + # 루프가 호출 스레드의 context 를 이어받아야 한다. + with _tracer.start_as_current_span("invoke_agent host"): + return runner.run_turn(pipe, "가져와", PipelineState(session_id="s", model="m")) + + +def _by_name(name: str) -> List[Any]: + return [s for s in _exporter.get_finished_spans() if s.name == name] + + +def _parent_name(span: Any) -> str: + parent_id = span.parent.span_id if span.parent is not None else None + for s in _exporter.get_finished_spans(): + if s.context.span_id == parent_id: + return s.name + return "" + + +def test_each_llm_call_is_a_chat_span_under_the_host_turn_with_its_own_usage() -> None: + assert _run_under_host_turn(_Fetch()) == "done" + + chats = _by_name("chat m") + assert len(chats) == 2 + assert all(_parent_name(s) == "invoke_agent host" for s in chats) + first, second = sorted(chats, key=lambda s: s.start_time) + assert first.attributes["gen_ai.operation.name"] == "chat" + assert first.attributes["gen_ai.provider.name"] == "fake" + assert first.attributes["gen_ai.response.model"] == "m-snapshot" + assert first.attributes["gen_ai.usage.input_tokens"] == 100 + assert first.attributes["gen_ai.usage.output_tokens"] == 7 + assert first.attributes["gen_ai.usage.cache_read.input_tokens"] == 40 + assert tuple(first.attributes["gen_ai.response.finish_reasons"]) == ("tool_use",) + assert first.attributes["xgen.llm.tool_calls"] == 1 + assert second.attributes["gen_ai.usage.input_tokens"] == 130 + assert "gen_ai.usage.cache_read.input_tokens" not in second.attributes + + # 벤더 SDK 의 HTTP span 은 그 호출 span 아래에 달린다. + http = _by_name("POST /v1/chat/completions") + assert len(http) == 2 and all(_parent_name(s) == "chat m" for s in http) + + +def test_tool_run_is_an_execute_tool_span_and_its_calls_nest_under_it() -> None: + _run_under_host_turn(_Fetch()) + + (tool,) = _by_name("execute_tool fetch") + assert _parent_name(tool) == "invoke_agent host" + assert tool.attributes["gen_ai.tool.name"] == "fetch" + assert tool.attributes["xgen.tool.is_error"] is False + assert tool.attributes["xgen.tool.result_size"] == 5 + assert tool.status.status_code != StatusCode.ERROR + (inner,) = _by_name("POST") + assert _parent_name(inner) == "execute_tool fetch" + + +def test_failed_tool_is_an_error_span_without_its_text() -> None: + _run_under_host_turn(_Fetch(fail=True)) + + (tool,) = _by_name("execute_tool fetch") + assert tool.status.status_code == StatusCode.ERROR + assert tool.attributes["xgen.tool.is_error"] is True + assert tool.attributes["error.type"] == "tool_error" + assert "boom" not in str(dict(tool.attributes)) + + +def test_switch_off_emits_nothing(monkeypatch) -> None: + monkeypatch.setenv(observability.ENV_SWITCH, "0") + assert observability.spans_enabled() is False + _run_under_host_turn(_Fetch()) + assert not _by_name("chat m") and not _by_name("execute_tool fetch") + # 런타임 span 이 빠져도 HTTP span 은 그대로 호스트 턴 아래에 남는다. + assert all(_parent_name(s) == "invoke_agent host" for s in _by_name("POST /v1/chat/completions")) + + +def test_broken_tracer_never_breaks_the_turn(monkeypatch) -> None: + def _boom() -> Any: + raise RuntimeError("tracer down") + + monkeypatch.setattr(obs_spans, "get_tracer", _boom) + assert _run_under_host_turn(_Fetch()) == "done" + + +def test_spans_enabled_reflects_the_api_install() -> None: + assert observability.spans_enabled() is (otel_trace is not None) From d8e85dc084d874dc01ce3e812f9227e904a6f87b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=A0=84=EC=9D=B8=EC=88=98?= Date: Thu, 15 Oct 2026 11:44:02 +0900 Subject: [PATCH 2/3] =?UTF-8?q?refactor(observability):=20ruff=20format=20?= =?UTF-8?q?=EC=A0=81=EC=9A=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI 의 `ruff format --check` 를 통과하도록 observability/genai.py 의 긴 `with` 줄과 dict 인자 줄바꿈을 맞춘다. 동작 변경 없음. --- src/xgen_agent_runtime/observability/genai.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/src/xgen_agent_runtime/observability/genai.py b/src/xgen_agent_runtime/observability/genai.py index 3593009..be8a43a 100644 --- a/src/xgen_agent_runtime/observability/genai.py +++ b/src/xgen_agent_runtime/observability/genai.py @@ -94,7 +94,9 @@ def tool_span(*, tool_name: str) -> Iterator[SpanHandle]: attrs.GEN_AI_TOOL_NAME: tool_name, attrs.GEN_AI_TOOL_TYPE: attrs.TOOL_TYPE_FUNCTION, } - with current_span(_span_name(attrs.OP_EXECUTE_TOOL, tool_name), internal_kind(), attributes) as span: + with current_span( + _span_name(attrs.OP_EXECUTE_TOOL, tool_name), internal_kind(), attributes + ) as span: yield span @@ -110,10 +112,12 @@ def record_tool_result(span: SpanHandle, result: Any) -> None: if not span.recording or result is None: return is_error = bool(getattr(result, "is_error", False)) - span.set({ - attrs.XGEN_TOOL_IS_ERROR: is_error, - attrs.XGEN_TOOL_RESULT_SIZE: _result_size(result), - }) + span.set( + { + attrs.XGEN_TOOL_IS_ERROR: is_error, + attrs.XGEN_TOOL_RESULT_SIZE: _result_size(result), + } + ) if is_error: span.fail("tool_error") From 3e31ea5333ca4c40bc1584213ad7b85d14c7c355 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=A0=84=EC=9D=B8=EC=88=98?= Date: Thu, 15 Oct 2026 11:53:22 +0900 Subject: [PATCH 3/3] =?UTF-8?q?test(observability):=20dev=20extra=20?= =?UTF-8?q?=EC=97=90=20opentelemetry-sdk=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test(observability): dev extra 에 opentelemetry-sdk 추가 CI 는 `pip install -e ".[dev]"` 만 하므로 SDK 가 없어 tests/test_otel_spans.py 가 importorskip 으로 통째로 skip 되었다. 테스트 전용으로 dev extra 에만 넣는다. 런타임 의존성은 그대로다. --- pyproject.toml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 3288cdd..bfe0755 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -115,6 +115,10 @@ dev = [ "ruff>=0.4.0", "mypy>=1.10", "asyncssh>=2.14", + # 테스트 전용: observability(LLM 호출·도구 실행 span) 테스트가 span 을 모으는 데 쓴다. + # 런타임은 opentelemetry 를 의존성으로 선언하지 않는다 — 없으면 span 을 내지 않을 뿐이다. + # 이게 없으면 tests/test_otel_spans.py 가 통째로 skip 되어 CI 가 이 경로를 검증하지 못한다. + "opentelemetry-sdk>=1.30", ] [project.urls]