当前位置：首页 > news >正文

AutoGen - model_clients和model_context使用示例

news 2025/7/31 20:59:49

AutoGen 中的model_clients和model_context使用示例

环境变量设置

export API_KEY="your_openai_api_key"
export API_BASE_URL="your_api_base_url"  # 可选，默认使用 OpenAI 官方 API

基础客户端配置

from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelInfoapi_client = OpenAIChatCompletionClient(model="gpt-4.1-nano",api_key=openai_api_key,base_url=api_base_url,model_info=ModelInfo(vision=True,           # 支持视觉输入function_calling=True, # 支持函数调用json_output=True,      # 支持 JSON 输出family="unknown",      # 模型家族structured_output=True # 支持结构化输出),
)

示例详解

1. 基础模型调用 (`main`)

功能：最基本的模型调用方式，发送消息并获取响应。

示例代码：

async def main(city: str) -> str:message = UserMessage(content=f"What is the weather in {city}?", source="user")result = await api_client.create([message])return str(result.content) if result.content is not None else ""

使用场景：

简单的问答对话
单次请求响应
基础的文本生成

运行方式：

import asyncio
response = asyncio.run(main("San Francisco"))
print(response)

2. 流式输出 (`main_stream`)

功能：实时流式接收模型响应，适合长文本生成或实时对话。

示例代码：

async def main_stream(city: str):message = UserMessage(content=f"What is the weather in {city}?", source="user")result = api_client.create_stream([message])print("Streamed responses:")async for chunk in result:if isinstance(chunk, str):# 流式文本片段print(chunk, flush=True, end="")else:# 最终完整响应assert isinstance(chunk, CreateResult)print("\n\n------------\n")print("The complete response:", flush=True)print(chunk.content, flush=True)

使用场景：

长文本生成（文章、代码等）
实时聊天界面
需要即时反馈的应用

优势：

降低用户等待时间
提供实时反馈
更好的用户体验

3. 结构化输出 (`main_structured_output`)

功能：使用 Pydantic 模型定义响应格式，确保输出符合预期结构。

示例代码：

from pydantic import BaseModel
from typing import Literalclass CustomMessage(BaseModel):thoughts: strresponse: Literal["happy", "sad", "angry"]# 配置客户端使用结构化输出
api_client = OpenAIChatCompletionClient(# ... 其他配置response_format=CustomMessage,  # 指定响应格式
)async def main_structured_output():message = UserMessage(content="I am happy.", source="user")response = await api_client.create([message])# 解析结构化响应parsed_response = CustomMessage.model_validate_json(response.content)print(parsed_response.thoughts)   # 输出思考过程print(parsed_response.response)   # 输出情感分类

使用场景：

情感分析
分类任务
需要特定格式输出的应用
API 集成

优势：

类型安全
数据验证
易于后续处理

4. 缓存机制 (`chat_cache_example`)

功能：缓存模型响应，避免重复请求，节约 token 和成本。

示例代码：

import tempfile
from autogen_ext.cache_store.diskcache import DiskCacheStore
from autogen_ext.models.cache import ChatCompletionCache
from diskcache import Cacheasync def chat_cache_example():with tempfile.TemporaryDirectory() as tmpdirname:# 创建缓存存储cache_store = DiskCacheStore(Cache(tmpdirname))# 创建缓存客户端cache_client = ChatCompletionCache(api_client, cache_store)message = UserMessage(content="What is the weather like in Boston?", source="user")# 第一次请求（会调用 API）response1 = await cache_client.create(messages=[message])print(response1.content)# 第二次相同请求（从缓存返回）response2 = await cache_client.create(messages=[message])print(response2.content)  # 内容相同，但无需 API 调用

使用场景：

开发测试阶段
重复性查询
成本控制
提高响应速度

优势：

节约 API 调用成本
提高响应速度
减少网络请求

5. 基础 Agent (`main_agent`)

功能：使用模型客户端构建简单的 Agent，处理消息并返回响应。

示例代码：

from autogen_core import RoutedAgent, message_handler, MessageContext
from dataclasses import dataclass@dataclass
class Message:content: strclass MyAgent(RoutedAgent):def __init__(self, name: str, model_client: ChatCompletionClient):super().__init__(name)self.system_messages = [SystemMessage(content="You are a helpful assistant.")]self.model_client = model_client@message_handlerasync def handle_message(self, message: Message, context: MessageContext) -> Message:user_message = UserMessage(content=message.content, source='user')response = await self.model_client.create(self.system_messages + [user_message], cancellation_token=context.cancellation_token)return Message(content=response.content)async def main_agent():runtime = SingleThreadedAgentRuntime()await MyAgent.register(runtime, 'simple_agent', lambda: MyAgent('A simple agent', 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()

使用场景：

简单的聊天机器人
单轮对话系统
基础的 AI 助手

特点：

无状态
每次请求独立
简单直接

6. 带上下文的 Agent (`main_agent_buffered_chat`)

功能：具有记忆能力的 Agent，能够维护对话历史和上下文。

示例代码：

from autogen_core.model_context import BufferedChatCompletionContext
from autogen_core.models import AssistantMessageclass MyAgentWithContext(RoutedAgent):def __init__(self, name: str, model_client: ChatCompletionClient):super().__init__(name)self._system_messages = [SystemMessage(content="You are a helpful assistant.")]self._model_client = model_clientself._model_context = BufferedChatCompletionContext(buffer_size=5)  # 缓存5条消息@message_handlerasync def handle_message(self, message: Message, context: MessageContext) -> Message:user_message = UserMessage(content=message.content, source='user')# 添加用户消息到上下文await self._model_context.add_message(user_message)# 使用完整上下文生成响应response = await self._model_client.create(self._system_messages + (await self._model_context.get_messages()),cancellation_token=context.cancellation_token)# 添加助手响应到上下文await self._model_context.add_message(AssistantMessage(content=response.content, source=self.metadata["type"]))return Message(content=response.content)async def main_agent_buffered_chat():runtime = SingleThreadedAgentRuntime()await MyAgentWithContext.register(runtime, 'simple_agent_context',lambda: MyAgentWithContext('A simple agent', model_client))runtime.start()agent_id = AgentId("simple_agent_context", "default")# 第一条消息message1 = Message("Hello, what are some fun things to do in Seattle?")response1 = await runtime.send_message(message1, agent_id)print(response1.content)# 第二条消息（引用前面的内容）message2 = Message("What's the last thing you mentioned?")response2 = await runtime.send_message(message2, agent_id)print(response2.content)  # 能够引用之前的回答await runtime.stop()await model_client.close()