k-agentops 1.1.0

K-AgentOps SDK - Agent observability, tracing, and evaluation metrics

K-AgentOps v1.1.0

AI 에이전트 관측성(Observability) SDK. 데코레이터 한 줄이면 LLM/Tool/Agent 호출이 자동으로 트레이싱됩니다.

OTel GenAI Semantic Convention 기반 — Langfuse, Grafana Tempo, Jaeger 등 모든 OTel 호환 백엔드에서 시각화 가능.

---

목차

1. 설치
2. 빠른 시작
3. init() API
4. 데코레이터 7종
5. 자동 계측
6. Langfuse 연동
7. 컨텍스트 전파
8. 수동 관측 기록
9. 환경변수
10. 설정 파라미터
11. 아키텍처
12. 폴더 구조
13. FAQ

---

1. 설치

pip install k-agentops

의존성 3개만 설치됩니다: opentelemetry-api, opentelemetry-sdk, opentelemetry-exporter-otlp-proto-grpc

프로바이더/프레임워크 extras

# 특정 프로바이더
pip install k-agentops[openai]
pip install k-agentops[anthropic]
pip install k-agentops[google]

# 프레임워크
pip install k-agentops[langgraph]
pip install k-agentops[langchain]
pip install k-agentops[crewai]

# 전체
pip install k-agentops[all]

---

2. 빠른 시작

from k_agentops import KAgentOps, agent, tool, generation

# 1) 초기화
client = KAgentOps(
    service_name="my-agent",
    otlp_endpoint="http://localhost:4317",
)

# 2) 데코레이터만 붙이면 끝
@agent
def my_agent(query: str) -> str:
    docs = search(query)
    return summarize(query, docs)

@tool
def search(query: str) -> list[str]:
    return ["문서1", "문서2"]

@generation
def summarize(query: str, docs: list[str]) -> str:
    return "요약 결과"

# 3) 실행
result = my_agent("최근 AI 뉴스 알려줘")

# 4) 종료 (버퍼 플러시)
client.shutdown()

트레이스 워터폴:

my_agent (agent)           ████████████████████  350ms
├── search (tool)          ██████                 100ms
└── summarize (generation) ████████████           250ms

---

3. init() API

init()은 글로벌 싱글톤 클라이언트를 초기화합니다. start_trace()/end_trace()와 함께 사용하면 Langfuse에 트레이스가 깔끔하게 표시됩니다.

import k_agentops

client = k_agentops.init(
    service_name="my-agent",
    auto_instrument=True,               # LLM 프로바이더 자동 패치
    instrument_frameworks=True,          # 프레임워크 자동 패치
    environment="production",            # Langfuse Environment 필드
    tags=["rag", "v2"],                  # Langfuse Tags 필드
    user_id="user-123",                  # Langfuse User ID
    session_id="sess-456",              # Langfuse Session ID
    metadata={"version": "1.1"},        # Langfuse Metadata
)

# 트레이스 시작 (Langfuse 트레이스 단위)
ctx = k_agentops.start_trace("my-pipeline", tags=["production"])

# ... @agent, @tool, @generation 데코레이터가 붙은 함수 실행 ...
result = my_agent("질문")

# 트레이스 종료
ctx.end(output=result)  # 또는 k_agentops.end_trace(ctx, output=result)

client.shutdown()

Context Manager 패턴

with k_agentops.start_trace("my-pipeline") as ctx:
    result = my_agent("질문")
    ctx.set_output(result)
# 블록 종료 시 자동으로 end_trace() 호출

자동 Input/Output

start_trace() 안에서 @instrument 데코레이터 함수가 호출되면, 첫 번째 함수의 입력이 Trace Input으로, 마지막 함수의 출력이 Trace Output으로 Langfuse에 자동 표시됩니다. 명시적으로 ctx.set_input()/ctx.set_output()을 호출할 필요가 없습니다.

---

4. 데코레이터 7종

데코레이터	용도	OTel operation
@agent	에이전트 (최상위 실행 단위)	invoke_agent
@tool	도구 호출 (검색, API, DB)	execute_tool
@generation	LLM 호출	chat
@task	작업 단위	execute_task
@workflow	워크플로우/파이프라인	execute_workflow
@guardrail	입출력 검증	guardrail_check
@retriever	RAG 검색기	retrieve

# 커스텀 이름
@agent(name="customer-support")
def support(query): ...

# async 함수
@generation
async def call_llm(prompt: str) -> str:
    return await llm.generate(prompt)

# 범용 데코레이터
from k_agentops import instrument

@instrument(as_type="agent", name="custom-name")
def custom_agent(query): ...

지원 함수 타입

