Xpander

AgentOps의 자동 계측으로 Xpander 에이전트 워크플로우를 모니터링하고 분석하는 방법을 소개해요. 설치, API 키 설정, 그리고 초기화 순서가 중요한 통합 예제까지 알아볼게요.

출처: 문서

본문

Xpander는 정교한 워크플로우 관리 기능을 갖춘 AI 에이전트를 구축하고 배포하기 위한 강력한 플랫폼이에요. AgentOps는 Xpander SDK와 원활하게 통합되어, 수동 설정 없이 모든 에이전트 활동, 도구 실행, LLM 상호작용을 자동으로 계측합니다.

설치 (Installation)

AgentOps와 Xpander SDK를 필요한 의존성과 함께 설치하세요.

```bash pip theme={null} pip install agentops xpander-sdk xpander-utils openai python-dotenv loguru ```

poetry add agentops xpander-sdk xpander-utils openai python-dotenv loguru

uv add agentops xpander-sdk xpander-utils openai python-dotenv loguru

API 키 설정 (Setting Up API Keys)

AgentOps, Xpander, 그리고 OpenAI에 대한 API 키가 필요해요.

AGENTOPS_API_KEY: AgentOps Dashboard에서
XPANDER_API_KEY: Xpander Dashboard에서
XPANDER_AGENT_ID: Xpander 에이전트의 ID
OPENAI_API_KEY: OpenAI Platform에서

이것들을 환경 변수나 .env 파일로 설정하세요.

```bash Export to CLI theme={null} export AGENTOPS_API_KEY="your_agentops_api_key_here" export XPANDER_API_KEY="your_xpander_api_key_here" export XPANDER_AGENT_ID="your_xpander_agent_id_here" export OPENAI_API_KEY="your_openai_api_key_here" ```

AGENTOPS_API_KEY="your_agentops_api_key_here"
XPANDER_API_KEY="your_xpander_api_key_here"
XPANDER_AGENT_ID="your_xpander_agent_id_here"
OPENAI_API_KEY="your_openai_api_key_here"

설정을 xpander_config.json 파일에 저장할 수도 있습니다.

{
  "api_key": "your_xpander_api_key_here",
  "agent_id": "your_xpander_agent_id_here"
}

퀵 스타트 (Quick Start)

AgentOps + Xpander 통합의 핵심은 초기화 순서예요. 자동 계측을 활성화하려면 Xpander SDK를 import 하기 전에 AgentOps를 초기화하세요.

The following example shows the callback-based integration pattern. For a complete working example, see our [Xpander example](/v2/examples/xpander).

# ruff: noqa: E402
import os
import json
import asyncio
from pathlib import Path
from dotenv import load_dotenv

# Load environment variables first
load_dotenv()

# 1. Initialize AgentOps FIRST (this enables auto-instrumentation)
import agentops
agentops.init(
    api_key=os.getenv("AGENTOPS_API_KEY"),
    trace_name="my-xpander-coding-agent-callbacks",
    default_tags=["xpander", "coding-agent", "callbacks"],
)

# 2. Now import Xpander SDK (instrumentation will automatically activate)
from xpander_sdk import XpanderClient, LLMProvider, LLMTokens, Tokens, Agent, ExecutionStatus
from xpander_utils.events import XpanderEventListener, AgentExecutionResult, AgentExecution
from openai import AsyncOpenAI

class MyAgent:
    def __init__(self):
        # Load config
        config_path = Path(__file__).parent / "xpander_config.json"
        config = json.loads(config_path.read_text())
        
        # Get API keys
        xpander_key = config.get("api_key") or os.getenv("XPANDER_API_KEY")
        agent_id = config.get("agent_id") or os.getenv("XPANDER_AGENT_ID")
        openai_key = os.getenv("OPENAI_API_KEY")
        
        # Initialize clients
        self.openai = AsyncOpenAI(api_key=openai_key)
        xpander_client = XpanderClient(api_key=xpander_key)
        self.agent_backend: Agent = xpander_client.agents.get(agent_id=agent_id)
        self.agent_backend.select_llm_provider(LLMProvider.OPEN_AI)

    async def run(self, user_input: str) -> dict:
        tokens = Tokens(worker=LLMTokens(0, 0, 0))
        
        while not self.agent_backend.is_finished():
            # Call LLM
            response = await self.openai.chat.completions.create(
                model="gpt-4",
                messages=self.agent_backend.messages,
                tools=self.agent_backend.get_tools(),
                tool_choice=self.agent_backend.tool_choice,
                temperature=0,
            )
            
            # Track tokens
            if hasattr(response, "usage"):
                tokens.worker.prompt_tokens += response.usage.prompt_tokens
                tokens.worker.completion_tokens += response.usage.completion_tokens
                tokens.worker.total_tokens += response.usage.total_tokens
            
            # Add response to agent context
            self.agent_backend.add_messages(response.model_dump())
            self.agent_backend.report_execution_metrics(llm_tokens=tokens, ai_model="gpt-4")
            
            # Execute any tool calls
            tool_calls = self.agent_backend.extract_tool_calls(response.model_dump())
            if tool_calls:
                tool_results = await asyncio.to_thread(self.agent_backend.run_tools, tool_calls)
        
        result = self.agent_backend.retrieve_execution_result()
        return {"result": result.result, "thread_id": result.memory_thread_id}

