컨텍스트 매니저

컨텍스트 매니저 (Context Managers)

AgentOps 트레이스를 Python 컨텍스트 매니저로 사용해 자동으로 수명주기를 관리하는 방법을 소개해요. 기본 사용법부터 고급 패턴, 하위 호환성까지 알아볼게요.

출처: 문서

본문

컨텍스트 매니저

AgentOps는 트레이스에 대한 네이티브 컨텍스트 매니저 지원을 제공하여, Python의 with 문으로 자동 트레이스 수명주기 관리를 할 수 있게 해줍니다. 이 방식은 예외가 발생해도 트레이스가 제대로 시작되고 종료되도록 보장합니다.

기본 사용법 (Basic Usage)

컨텍스트 매니저를 사용하는 가장 간단한 방법은 start_trace() 함수를 사용하는 것입니다.

import agentops

# Initialize AgentOps
agentops.init(api_key="your-api-key")

# Use context manager for automatic trace management
with agentops.start_trace("my_workflow") as trace:
    # Your code here
    print("Processing data...")
    # Trace automatically ends when exiting the with block

트레이스는 자동으로:

with 블록에 진입할 때 시작
정상 종료 시 "Success" 상태로 종료
예외 발생 시 "Error" 상태로 종료
모든 경우에 리소스를 적절히 정리

고급 사용법 (Advanced Usage)

태그가 있는 트레이스 (Traces with Tags)

더 나은 조직화와 필터링을 위해 트레이스에 태그를 추가할 수 있어요.

import agentops

agentops.init(api_key="your-api-key")

# Using list tags
with agentops.start_trace("data_processing", tags=["batch", "production"]):
    process_batch_data()

# Using dictionary tags for more structured metadata
with agentops.start_trace("user_request", tags={
    "user_id": "12345",
    "request_type": "query",
    "priority": "high"
}):
    handle_user_request()

병렬 트레이스 (Parallel Traces)

컨텍스트 매니저는 부모-자식 관계가 아닌 독립적인 병렬 트레이스를 만듭니다.

import agentops

agentops.init(api_key="your-api-key")

# Sequential parallel traces
with agentops.start_trace("task_1"):
    print("Task 1 executing")

with agentops.start_trace("task_2"):
    print("Task 2 executing")

# Nested context managers create parallel traces
with agentops.start_trace("outer_workflow"):
    print("Outer workflow started")
    
    with agentops.start_trace("inner_task"):
        print("Inner task executing (parallel to outer)")
    
    print("Outer workflow continuing")

예외 처리 (Exception Handling)

컨텍스트 매니저는 예외를 자동으로 처리하고 적절한 트레이스 상태를 설정합니다.

import agentops

agentops.init(api_key="your-api-key")

# Automatic error handling
try:
    with agentops.start_trace("risky_operation"):
        # This will automatically set trace status to "Error"
        raise ValueError("Something went wrong")
except ValueError as e:
    print(f"Caught error: {e}")
    # Trace has already been ended with Error status

# Graceful degradation pattern
try:
    with agentops.start_trace("primary_service"):
        result = call_primary_service()
except ServiceUnavailableError:
    with agentops.start_trace("fallback_service"):
        result = call_fallback_service()

동시 실행 (Concurrent Execution)

컨텍스트 매니저는 threading과 asyncio와 완벽하게 동작합니다.

```python Threading theme={null} import agentops import threading

agentops.init(api_key="your-api-key")

With threading

def worker_function(worker_id): with agentops.start_trace(f"worker_{worker_id}"): # Each thread gets its own independent trace process_work(worker_id)

threads = [] for i in range(3): thread = threading.Thread(target=worker_function, args=(i,)) threads.append(thread) thread.start()

for thread in threads: thread.join()


```python Asyncio theme={null}
import agentops
import asyncio

agentops.init(api_key="your-api-key")

# With asyncio
async def async_task(task_id):
    with agentops.start_trace(f"async_task_{task_id}"):
        await asyncio.sleep(0.1)  # Simulate async work
        return f"result_{task_id}"

