在开发 AI 应用时，我们常常面临模型集成的难题：不同平台的 API 格式各异，流式响应处理复杂，缓存优化难以实现。今天要分享的 AutoGen 模型客户端体系，就像一套标准化的 "模型适配器"，让我们能轻松对接多种 LLM 服务。从 OpenAI 到 Azure，从本地模型到云端服务，AutoGen 提供了统一的调用接口和丰富的扩展能力，一起来探索如何让模型集成变得简单高效。

一、内置模型客户端全景：适配多种服务的瑞士军刀

AutoGen 目前内置了多种模型客户端，覆盖主流 LLM 服务和本地部署场景，我们可以根据需求选择最合适的工具：

1. 云端模型客户端

• OpenAIChatCompletionClient：最常用的客户端，支持 OpenAI 全系模型（如 gpt-4、gpt-3.5-turbo），同时兼容 Gemini 等 API 兼容模型

  python

  # 初始化客户端（环境变量需设置OPENAI_API_KEY）
  from autogen_ext.models.openai import OpenAIChatCompletionClient
  client = OpenAIChatCompletionClient(model="gpt-4o", temperature=0.3)
  

• AzureOpenAIChatCompletionClient：专为 Azure OpenAI 服务设计，处理特殊的部署名称和认证方式

  python

  from autogen_ext.models.azure import AzureOpenAIChatCompletionClient
  client = AzureOpenAIChatCompletionClient(
      model="my-azure-deployment",
      api_base="https://my-azure-endpoint.com",
      api_version="2023-05-15"
  )
  

• AzureAIChatCompletionClient：扩展支持 GitHub 模型和其他 Azure 托管模型，更灵活的部署适配

  python

  from autogen_ext.models.azure import AzureAIChatCompletionClient
  client = AzureAIChatCompletionClient(
      model="github-model",
      endpoint="https://github-endpoint.com"
  )
  

2. 实验性与本地模型客户端

• OllamaChatCompletionClient：用于本地部署在 Ollama 的模型，适合隐私要求高的场景（实验性）

  python

  from autogen_ext.models.ollama import OllamaChatCompletionClient
  client = OllamaChatCompletionClient(model="llama3", base_url="http://localhost:11434")
  

• AnthropicChatCompletionClient：对接 Anthropic 平台模型（如 Claude），支持其特有的提示词格式（实验性）

  python

  from autogen_ext.models.anthropic import AnthropicChatCompletionClient
  client = AnthropicChatCompletionClient(model="claude-3")
  

• SKChatCompletionAdapter：语义内核 (Semantic Kernel) 连接器适配器，桥接不同 AI 框架

  python

  from autogen_ext.models.sk import SKChatCompletionAdapter
  client = SKChatCompletionAdapter(kernel=my_semantic_kernel)
  

二、模型调用基础：从单次请求到日志记录

1. 基本调用流程与日志记录

python

import logging
from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
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)

# 初始化模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4", temperature=0.3)

# 构造用户消息
messages = [UserMessage(content="What is the capital of France?", source="user")]

# 调用模型（异步方式）
result = await model_client.create(messages)

# 输出结果
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

2. 日志记录详解

• 日志名称：使用autogen_core.EVENT_LOGGER_NAME
• 事件类型：模型调用记录为LLMCall事件
• 日志内容：包含请求参数、响应结果、令牌使用情况
• 查看方式：配置 StreamHandler 后，日志会输出到控制台或文件

三、流式响应处理：打造实时交互体验

1. 流式响应的实现与处理

python

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

# 初始化支持流式的客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o")

# 构造请求消息
messages = [UserMessage(content="Write a very short story about a dragon.", source="user")]

# 创建流式响应
stream = model_client.create_stream(messages=messages)

# 处理流式数据
print("Streamed responses:")
async for chunk in stream:
    if isinstance(chunk, str):
        # 中间数据块（字符串）
        print(chunk, flush=True, end="")
    else:
        # 最终响应（CreateResult对象）
        assert isinstance(chunk, CreateResult)
        print("\n\n------------\n")
        print("The complete response:", flush=True)
        print(chunk.content, flush=True)

2. 流式响应关键特性

• 实时性：令牌逐个生成，适合聊天界面等实时场景
• 最终响应：最后一个 chunk 一定是 CreateResult 对象，包含完整内容
• 令牌使用：默认 usage 为 0，需特殊配置才能获取准确统计
• 处理方式：异步迭代处理，保持界面响应不阻塞