# Set up event listener with callback handlers
listener = XpanderEventListener(
    api_key=os.getenv("XPANDER_API_KEY"),
    agent_id=os.getenv("XPANDER_AGENT_ID")
)

async def on_execution_request(execution_task: AgentExecution) -> AgentExecutionResult:
    agent = MyAgent()
    agent.agent_backend.init_task(execution=execution_task.model_dump())
    
    try:
        await agent.run(execution_task.input.text)
        execution_result = agent.agent_backend.retrieve_execution_result()
        return AgentExecutionResult(
            result=execution_result.result,
            is_success=execution_result.status == ExecutionStatus.COMPLETED,
        )
    except Exception as e:
        print(f"Error: {e}")
        raise

# Register the callback
listener.register(on_execution_request=on_execution_request)

자동으로 추적되는 것 (What's Automatically Tracked)

AgentOps는 Xpander 에이전트에서 포괄적인 텔레메트리를 자동으로 캡처합니다.

🤖 에이전트 활동 (Agent Activities)

에이전트 초기화와 설정
태스크 수명주기 (시작, 실행 단계, 완료)
워크플로우 단계 전환 (planning → executing → finished)
세션 관리와 컨텍스트 지속

🧠 LLM 상호작용 (LLM Interactions)

전체 요청/응답 데이터가 포함된 모든 OpenAI API 호출
모델 전반의 토큰 사용량과 비용 추적
대화 기록과 컨텍스트 관리
모델 파라미터와 설정

🛠️ 도구 실행 (Tool Executions)

파라미터와 인자가 포함된 도구 호출 감지
도구 실행 결과와 성공/실패 상태
도구 성능 메트릭과 타이밍
도구 호출 계층 구조와 의존성

📊 성능 메트릭 (Performance Metrics)

종단 간 실행 기간과 타이밍
단계별 워크플로우 진행
리소스 활용과 효율성 메트릭
오류 처리와 예외 추적

핵심 기능 (Key Features)

✅ 제로 설정 (Zero-Configuration Setup)

수동 트레이스 생성이나 스팬 관리가 필요 없어요. Xpander SDK를 import 하기 전에 AgentOps를 초기화하기만 하면 됩니다.

✅ 완전한 워크플로우 가시성 (Complete Workflow Visibility)

태스크 시작부터 완료까지 모든 중간 단계를 포함한 전체 에이전트 실행 흐름을 추적합니다.

✅ 실시간 모니터링 (Real-time Monitoring)

에이전트 활동이 실행되는 대로 AgentOps 대시보드에서 실시간으로 볼 수 있어요.

✅ 도구 실행 인사이트 (Tool Execution Insights)

어떤 도구가 호출되는지, 그 파라미터, 실행 시간, 결과를 모니터링합니다.

✅ 비용 추적 (Cost Tracking)

비용 분석과 함께 모든 LLM 상호작용의 자동 토큰 사용량 추적.

콜백 핸들러 패턴 (Callback Handler Pattern)

Xpander 통합은 두 가지 주요 패턴을 지원합니다.

Direct Integration: 에이전트 코드를 직접 계측 (위에서 보여줌)
Callback Handler: 웹훅 스타일 통합을 위해 XpanderEventListener 사용

콜백 핸들러 패턴은 다음에 특히 유용합니다.

중앙 집중식 모니터링이 있는 프로덕션 배포
멀티 에이전트 오케스트레이션 시스템
이벤트 기반 아키텍처

런타임 특화 계측 (Runtime-Specific Instrumentation)

Xpander SDK는 JSII를 사용해 런타임에 메서드를 생성하므로 특화된 계측이 필요합니다. AgentOps는 다음을 통해 이를 자동으로 처리합니다.

Method Wrapping: 생성되는 대로 에이전트 메서드를 동적으로 래핑
Context Persistence: 런타임 객체 수명주기 전반에 걸쳐 세션 컨텍스트 유지
Agent Detection: 새 에이전트 인스턴스를 자동으로 감지하고 계측
Tool Result Extraction: JSII 객체 참조에서 결과를 올바르게 추출

문제 해결 (Troubleshooting)

import 순서 문제 (Import Order Issues)

트레이스가 보이지 않는다면 Xpander SDK를 import 하기 전에 AgentOps가 초기화되었는지 확인하세요.

# ✅ Correct order
import agentops
agentops.init()
from xpander_sdk import XpanderClient

# ❌ Incorrect order
from xpander_sdk import XpanderClient
import agentops
agentops.init()  # Too late - instrumentation won't activate

도구 결과 누락 (Missing Tool Results)

도구 결과가 실제 콘텐츠 대신 {"__jsii_ref__": "..."}로 표시된다면, 개선된 JSII 객체 처리가 포함된 최신 버전의 AgentOps를 사용하고 있는지 확인하세요.

import 오류 (E402) (Import Errors (E402))

import가 파일 상단에 없다는 린트 오류가 보인다면, 이는 Xpander 통합에서 기대되는 동작입니다. 파일 상단에 # ruff: noqa: E402를 추가해 경고를 억제하세요. 적절한 계측을 위해 이 import 순서가 필요하기 때문입니다.

예제 (Examples)

콜백 핸들러를 사용한 완전한 단일 파일 구현 완전한 소스 코드와 설정 파일 보기

더 알아보기 (Learn more)

AgentOps 공식 문서