async def main():
    tasks = [async_task(i) for i in range(3)]
    results = await asyncio.gather(*tasks)
    return results

# Run async tasks
results = asyncio.run(main())

프로덕션 패턴 (Production Patterns)

API 엔드포인트 모니터링 (API Endpoint Monitoring)

import agentops
from flask import Flask, request

app = Flask(__name__)
agentops.init(api_key="your-api-key")

@app.route('/api/process', methods=['POST'])
def process_request():
    # Create trace for each API request
    with agentops.start_trace("api_request", tags={
        "endpoint": "/api/process",
        "method": "POST",
        "user_id": request.headers.get("user-id")
    }):
        try:
            data = request.get_json()
            result = process_data(data)
            return {"status": "success", "result": result}
        except Exception as e:
            # Exception automatically sets trace to Error status
            return {"status": "error", "message": str(e)}, 500

배치 처리 (Batch Processing)

import agentops

agentops.init(api_key="your-api-key")

def process_batch(items):
    with agentops.start_trace("batch_processing", tags={
        "batch_size": len(items),
        "batch_type": "data_processing"
    }):
        successful = 0
        failed = 0
        
        for item in items:
            try:
                with agentops.start_trace("item_processing", tags={
                    "item_id": item.get("id"),
                    "item_type": item.get("type")
                }):
                    process_item(item)
                    successful += 1
            except Exception as e:
                failed += 1
                print(f"Failed to process item {item.get('id')}: {e}")
        
        print(f"Batch completed: {successful} successful, {failed} failed")

재시도 로직 (Retry Logic)

import agentops
import time

agentops.init(api_key="your-api-key")

def retry_operation(operation_name, max_retries=3):
    for attempt in range(max_retries):
        try:
            with agentops.start_trace(f"{operation_name}_attempt_{attempt + 1}", tags={
                "operation": operation_name,
                "attempt": attempt + 1,
                "max_retries": max_retries
            }):
                # Your operation here
                result = perform_operation()
                return result  # Success - exit retry loop
                
        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                print(f"All {max_retries} attempts failed")
                raise

하위 호환성 (Backward Compatibility)

컨텍스트 매니저는 기존 AgentOps 코드 패턴과 완전히 하위 호환됩니다.

```python Manual Management theme={null} import agentops

agentops.init(api_key="your-api-key")

Manual trace management (legacy)

trace = agentops.start_trace("manual_trace")

... your code ...

agentops.end_trace(trace, "Success")


```python Context Manager theme={null}
import agentops

agentops.init(api_key="your-api-key")

# Context manager (new, recommended)
with agentops.start_trace("context_managed_trace") as trace:
    # ... your code ...
    pass  # Automatically ended

import agentops

agentops.init(api_key="your-api-key")

# Accessing trace properties
with agentops.start_trace("property_access") as trace:
    span = trace.span  # Access underlying span
    trace_id = trace.span.get_span_context().trace_id

import agentops

agentops.init(api_key="your-api-key")

# Mixed usage
trace = agentops.start_trace("mixed_usage")
try:
    with trace:  # Use existing trace as context manager
        # ... your code ...
        pass
except Exception:
    agentops.end_trace(trace, "Error")

예제 (Examples)

완전한 동작 예제는 AgentOps 저장소의 다음 파일에서 볼 수 있습니다.

간단한 컨텍스트 매니저 패턴과 오류 처리 순차, 중첩, 동시 트레이스 패턴 예외 처리, 재시도 패턴, 그리고 우아한 성능 저하 API 엔드포인트, 배치 처리, 마이크로서비스, 모니터링

이 예제들은 프로덕션 애플리케이션에서 AgentOps 컨텍스트 매니저를 사용하는 실제 사용 패턴과 모범 사례를 보여줍니다.

API 레퍼런스

자세한 API 정보는 SDK 레퍼런스 문서를 참조하세요.

더 알아보기 (Learn more)

AgentOps 공식 문서