模型客户端#

AutoGen 提供了一套内置的模型客户端，用于使用 ChatCompletion API。所有模型客户端都实现了 ChatCompletionClient 协议类。

目前我们支持以下内置模型客户端

OpenAIChatCompletionClient：用于 OpenAI 模型和具有 OpenAI API 兼容性的模型（例如，Gemini）。
AzureOpenAIChatCompletionClient：用于 Azure OpenAI 模型。
AzureAIChatCompletionClient：用于 GitHub 模型和托管在 Azure 上的模型。
OllamaChatCompletionClient（实验性）：用于托管在 Ollama 上的本地模型。
AnthropicChatCompletionClient（实验性）：用于托管在 Anthropic 上的模型。
SKChatCompletionAdapter：Semantic Kernel AI 连接器的适配器。

有关如何使用这些模型客户端的更多信息，请参阅每个客户端的文档。

记录模型调用#

AutoGen 使用标准 Python 日志模块来记录事件，例如模型调用和响应。记录器名称是 autogen_core.EVENT_LOGGER_NAME，事件类型是 LLMCall。

import logging

from autogen_core import EVENT_LOGGER_NAME

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(EVENT_LOGGER_NAME)
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.INFO)

调用模型客户端#

要调用模型客户端，可以使用 create() 方法。此示例使用 OpenAIChatCompletionClient 调用 OpenAI 模型。

from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4", temperature=0.3
)  # assuming OPENAI_API_KEY is set in the environment.

result = await model_client.create([UserMessage(content="What is the capital of France?", source="user")])
print(result)

finish_reason='stop' content='The capital of France is Paris.' usage=RequestUsage(prompt_tokens=15, completion_tokens=8) cached=False logprobs=None thought=None

流式传输令牌#

您可以使用 create_stream() 方法创建具有流式传输令牌块的聊天完成请求。

from autogen_core.models import CreateResult, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o")  # assuming OPENAI_API_KEY is set in the environment.

messages = [
    UserMessage(content="Write a very short story about a dragon.", source="user"),
]

# Create a stream.
stream = model_client.create_stream(messages=messages)

# Iterate over the stream and print the responses.
print("Streamed responses:")
async for chunk in stream:  # type: ignore
    if isinstance(chunk, str):
        # The chunk is a string.
        print(chunk, flush=True, end="")
    else:
        # The final chunk is a CreateResult object.
        assert isinstance(chunk, CreateResult) and isinstance(chunk.content, str)
        # The last response is a CreateResult object with the complete message.
        print("\n\n------------\n")
        print("The complete response:", flush=True)
        print(chunk.content, flush=True)

Streamed responses:
In the heart of an ancient forest, beneath the shadow of snow-capped peaks, a dragon named Elara lived secretly for centuries. Elara was unlike any dragon from the old tales; her scales shimmered with a deep emerald hue, each scale engraved with symbols of lost wisdom. The villagers in the nearby valley spoke of mysterious lights dancing across the night sky, but none dared venture close enough to solve the enigma.

One cold winter's eve, a young girl named Lira, brimming with curiosity and armed with the innocence of youth, wandered into Elara’s domain. Instead of fire and fury, she found warmth and a gentle gaze. The dragon shared stories of a world long forgotten and in return, Lira gifted her simple stories of human life, rich in laughter and scent of earth.

From that night on, the villagers noticed subtle changes—the crops grew taller, and the air seemed sweeter. Elara had infused the valley with ancient magic, a guardian of balance, watching quietly as her new friend thrived under the stars. And so, Lira and Elara’s bond marked the beginning of a timeless friendship that spun tales of hope whispered through the leaves of the ever-verdant forest.

------------

The complete response:
In the heart of an ancient forest, beneath the shadow of snow-capped peaks, a dragon named Elara lived secretly for centuries. Elara was unlike any dragon from the old tales; her scales shimmered with a deep emerald hue, each scale engraved with symbols of lost wisdom. The villagers in the nearby valley spoke of mysterious lights dancing across the night sky, but none dared venture close enough to solve the enigma.

One cold winter's eve, a young girl named Lira, brimming with curiosity and armed with the innocence of youth, wandered into Elara’s domain. Instead of fire and fury, she found warmth and a gentle gaze. The dragon shared stories of a world long forgotten and in return, Lira gifted her simple stories of human life, rich in laughter and scent of earth.

