Tracing과 관측성
Tracing과 관측성 (Monitoring)
AutoGen에는 애플리케이션 실행에 대한 종합적인 기록을 수집하는 추적(tracing)·관측성(observability)이 내장돼 있어요. 디버깅·성능 분석·애플리케이션 흐름 이해에 유용해요.
이 기능은 OpenTelemetry 라이브러리로 구동돼요. 즉 어떤 OpenTelemetry 호환 백엔드로든 트레이스를 수집·분석할 수 있어요.
AutoGen은 에이전트와 도구에 대해 OpenTelemetry Semantic Conventions을 따르고, 현재 개발 중인 Semantic Conventions for GenAI Systems도 따르고 있어요.
설정 (Setup)
먼저 OpenTelemetry Python 패키지를 설치해야 해요. pip로 설치할 수 있어요.
pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc opentelemetry-instrumentation-openai
SDK를 설치한 뒤 AutoGen에서 추적을 설정하는 가장 간단한 방법은 다음과 같아요.
- OpenTelemetry tracer provider 구성
- 트레이스를 백엔드로 보낼 exporter 설정
- tracer provider를 AutoGen 런타임에 연결
텔레메트리 백엔드 (Telemetry Backend)
트레이스를 수집·확인하려면 텔레메트리 백엔드를 설정해야 해요. Jaeger, Zipkin 같은 오픈소스 옵션이 여러 가지 있어요. 이 예시에선 Jaeger를 백엔드로 쓸게요.
빠르게 시작하려면 Docker로 Jaeger를 로컬에서 실행할 수 있어요.
docker run -d --name jaeger \
-e COLLECTOR_OTLP_ENABLED=true \
-p 16686:16686 \
-p 4317:4317 \
-p 4318:4318 \
jaegertracing/all-in-one:latest
이 명령은 Jaeger UI용 포트 16686, OpenTelemetry 컬렉터용 포트 4317에서 수신 대기하는 Jaeger 인스턴스를 시작해요. Jaeger UI는 http://localhost:16686에서 접근할 수 있어요.
AgentChat 팀 추적 (Tracing an AgentChat Team)
다음 섹션에서 AutoGen GroupChat 팀으로 추적을 활성화하는 방법을 살펴볼게요. AutoGen 런타임은 이미 OpenTelemetry를 지원해요(메시지 메타데이터를 자동 기록). 먼저 AutoGen 런타임을 계측하는 데 쓸 추적 서비스를 만들어 볼게요.
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
# Set up telemetry span exporter.
otel_exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
span_processor = BatchSpanProcessor(otel_exporter)
# Set up telemetry trace provider.
tracer_provider = TracerProvider(resource=Resource({"service.name": "autogen-test-agentchat"}))
tracer_provider.add_span_processor(span_processor)
trace.set_tracer_provider(tracer_provider)
# Instrument the OpenAI Python library
OpenAIInstrumentor().instrument()
# we will get reference this tracer later using its service name
# tracer = trace.get_tracer("autogen-test-agentchat")
팀을 만드는 코드는 이미 익숙할 거예요.
AgentChat 팀은 AutoGen Core의 에이전트 런타임으로 실행돼요. 그리고 런타임은 이미 로그하도록 계측돼 있어요. Core Telemetry Guide를 참고하세요. 에이전트 런타임 텔레메트리를 끄려면 런타임 생성자에서
trace_provider를opentelemetry.trace.NoOpTracerProvider로 설정하면 돼요.런타임 생성자에 접근할 수 없으면(예:
ComponentConfig를 쓸 때) 환경 변수AUTOGEN_DISABLE_RUNTIME_TRACING을true로 설정해 에이전트 런타임 텔레메트리를 끌 수도 있어요.
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_core import SingleThreadedAgentRuntime
from autogen_ext.models.openai import OpenAIChatCompletionClient
def search_web_tool(query: str) -> str:
if "2006-2007" in query:
return """Here are the total points scored by Miami Heat players in the 2006-2007 season:
Udonis Haslem: 844 points
Dwayne Wade: 1397 points
James Posey: 550 points
...
"""
elif "2007-2008" in query:
return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2007-2008 is 214."
elif "2008-2009" in query:
return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2008-2009 is 398."
return "No data found."
def percentage_change_tool(start: float, end: float) -> float:
return ((end - start) / start) * 100
async def main() -> None:
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# Get a tracer with the default tracer provider.
tracer = trace.get_tracer("tracing-autogen-agentchat")
# Use the tracer to create a span for the main function.
with tracer.start_as_current_span("run_team"):
planning_agent = AssistantAgent(
"PlanningAgent",
description="An agent for planning tasks, this agent should be the first to engage when given a new task.",
model_client=model_client,
system_message="""
You are a planning agent.
Your job is to break down complex tasks into smaller, manageable subtasks.
Your team members are:
WebSearchAgent: Searches for information
DataAnalystAgent: Performs calculations
You only plan and delegate tasks - you do not execute them yourself.
When assigning tasks, use this format:
1. <agent> : <task>
After all tasks are complete, summarize the findings and end with "TERMINATE".
""",
)
web_search_agent = AssistantAgent(
"WebSearchAgent",
description="An agent for searching information on the web.",
tools=[search_web_tool],
model_client=model_client,
system_message="""
You are a web search agent.
Your only tool is search_tool - use it to find information.
You make only one search call at a time.
Once you have the results, you never do calculations based on them.
""",
)
data_analyst_agent = AssistantAgent(
"DataAnalystAgent",
description="An agent for performing calculations.",
model_client=model_client,
tools=[percentage_change_tool],
system_message="""
You are a data analyst.
Given the tasks you have been assigned, you should analyze the data and provide results using the tools provided.
If you have not seen the data, ask for it.
""",
)
text_mention_termination = TextMentionTermination("TERMINATE")
max_messages_termination = MaxMessageTermination(max_messages=25)
termination = text_mention_termination | max_messages_termination
selector_prompt = """Select an agent to perform task.
{roles}
Current conversation context:
{history}
Read the above conversation, then select an agent from {participants} to perform the next task.
Make sure the planner agent has assigned tasks before other agents start working.
Only select one agent.
"""
task = "Who was the Miami Heat player with the highest points in the 2006-2007 season, and what was the percentage change in his total rebounds between the 2007-2008 and 2008-2009 seasons?"
runtime = SingleThreadedAgentRuntime(
tracer_provider=trace.NoOpTracerProvider(), # Disable telemetry for runtime.
)
runtime.start()
team = SelectorGroupChat(
[planning_agent, web_search_agent, data_analyst_agent],
model_client=model_client,
termination_condition=termination,
selector_prompt=selector_prompt,
allow_repeated_speaker=True,
runtime=runtime,
)
await Console(team.run_stream(task=task))
await runtime.stop()
await model_client.close()
# asyncio.run(main())
await main()
그러면 위 애플리케이션 실행에서 수집한 트레이스를 Jaeger UI로 확인할 수 있어요.