openinference-instrumentation-autogen-agentchat (original) (raw)

PyPI Version

OpenTelelemetry instrumentation for Autogen AgentChat, enabling tracing of agent interactions and conversations.

The traces emitted by this instrumentation are fully OpenTelemetry compatible and can be sent to an OpenTelemetry collector for viewing, such as arize-phoenix

Installation

pip install openinference-instrumentation-autogen-agentchat

Quickstart

In this example we will instrument a simple Autogen AgentChat application and observe the traces via arize-phoenix.

Install required packages.

pip install openinference-instrumentation-autogen-agentchat autogen-agentchat arize-phoenix opentelemetry-sdk opentelemetry-exporter-otlp

Start the phoenix server so that it is ready to collect traces. The Phoenix server runs entirely on your machine and does not send data over the internet.

phoenix serve

Here's a simple example using a single assistant agent:

import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_ext.models.openai import OpenAIChatCompletionClient from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk import trace as trace_sdk from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor

from openinference.instrumentation.autogen_agentchat import AutogenAgentChatInstrumentor

Set up the tracer provider

endpoint = "http://127.0.0.1:6006/v1/traces" tracer_provider = trace_sdk.TracerProvider() tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint)))

Instrument AutogenAgentChat

AutogenAgentChatInstrumentor().instrument(tracer_provider=tracer_provider)

async def main(): model_client = OpenAIChatCompletionClient( model="gpt-3.5-turbo", )

def get_weather(city: str) -> str:
    """Get the weather for a given city."""
    return f"The weather in {city} is 73 degrees and Sunny."

# Create an assistant agent with tools
agent = AssistantAgent(
    name="weather_agent",
    model_client=model_client,
    tools=[get_weather],
    system_message="You are a helpful assistant that can check the weather.",
    reflect_on_tool_use=True,
    model_client_stream=True,
)

result = await agent.run(task="What is the weather in New York?")
await model_client.close()
print(result)

if name == "main": asyncio.run(main())

For a more complex example using multiple agents in a team:

import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.conditions import TextMentionTermination from autogen_agentchat.teams import RoundRobinGroupChat from autogen_ext.models.openai import OpenAIChatCompletionClient from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk import trace as trace_sdk from opentelemetry.sdk.trace.export import SimpleSpanProcessor

from openinference.instrumentation.autogen_agentchat import AutogenAgentChatInstrumentor

Set up the tracer provider

endpoint = "http://127.0.0.1:6006/v1/traces" tracer_provider = trace_sdk.TracerProvider() tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint)))

Instrument AutogenAgentChat

AutogenAgentChatInstrumentor().instrument(tracer_provider=tracer_provider)

async def main(): model_client = OpenAIChatCompletionClient( model="gpt-4", )

# Create two agents: a primary and a critic
primary_agent = AssistantAgent(
    "primary",
    model_client=model_client,
    system_message="You are a helpful AI assistant.",
)

critic_agent = AssistantAgent(
    "critic",
    model_client=model_client,
    system_message="""
    Provide constructive feedback.
    Respond with 'APPROVE' when your feedbacks are addressed.
    """,
)

# Termination condition: stop when the critic says "APPROVE"
text_termination = TextMentionTermination("APPROVE")

# Create a team with both agents
team = RoundRobinGroupChat(
    [primary_agent, critic_agent],
    termination_condition=text_termination
)

# Run the team on a task
result = await team.run(task="Write a short poem about the fall season.")
await model_client.close()
print(result)

if name == "main": asyncio.run(main())

Since we are using OpenAI, we must set the OPENAI_API_KEY environment variable to authenticate with the OpenAI API.

export OPENAI_API_KEY=[your_key_here]

Now simply run the python file and observe the traces in Phoenix.

python your_file.py

More Info