From that night on, the villagers noticed subtle changes—the crops grew taller, and the air seemed sweeter. Elara had infused the valley with ancient magic, a guardian of balance, watching quietly as her new friend thrived under the stars. And so, Lira and Elara’s bond marked the beginning of a timeless friendship that spun tales of hope whispered through the leaves of the ever-verdant forest.

------------

The token usage was:
RequestUsage(prompt_tokens=0, completion_tokens=0)

注意

流式传输响应中的最后一个响应始终是类型为 CreateResult 的最终响应。

注意

默认的使用情况响应是返回零值。要启用使用情况，请参阅 create_stream() 以获取更多详细信息。

结构化输出#

可以通过在 OpenAIChatCompletionClient 和 AzureOpenAIChatCompletionClient 中将 response_format 字段设置为 Pydantic BaseModel 类来启用结构化输出。

注意

结构化输出仅适用于支持它的模型。它还需要模型客户端也支持结构化输出。目前，OpenAIChatCompletionClient 和 AzureOpenAIChatCompletionClient 支持结构化输出。

from typing import Literal

from pydantic import BaseModel


# The response format for the agent as a Pydantic base model.
class AgentResponse(BaseModel):
    thoughts: str
    response: Literal["happy", "sad", "neutral"]


# Create an agent that uses the OpenAI GPT-4o model with the custom response format.
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    response_format=AgentResponse,  # type: ignore
)

# Send a message list to the model and await the response.
messages = [
    UserMessage(content="I am happy.", source="user"),
]
response = await model_client.create(messages=messages)
assert isinstance(response.content, str)
parsed_response = AgentResponse.model_validate_json(response.content)
print(parsed_response.thoughts)
print(parsed_response.response)

# Close the connection to the model client.
await model_client.close()

I'm glad to hear that you're feeling happy! It's such a great emotion that can brighten your whole day. Is there anything in particular that's bringing you joy today? 😊
happy

您还可以使用 create() 方法中的 extra_create_args 参数来设置 response_format 字段，以便可以为每个请求配置结构化输出。

缓存模型响应#

autogen_ext 实现了 ChatCompletionCache，它可以包装任何 ChatCompletionClient。当多次使用相同的提示查询底层客户端时，使用此包装器可以避免产生令牌使用量。

ChatCompletionCache 使用 CacheStore 协议。我们已经实现了一些有用的 CacheStore 变体，包括 DiskCacheStore 和 RedisStore。

这是一个使用 diskcache 进行本地缓存的示例

# pip install -U "autogen-ext[openai, diskcache]"

import asyncio
import tempfile

from autogen_core.models import UserMessage
from autogen_ext.cache_store.diskcache import DiskCacheStore
from autogen_ext.models.cache import CHAT_CACHE_VALUE_TYPE, ChatCompletionCache
from autogen_ext.models.openai import OpenAIChatCompletionClient
from diskcache import Cache


async def main() -> None:
    with tempfile.TemporaryDirectory() as tmpdirname:
        # Initialize the original client
        openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")

        # Then initialize the CacheStore, in this case with diskcache.Cache.
        # You can also use redis like:
        # from autogen_ext.cache_store.redis import RedisStore
        # import redis
        # redis_instance = redis.Redis()
        # cache_store = RedisCacheStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
        cache_store = DiskCacheStore[CHAT_CACHE_VALUE_TYPE](Cache(tmpdirname))
        cache_client = ChatCompletionCache(openai_model_client, cache_store)

        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # Should print response from OpenAI
        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # Should print cached response

        await openai_model_client.close()
        await cache_client.close()


asyncio.run(main())

True

在缓存响应之前和之后检查 cached_client.total_usage()（或 model_client.total_usage()）应产生相同的计数。

请注意，缓存对提供给 cached_client.create 或 cached_client.create_stream 的确切参数很敏感，因此更改 tools 或 json_output 参数可能会导致缓存未命中。

使用模型客户端构建代理#

让我们创建一个简单的 AI 代理，该代理可以使用 ChatCompletion API 响应消息。

from dataclasses import dataclass

from autogen_core import MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler
from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient


@dataclass
class Message:
    content: str


class SimpleAgent(RoutedAgent):
    def __init__(self, model_client: ChatCompletionClient) -> None:
        super().__init__("A simple agent")
        self._system_messages = [SystemMessage(content="You are a helpful AI assistant.")]
        self._model_client = model_client

    @message_handler
    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message:
        # Prepare input to the chat completion model.
        user_message = UserMessage(content=message.content, source="user")
        response = await self._model_client.create(
            self._system_messages + [user_message], cancellation_token=ctx.cancellation_token
        )
        # Return with the model's response.
        assert isinstance(response.content, str)
        return Message(content=response.content)

