如何基于 AutoGen AgentChat BaseChatAgent 实现自定义智能体和自定义模型客户端？

如何基于 AutoGen AgentChat BaseChatAgent 实现自定义智能体和自定义模型客户端【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen当你使用的 AutoGen AgentChat 预设智能体如AssistantAgent无法满足需求时——例如想让智能体执行预设之外的自定义行为或者想接入一个官方扩展包没有提供的模型文档以直接调用 Google Gemini SDK 为例——可以继承BaseChatAgent自己实现一个智能体。AutoGen AgentChat 中所有智能体都继承自 {py:class}autogen_agentchat.agents.BaseChatAgent只要实现了它的抽象方法和属性就能单独运行也可以作为团队成员参与群聊。本文按实现一个最简单智能体 → 接入自定义模型客户端 → 放进团队 → 可选的声明式配置的顺序完成这条路径官方示例见 custom-agents.ipynb。准备环境按 安装文档 的要求Python 需要 3.10 或更高版本autogen-agentchat包通过 pip 安装pip install -U autogen-agentchat后文的团队示例SelectorGroupChat/RoundRobinGroupChat用到了 OpenAI 模型客户端需要额外安装扩展pip install autogen-ext[openai]如果走 Gemini 自定义客户端路径则按文档要求安装 Google Gemini SDKpip install google-genaiGemini 示例的构造函数默认读取环境变量GEMINI_API_KEYapi_key: str os.environ[GEMINI_API_KEY]运行前需要设置好该变量OpenAI 客户端同理需要对应的密钥环境变量文档未展开。BaseChatAgent 约定要重写哪些成员抽象基类定义在 _base_chat_agent.py它要求实现三项另有一项可选on_messages抽象方法处理消息并返回Response对象。run方法内部会调用它。on_reset抽象方法把智能体重置回初始状态。produced_message_types抽象属性返回该智能体可能产生的BaseChatMessage类型列表。on_messages_stream可选流式产出消息。不实现时默认实现会调用on_messages并把响应中的消息依次 yield 出来run_stream依赖它。两条必须遵守的状态约定写错了行为会不对但通常不会直接报错智能体是有状态的。每次调用on_messages传入的应只包含新增消息不要每次都传完整对话历史需要历史时要自己维护。文档明确提示on_messages可能收到空消息列表这表示该智能体之前被调用过、这次调用没有新消息所以维护历史很关键例如 SelectorGroupChat 中的示例 里对空列表的注释。智能体名称必须是合法的 Python 标识符BaseChatAgent.__init__会对name.isidentifier()为假的名称抛出ValueError。最短路径一个不依赖模型的自定义智能体先用文档中的CountDownAgent走通实现 → 运行 → 验证的最小闭环。它从给定数字倒数到 0并流式产出中间消息不涉及任何模型调用因此不需要 API key适合作为第一步验证代码结构是否正确from typing import AsyncGenerator, List, Sequence from autogen_agentchat.agents import BaseChatAgent from autogen_agentchat.base import Response from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage, TextMessage from autogen_core import CancellationToken class CountDownAgent(BaseChatAgent): def __init__(self, name: str, count: int 3): super().__init__(name, A simple agent that counts down.) self._count count property def produced_message_types(self) - Sequence[type[BaseChatMessage]]: return (TextMessage,) async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) - Response: # Calls the on_messages_stream. response: Response | None None async for message in self.on_messages_stream(messages, cancellation_token): if isinstance(message, Response): response message assert response is not None return response async def on_messages_stream( self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken ) - AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None]: inner_messages: List[BaseAgentEvent | BaseChatMessage] [] for i in range(self._count, 0, -1): msg TextMessage(contentf{i}..., sourceself.name) inner_messages.append(msg) yield msg # The response is returned at the end of the stream. # It contains the final message and all the inner messages. yield Response(chat_messageTextMessage(contentDone!, sourceself.name), inner_messagesinner_messages) async def on_reset(self, cancellation_token: CancellationToken) - None: pass async def run_countdown_agent() - None: # Create a countdown agent. countdown_agent CountDownAgent(countdown) # Run the agent with a given task and stream the response. async for message in countdown_agent.on_messages_stream([], CancellationToken()): if isinstance(message, Response): print(message.chat_message) else: print(message)在 Jupyter 中直接await run_countdown_agent()写成脚本时按文档提示改用asyncio.run(run_countdown_agent())。文档示例的运行输出为3... 2... 1... Done!看到倒数消息后以Response(chat_messageTextMessage(Done!))收尾说明on_messages/on_messages_stream的契约实现正确中间过程消息通过流逐条 yield最终Response含chat_message和inner_messages作为流的最后一项。接入自定义模型客户端Gemini 示例AssistantAgent接收model_client参数使用官方支持的模型客户端。当需要的模型客户端不在支持列表中或想要自定义模型行为时做法是把模型客户端直接封装进自定义智能体。文档示例用 Google Gemini SDK 实现了GeminiAssistantAgent要点是用UnboundedChatCompletionContextautogen_core.model_context维护对话上下文把收到的消息经msg.to_model_message()写入上下文从上下文取出历史调用self._model_client.models.generate_content(...)生成回复用返回的usage_metadata构造RequestUsage把助手回复写回上下文最后 yield 一个携带TextMessage带models_usage的Responseon_reset里await self._model_context.clear()清掉上下文完成重置语义。import os from typing import AsyncGenerator, Sequence from autogen_agentchat.agents import BaseChatAgent from autogen_agentchat.base import Response from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage from autogen_core import CancellationToken from autogen_core.model_context import UnboundedChatCompletionContext from autogen_core.models import AssistantMessage, RequestUsage, UserMessage from google import genai from google.genai import types class GeminiAssistantAgent(BaseChatAgent): def __init__( self, name: str, description: str An agent that provides assistance with ability to use tools., model: str gemini-1.5-flash-002, api_key: str os.environ[GEMINI_API_KEY], system_message: str | None You are a helpful assistant that can respond to messages. Reply with TERMINATE when the task has been completed., ): super().__init__(namename, descriptiondescription) self._model_context UnboundedChatCompletionContext() self._model_client genai.Client(api_keyapi_key) self._system_message system_message self._model model property def produced_message_types(self) - Sequence[type[BaseChatMessage]]: return (TextMessage,) async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) - Response: final_response None async for message in self.on_messages_stream(messages, cancellation_token): if isinstance(message, Response): final_response message if final_response is None: raise AssertionError(The stream should have returned the final result.) return final_response async def on_messages_stream( self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken ) - AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None]: # Add messages to the model context for msg in messages: await self._model_context.add_message(msg.to_model_message()) # Get conversation history history [ (msg.source if hasattr(msg, source) else system) : (msg.content if isinstance(msg.content, str) else ) \n for msg in await self._model_context.get_messages() ] # Generate response using Gemini response self._model_client.models.generate_content( modelself._model, contentsfHistory: {history}\nGiven the history, please provide a response, configtypes.GenerateContentConfig( system_instructionself._system_message, temperature0.3, ), ) # Create usage metadata usage RequestUsage( prompt_tokensresponse.usage_metadata.prompt_token_count, completion_tokensresponse.usage_metadata.candidates_token_count, ) # Add response to model context await self._model_context.add_message(AssistantMessage(contentresponse.text, sourceself.name)) # Yield the final response yield Response( chat_messageTextMessage(contentresponse.text, sourceself.name, models_usageusage), inner_messages[], ) async def on_reset(self, cancellation_token: CancellationToken) - None: Reset the assistant by clearing the model context. await self._model_context.clear()验证方式与前面一致——直接运行并观察流式输出。文档示例中GeminiAssistantAgent(gemini_assistant)回答 What is the capital of New York? 的输出文档示例---------- user ---------- What is the capital of New York? ---------- gemini_assistant ---------- Albany TERMINATE返回的TaskResult中每条消息带models_usage示例为RequestUsage(prompt_tokens46, completion_tokens5)具体数值随响应变化能确认自定义客户端确实产出了带用量统计的响应。文档也说明model、api_key、system_message只是示例参数你可以按所用模型客户端和应用设计提供其它参数。把自定义智能体放进团队继承BaseChatAgent的自定义智能体可以直接作为团队成员使用。文档示例把GeminiAssistantAgent作为评审者与AssistantAgentOpenAI 客户端gpt-4o-mini组成RoundRobinGroupChatfrom autogen_agentchat.agents import AssistantAgent from autogen_agentchat.conditions import TextMentionTermination from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.ui import Console model_client OpenAIChatCompletionClient(modelgpt-4o-mini) # Create the primary agent. primary_agent AssistantAgent( primary, model_clientmodel_client, system_messageYou are a helpful AI assistant., ) # Create a critic agent based on our new GeminiAssistantAgent. gemini_critic_agent GeminiAssistantAgent( gemini_critic, system_messageProvide constructive feedback. Respond with APPROVE to when your feedbacks are addressed., ) # Define a termination condition that stops the task if the critic approves or after 10 messages. termination TextMentionTermination(APPROVE) | MaxMessageTermination(10) # Create a team with the primary and critic agents. team RoundRobinGroupChat([primary_agent, gemini_critic_agent], termination_conditiontermination) await Console(team.run_stream(taskWrite a Haiku poem with 4 lines about the fall season.)) await model_client.close()这里的验证点是终止条件TextMentionTermination(APPROVE) | MaxMessageTermination(10)让任务在评审者回复中出现APPROVE或达到 10 条消息时停止。文档示例运行结果文档示例显示主智能体与评审智能体交替发言两轮后评审方输出含 APPROVE 的反馈最终TaskResult的stop_reason为Text APPROVE mentioned——即按预期条件结束而不是跑满消息上限。OpenAIChatCompletionClient来自autogen_ext.models.openai即autogen-ext[openai]扩展。另一条可选路径如果自定义智能体不做模型调用如算术/规则型智能体文档示例用ArithmeticAgent参与SelectorGroupChat并通过allow_repeated_speakerTrue和自定义selector_prompt控制选择行为以MaxMessageTermination(10)作为终止条件其输出示例中数字由 10 经过乘、加、除、加等若干步变为 25文档示例。可选让自定义智能体可序列化Component 接口如果需要保存/加载智能体配置或分享配置可以让智能体同时继承Component接口来自autogen_core实现声明式格式定义一个 pydanticBaseModel作为配置类如GeminiAssistantAgentConfig含name、description、model、system_message等字段并在类上声明component_config_schema GeminiAssistantAgentConfig类声明变为class GeminiAssistantAgent(BaseChatAgent, Component[GeminiAssistantAgentConfig])实现_from_config(cls, config)classmethod从配置构造实例和_to_config(self)返回配置对象两个方法用dump_component()序列化为 JSON用load_component(config)反序列化实例gemini_assistant GeminiAssistantAgent(gemini_assistant) config gemini_assistant.dump_component() print(config.model_dump_json(indent2)) loaded_agent GeminiAssistantAgent.load_component(config) print(loaded_agent)文档示例输出文档示例中 JSON 的provider字段为__main__.GeminiAssistantAgent。注意跨进程或跨文件加载时应把类变量component_provider_override设置为包含该自定义智能体类的模块全路径例如mypackage.agents.GeminiAssistantAgentload_component依据它来确定如何实例化。验证成功的判据dump_component输出的 JSON 中包含provider、component_type: agent和完整config段load_component返回同一类的新实例。限制与下一步自定义智能体每次调用只应接收新消息、自行维护状态并在on_reset中清掉这些状态模型上下文、自维护的历史等否则重复运行时行为会漂移。智能体名必须是合法 Python 标识符否则构造时直接抛ValueError。流式实现中on_messages_stream的最后 yield 必须是Responseon_messages的默认实现以及文档示例都依赖这一点若流中没有Response示例代码会以AssertionError暴露。文档给出的延伸方向给自定义模型客户端补上函数调用能力参照AssistantAgent的实现与 Google Gemini function calling 文档以及把带声明式配置的自定义智能体打包成包配合 AutoGen Studio 使用。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考