• 동기 함수
• 비동기 함수 (async def)
• 동기 제너레이터 (yield)
• 비동기 제너레이터 (async yield)
• 클래스 (__call__ 메서드)

---

5. 자동 계측 (코드 수정 0)

LLM 프로바이더 자동 패치

client = KAgentOps(auto_instrument=True)

# 이후 모든 LLM 호출이 자동으로 트레이싱됨
import openai
response = openai.chat.completions.create(...)  # 자동 span 생성

프로바이더	패키지	추출 정보
OpenAI	openai>=1.0	모델, 토큰, finish_reason, cached_tokens
Anthropic	anthropic>=0.32	모델, 토큰, cache_creation/read_tokens
Google GenAI	google-genai>=1.0	모델, 토큰, finish_reason
Cohere	cohere>=5.0	모델, 토큰
LiteLLM	litellm>=1.0	100+ 모델 프록시
Groq	groq>=0.5	모델, 토큰
Together AI	together>=1.0	모델, 토큰

프레임워크 자동 패치

client = KAgentOps(
    auto_instrument=True,
    instrument_frameworks=True,
)

프레임워크	패치 대상
LangGraph	CompiledGraph.invoke/ainvoke/stream/astream
LangChain	BaseTool.invoke/ainvoke + chain 실행
CrewAI	Task.execute_sync/execute_async, Crew.kickoff
AutoGen/AG2	ConversableAgent.initiate_chat/a_initiate_chat
OpenAI Agents SDK	Runner.run/run_sync/run_streamed
Google ADK	Runner.run_async
LlamaIndex	QueryEngine/ChatEngine
Haystack	Pipeline.run
Agno	Agent.run/arun
Smolagents	CodeAgent/ToolCallingAgent.run

특정 프로바이더만 선택

client = KAgentOps(
    auto_instrument=True,
    instrument_providers=["openai", "anthropic"],   # LLM 선택
    instrument_framework_list=["langgraph"],         # 프레임워크 선택
)

---

6. Langfuse 연동

K-AgentOps는 OTel 표준 속성 + Langfuse 전용 속성을 함께 설정하여 Langfuse UI에 최적화된 표시를 제공합니다.

Langfuse 트레이스 목록 컬럼 매핑

Langfuse 필드	SDK 파라미터	OTel 속성
Name	start_trace(trace_name)	span name
Input	자동 (첫 번째 함수 입력)	langfuse.trace.input
Output	자동 (마지막 함수 출력)	langfuse.trace.output
Tags	init(tags=[...])	langfuse.trace.tags
User ID	init(user_id="...")	langfuse.trace.user.id
Session ID	init(session_id="...")	langfuse.trace.session.id
Metadata	init(metadata={...})	langfuse.trace.metadata
Environment	init(environment="...")	deployment.environment

Observation (스팬) 레벨 속성

각 @instrument 데코레이터 스팬에는 다음 속성이 자동 설정됩니다:

속성	설명
gen_ai.input.messages	입력 (GenAI 표준)
gen_ai.output.messages	출력 (GenAI 표준)
input.value	Langfuse observation Input
output.value	Langfuse observation Output
gen_ai.operation.name	오퍼레이션 타입
gen_ai.request.model	사용 모델
gen_ai.usage.input_tokens	입력 토큰
gen_ai.usage.output_tokens	출력 토큰

Langfuse OTel 연결 예시

# Langfuse Cloud (SaaS)
export OTEL_EXPORTER_OTLP_ENDPOINT=https://cloud.langfuse.com/api/public/otel
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer pk-lf-..."

# Langfuse Self-hosted
export OTEL_EXPORTER_OTLP_ENDPOINT=http://your-langfuse:4318

---

7. 컨텍스트 전파

trace_context

OTel Baggage를 사용하여 식별자와 메타데이터를 모든 자식 스팬에 전파합니다.

from k_agentops import trace_context

with trace_context(
    user_id="user-123",
    session_id="sess-456",
    tenant_id="org-789",
    metadata={"channel": "web"},
    tags=["production"],
):
    result = my_agent("질문")
    # 이 블록 안에서 생성된 모든 스팬에 위 속성이 전파됨

현재 트레이스/스팬 ID 조회

from k_agentops import get_current_trace_id, get_current_observation_id

@agent
def my_agent(query: str) -> str:
    trace_id = get_current_trace_id()      # "abc123..."
    span_id = get_current_observation_id()  # "def456..."
    return "결과"

---

8. 수동 관측 기록

LLM 토큰/모델 기록

from k_agentops import generation, update_current_observation

