如何基于 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 SDK：

pip install google-genai

Gemini 示例的构造函数默认读取环境变量GEMINI_API_KEY（api_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(content=f"{i}...", source=self.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_message=TextMessage(content="Done!", source=self.name), inner_messages=inner_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_message=TextMessage("Done!"))收尾，说明on_messages/on_messages_stream的契约实现正确：中间过程消息通过流逐条 yield，最终Response（含chat_message和inner_messages）作为流的最后一项。

接入自定义模型客户端：Gemini 示例

AssistantAgent接收model_client参数，使用官方支持的模型客户端。当需要的模型客户端不在支持列表中，或想要自定义模型行为时，做法是把模型客户端直接封装进自定义智能体。文档示例用 Google Gemini SDK 实现了GeminiAssistantAgent，要点是：

用UnboundedChatCompletionContext（autogen_core.model_context）维护对话上下文，把收到的消息经msg.to_model_message()写入上下文；
从上下文取出历史，调用self._model_client.models.generate_content(...)生成回复；
用返回的usage_metadata构造RequestUsage，把助手回复写回上下文，最后 yield 一个携带TextMessage（带models_usage）的Response；
on_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__(name=name, description=description) self._model_context = UnboundedChatCompletionContext() self._model_client = genai.Client(api_key=api_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( model=self._model, contents=f"History: {history}\nGiven the history, please provide a response", config=types.GenerateContentConfig( system_instruction=self._system_message, temperature=0.3, ), ) # Create usage metadata usage = RequestUsage( prompt_tokens=response.usage_metadata.prompt_token_count, completion_tokens=response.usage_metadata.candidates_token_count, ) # Add response to model context await self._model_context.add_message(AssistantMessage(content=response.text, source=self.name)) # Yield the final response yield Response( chat_message=TextMessage(content=response.text, source=self.name, models_usage=usage), 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_tokens=46, completion_tokens=5)，具体数值随响应变化），能确认自定义客户端确实产出了带用量统计的响应。文档也说明model、api_key、system_message只是示例参数，你可以按所用模型客户端和应用设计提供其它参数。

把自定义智能体放进团队

继承BaseChatAgent的自定义智能体可以直接作为团队成员使用。文档示例把GeminiAssistantAgent作为评审者与AssistantAgent（OpenAI 客户端，gpt-4o-mini）组成RoundRobinGroupChat：

from 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(model="gpt-4o-mini") # Create the primary agent. primary_agent = AssistantAgent( "primary", model_client=model_client, system_message="You are a helpful AI assistant.", ) # Create a critic agent based on our new GeminiAssistantAgent. gemini_critic_agent = GeminiAssistantAgent( "gemini_critic", system_message="Provide 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_condition=termination) await Console(team.run_stream(task="Write 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_speaker=True和自定义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(indent=2)) loaded_agent = GeminiAssistantAgent.load_component(config) print(loaded_agent)

文档示例输出（文档示例）中 JSON 的provider字段为"__main__.GeminiAssistantAgent"。注意：跨进程或跨文件加载时应把类变量component_provider_override设置为包含该自定义智能体类的模块全路径（例如mypackage.agents.GeminiAssistantAgent），load_component依据它来确定如何实例化。验证成功的判据：dump_component输出的 JSON 中包含provider、component_type: "agent"和完整config段，load_component返回同一类的新实例。

限制与下一步

自定义智能体每次调用只应接收新消息、自行维护状态，并在on_reset中清掉这些状态（模型上下文、自维护的历史等），否则重复运行时行为会漂移。
智能体名必须是合法 Python 标识符，否则构造时直接抛ValueError。
流式实现中，on_messages_stream的最后 yield 必须是Response，on_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），仅供参考