SimpleAgent 类是 autogen_core.RoutedAgent 类的子类，为了方便地将消息自动路由到适当的处理程序。它有一个单一的处理程序 handle_user_message，用于处理来自用户的消息。它使用 ChatCompletionClient 生成消息的响应。然后，它按照直接通信模型将响应返回给用户。

注意

类型为 autogen_core.CancellationToken 的 cancellation_token 用于取消异步操作。它链接到消息处理程序中的异步调用，并且可以由调用者用来取消处理程序。

# Create the runtime and register the agent.
from autogen_core import AgentId

model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    # api_key="sk-...", # Optional if you have an OPENAI_API_KEY set in the environment.
)

runtime = SingleThreadedAgentRuntime()
await SimpleAgent.register(
    runtime,
    "simple_agent",
    lambda: SimpleAgent(model_client=model_client),
)
# Start the runtime processing messages.
runtime.start()
# Send a message to the agent and get the response.
message = Message("Hello, what are some fun things to do in Seattle?")
response = await runtime.send_message(message, AgentId("simple_agent", "default"))
print(response.content)
# Stop the runtime processing messages.
await runtime.stop()
await model_client.close()

Seattle is a vibrant city with a wide range of activities and attractions. Here are some fun things to do in Seattle:

**Space Needle**: Visit this iconic observation tower for stunning views of the city and surrounding mountains.

**Pike Place Market**: Explore this historic market where you can see the famous fish toss, buy local produce, and find unique crafts and eateries.

**Museum of Pop Culture (MoPOP)**: Dive into the world of contemporary culture, music, and science fiction at this interactive museum.

**Chihuly Garden and Glass**: Marvel at the beautiful glass art installations by artist Dale Chihuly, located right next to the Space Needle.

**Seattle Aquarium**: Discover the diverse marine life of the Pacific Northwest at this engaging aquarium.

**Seattle Art Museum**: Explore a vast collection of art from around the world, including contemporary and indigenous art.

**Kerry Park**: For one of the best views of the Seattle skyline, head to this small park on Queen Anne Hill.

**Ballard Locks**: Watch boats pass through the locks and observe the salmon ladder to see salmon migrating.

**Ferry to Bainbridge Island**: Take a scenic ferry ride across Puget Sound to enjoy charming shops, restaurants, and beautiful natural scenery.

**Olympic Sculpture Park**: Stroll through this outdoor park with large-scale sculptures and stunning views of the waterfront and mountains.

**Underground Tour**: Discover Seattle's history on this quirky tour of the city's underground passageways in Pioneer Square.

**Seattle Waterfront**: Enjoy the shops, restaurants, and attractions along the waterfront, including the Seattle Great Wheel and the aquarium.

**Discovery Park**: Explore the largest green space in Seattle, featuring trails, beaches, and views of Puget Sound.

**Food Tours**: Try out Seattle’s diverse culinary scene, including fresh seafood, international cuisines, and coffee culture (don’t miss the original Starbucks!).

**Attend a Sports Game**: Catch a Seahawks (NFL), Mariners (MLB), or Sounders (MLS) game for a lively local experience.

Whether you're interested in culture, nature, food, or history, Seattle has something for everyone to enjoy!

上面的 SimpleAgent 总是使用一个全新的上下文进行响应，该上下文仅包含系统消息和用户最新的消息。我们可以使用来自 autogen_core.model_context 的模型上下文类，使 agent 能够“记住”之前的对话。有关更多详细信息，请参阅模型上下文页面。

从环境变量获取 API 密钥#

在上面的示例中，我们展示了可以通过 api_key 参数提供 API 密钥。重要的是，OpenAI 和 Azure OpenAI 客户端使用 openai 包，如果未提供 API 密钥，它会自动从环境变量中读取。

对于 OpenAI，您可以设置 OPENAI_API_KEY 环境变量。
对于 Azure OpenAI，您可以设置 AZURE_OPENAI_API_KEY 环境变量。

此外，对于 Gemini (Beta)，您可以设置 GEMINI_API_KEY 环境变量。

探索这种方法是一个好习惯，因为它避免了在代码中包含敏感的 API 密钥。