@generation
def call_llm(prompt: str) -> str:
    response = openai.chat.completions.create(model="gpt-4o", messages=[...])

    update_current_observation(
        model="gpt-4o",
        provider="openai",
        usage_input_tokens=response.usage.prompt_tokens,
        usage_output_tokens=response.usage.completion_tokens,
    )

    return response.choices[0].message.content

Input/Output 오버라이드

from k_agentops import set_observation_io

@retriever
def search(query: str) -> list[str]:
    docs = vector_db.search(query)

    set_observation_io(
        input_text=query,
        output_text=str(docs),
        retrieved_docs=[doc.text for doc in docs],  # RAG 평가용
    )

    return docs

---

9. 환경변수

환경변수	설명	기본값
AGENTOPS_ENABLED	SDK 활성화	true
AGENTOPS_CAPTURE_CONTENT	프롬프트/응답 캡처	false
AGENTOPS_CONTENT_MAX_LENGTH	캡처 최대 문자 수	10000
OTEL_SERVICE_NAME	서비스 이름	agent-ops-sdk
OTEL_EXPORTER_OTLP_ENDPOINT	OTel Collector gRPC 주소	http://localhost:4317
OTEL_EXPORTER_OTLP_INSECURE	TLS 비활성화	true
OTEL_EXPORTER_OTLP_HEADERS	OTLP 추가 헤더	(없음)
OTEL_BSP_MAX_EXPORT_BATCH_SIZE	배치 크기	512
OTEL_BSP_MAX_QUEUE_SIZE	큐 크기	2048
OTEL_BSP_SCHEDULE_DELAY	전송 주기 (ms)	5000

환경변수만 설정하면 인자 없이 사용 가능:

client = KAgentOps()

---

10. 설정 파라미터

KAgentOpsConfig로 모든 값을 코드에서 제어할 수 있습니다. 우선순위: 코드 > 환경변수 > 기본값.

from k_agentops import KAgentOps
from k_agentops.types import KAgentOpsConfig

client = KAgentOps(
    config=KAgentOpsConfig(
        service_name="my-org.agent",
        otlp_endpoint="http://otel-collector:4317",
        otlp_headers=(("x-tenant-id", "my-org"),),
        capture_content=True,           # PII 주의
        content_max_length=5000,
    ),
    environment="production",
    tenant_id="my-org",
    tags=["prod", "customer-support"],
)

파라미터	타입	기본값	설명
service_name	str	"agent-ops-sdk"	OTel 서비스 식별자
enabled	bool	True	SDK 활성화
debug	bool	False	디버그 로깅
otlp_endpoint	str	"http://localhost:4317"	Collector gRPC 주소
otlp_insecure	bool	True	TLS 비활성화
otlp_headers	tuple	()	OTLP 추가 헤더
capture_content	bool	False	프롬프트/응답 캡처
content_max_length	int	10000	캡처 최대 문자 수
max_export_batch_size	int	512	배치 크기
max_queue_size	int	2048	큐 크기
schedule_delay_millis	int	5000	전송 주기 (ms)

---

11. 아키텍처

데이터 흐름

사용자 코드 (@agent, @tool, @generation, 자동 계측)
    │
    ▼
K-AgentOps SDK (OTel TracerProvider)
    │ gRPC (:4317)
    ▼
OTel Collector
    ├──► Langfuse     트레이스 (에이전트 워터폴 + Input/Output)
    ├──► Tempo        트레이스 (분산 추적)
    ├──► Prometheus   메트릭 (토큰, 레이턴시, 에러율)
    └──► Loki         로그
           │
           ▼
        Grafana (시각화/대시보드)

OTel 속성 체계

GenAI Semantic Convention (OTel 표준)
├── gen_ai.system                    # LLM 프로바이더
├── gen_ai.operation.name            # 오퍼레이션 타입
├── gen_ai.request.model             # 요청 모델
├── gen_ai.usage.input_tokens        # 입력 토큰
├── gen_ai.usage.output_tokens       # 출력 토큰
├── gen_ai.input.messages            # 입력 메시지
├── gen_ai.output.messages           # 출력 메시지
├── gen_ai.response.finish_reason    # 종료 사유
│
Langfuse 매핑 속성
├── input.value / output.value       # Observation Input/Output
├── langfuse.trace.input/output      # Trace 레벨 Input/Output
├── langfuse.trace.tags              # Tags
├── langfuse.trace.user.id           # User ID
├── langfuse.trace.session.id        # Session ID
├── langfuse.trace.metadata          # Metadata
└── deployment.environment           # Environment

---

12. 폴더 구조