四、结构化输出：让模型响应可预测

1. 使用 Pydantic 模型定义结构化输出

python

from typing import Literal
from pydantic import BaseModel
from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 定义结构化响应模型
class AgentResponse(BaseModel):
    thoughts: str
    response: Literal["happy", "sad", "neutral"]

# 初始化客户端并设置response_format
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    response_format=AgentResponse  # 关键：指定Pydantic模型
)

# 构造请求
messages = [UserMessage(content="I am happy.", source="user")]

# 调用模型
response = await model_client.create(messages=messages)

# 解析结构化响应
parsed_response = AgentResponse.model_validate_json(response.content)
print(parsed_response.thoughts)
print(parsed_response.response)

# 输出示例:
# I'm glad to hear that you're feeling happy! It's such a great emotion...
# happy

2. 结构化输出注意事项

• 模型支持：仅支持能理解response_format指令的模型（如 gpt-4o 及以上）
• 客户端支持：目前仅 OpenAIChatCompletionClient 和 AzureOpenAIChatCompletionClient 支持
• 两种设置方式：
  • 客户端初始化时设置response_format
  • 调用 create () 时通过extra_create_args设置
• 解析方式：使用 Pydantic 的model_validate_json确保数据格式正确

五、缓存优化：减少令牌消耗与响应延迟

1. 本地磁盘缓存实现

python

import asyncio
import tempfile
from autogen_core.models import UserMessage
from autogen_ext.cache_store.diskcache import DiskCacheStore
from autogen_ext.models.cache import ChatCompletionCache
from autogen_ext.models.openai import OpenAIChatCompletionClient
from diskcache import Cache

async def main():
    with tempfile.TemporaryDirectory() as tmpdir:
        # 初始化原始客户端
        original_client = OpenAIChatCompletionClient(model="gpt-4o")
        
        # 初始化磁盘缓存存储
        cache = Cache(tmpdir)
        cache_store = DiskCacheStore(cache)
        
        # 包装缓存客户端
        cache_client = ChatCompletionCache(original_client, cache_store)
        
        # 首次调用（实际请求模型）
        response1 = await cache_client.create([UserMessage(content="Hello, how are you?")])
        print("首次响应:", response1.content)
        
        # 二次调用（读取缓存）
        response2 = await cache_client.create([UserMessage(content="Hello, how are you?")])
        print("二次响应:", response2.content)
        
        # 验证缓存效果
        print("首次调用令牌使用:", original_client.total_usage())
        print("二次调用令牌使用:", original_client.total_usage())  # 应与首次相同
        
        # 关闭客户端
        await original_client.close()
        await cache_client.close()

asyncio.run(main())

2. 缓存机制核心要点

• 支持类型：可包装任何 ChatCompletionClient 实现缓存
• 缓存存储：
  • DiskCacheStore：本地磁盘存储（适合单机场景）
  • RedisStore：Redis 存储（适合分布式场景）
• 缓存 key：基于请求参数（消息内容、模型参数等）生成，参数变化会导致缓存未命中
• 使用建议：
  • 对相同输入的重复请求效果显著
  • 敏感内容请求谨慎使用缓存
  • 定期清理过期缓存释放空间

六、构建智能体：集成模型客户端到完整应用

1. 简单智能体的实现与使用

python

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:
        # 构造模型输入
        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 Message(content=response.content)

# 初始化运行时和智能体
from autogen_core import AgentId

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
runtime = SingleThreadedAgentRuntime()

# 注册智能体
await SimpleAgent.register(
    runtime,
    "simple_agent",
    lambda: SimpleAgent(model_client=model_client)
)

# 启动运行时
runtime.start()

# 发送消息并获取响应
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)

# 停止运行时
await runtime.stop()
await model_client.close()

2. 智能体构建关键步骤

• 消息定义：使用 dataclass 定义输入输出消息格式
• 模型集成：在智能体中持有模型客户端实例
• 消息处理：重写 message_handler 方法处理用户消息
• 运行时管理：
  • 注册智能体到运行时
  • 启动运行时处理消息
  • 正确关闭资源释放连接

结语：让模型集成成为开发加速器

通过今天的全面解析，我们已经掌握了 AutoGen 模型客户端的核心能力 —— 从基础调用到流式处理，从结构化输出到智能体构建。这套灵活的模型适配体系，不仅能大幅减少我们的集成工作量，还能为应用添加缓存优化、日志记录等高级功能。

如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～