k-agentops/
├── k_agentops/
│   ├── __init__.py                         # 공개 API (21개 심볼)
│   ├── client.py                           # KAgentOps 메인 클라이언트
│   ├── types.py                            # Enum, 설정 데이터클래스
│   ├── api/
│   │   └── compat.py                       # init(), start_trace(), end_trace()
│   ├── tracing/
│   │   ├── decorator.py                    # @instrument 구현
│   │   ├── shortcuts.py                    # @agent, @tool 등 축약형
│   │   ├── context.py                      # trace_context, baggage, Langfuse 매핑
│   │   ├── llm_extractors.py              # LLM 응답 자동 추출
│   │   └── _traced_streams.py             # 스트림 래퍼 (sync/async)
│   ├── patchers/
│   │   ├── registry.py                     # LLM 프로바이더 중앙 레지스트리
│   │   ├── llm/                            # 프로바이더별 패처
│   │   │   ├── base.py                    #   공통 래퍼 팩토리
│   │   │   ├── openai.py                  #   OpenAI
│   │   │   ├── anthropic.py               #   Anthropic
│   │   │   ├── google_genai.py            #   Google GenAI
│   │   │   └── ...                        #   Cohere, LiteLLM, Groq, Together
│   │   └── frameworks/                     # 프레임워크별 패처
│   │       ├── base.py                    #   공통 래퍼 팩토리
│   │       ├── registry.py                #   프레임워크 중앙 레지스트리
│   │       ├── langgraph.py               #   LangGraph
│   │       ├── crewai.py                  #   CrewAI
│   │       └── ...                        #   AutoGen, OpenAI Agents, etc.
│   ├── signals/                            # OTel 신호 provider
│   │   ├── traces.py                       #   TracerProvider
│   │   ├── metrics.py                      #   MeterProvider
│   │   └── logs.py                         #   LoggerProvider
│   ├── runtime/
│   │   └── config.py                       # 환경변수 로딩
│   ├── telemetry/                          # OTel 메트릭 계측기
│   ├── integration/                        # 프레임워크 콜백 어댑터
│   │   ├── callbacks/
│   │   │   ├── langchain.py               #   LangChainCallbackHandler
│   │   │   └── dspy.py                    #   DSPyCallbackHandler
│   │   └── dify.py                        #   DifyClient
│   └── validation/                         # 트레이스 검증 도구
├── tests/                                  # 단위/통합/E2E 테스트 (293+)
├── docs/                                   # 가이드, Azure 배포 문서
├── pyproject.toml                          # 빌드 설정 (hatchling)
├── Makefile                                # test, lint, typecheck, audit
├── LICENSE                                 # MIT
└── README.md

---

FAQ

Q: SDK 에러가 내 앱을 중단시키나요? A: 아닙니다. 모든 SDK 에러는 내부에서 catch됩니다. OTel Collector가 죽어도 앱은 정상 동작합니다.

Q: client.shutdown() 안 하면? A: 버퍼에 남은 트레이스가 전송되지 않습니다. 반드시 호출하거나 with KAgentOps() as client: 패턴을 사용하세요.

Q: 토큰이 자동으로 안 잡혀요 A: auto_instrument=True 설정하거나, @generation 안에서 update_current_observation()으로 수동 기록하세요.

Q: Langfuse에 Input/Output이 안 보여요 A: v1.1.0부터 @instrument 데코레이터와 자동 계측 모두 langfuse.trace.input/output을 자동 설정합니다. start_trace() 안에서 함수를 실행하면 자동 채워집니다.

Q: capture_content는 뭔가요? A: True로 설정하면 LLM 프롬프트/응답 원문이 스팬에 기록됩니다. PII(개인정보) 노출 위험이 있으므로 개발 환경에서만 사용을 권장합니다.

Q: 여러 KAgentOps 인스턴스를 동시에 쓸 수 있나요? A: 네. 내부적으로 ref-counted TracerProvider를 사용하여 마지막 인스턴스가 shutdown될 때만 정리됩니다.

---

CHANGELOG

v1.1.0 (2026-04-03)

• OTel GenAI Semantic Convention 전면 적용 (gen_ai.input.messages, gen_ai.output.messages, gen_ai.content.prompt/completion 이벤트)
• Langfuse 트레이스 목록 Input/Output 자동 채우기 (langfuse.trace.input/output)
• environment 파라미터 추가 (deployment.environment)
• tags, user_id, session_id, metadata → Langfuse 속성 자동 매핑
• 프레임워크 자동 계측 Input/Output 정제 (config noise 제거, args[0] 캡처)
• 스트림 래퍼 output 자동 추출

v1.0.0 (2026-04-02)

• 패키지명 k-agentops 확정
• 하드코딩 endpoint 제거
• 전체 리네임 완료

v0.9.0 (2026-04-01)

• 아키텍처 전면 리팩토링
• Codex adversarial 보안 수정
• ref-counted TracerProvider