AutoGen Chess Game 示例深度解析用双 Agent 对弈演示工具调用与反思循环【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogenAutoGen本仓库的python/侧实现即 autogen-core / autogen-agentchat 包族提供了一套基于事件与消息驱动的 Agent 运行时。python/samples/core_chess_game是一个极具教学价值的官方示例两个分别执白、执黑的棋手 Agent 借助同一块共享棋盘反复调用查询棋盘 / 列举合法走法 / 落子三类工具完成对弈。本篇指南以该示例为主线结合仓库源码逐步拆解它的运行环境、模型配置、双 Agent 消息拓扑、工具封装方式以及tool_agent_caller_loop背后的调用—执行—反思—再调用循环原理。读完本文你将掌握用 AutoGen Core 编排事件订阅型 Agent 工具执行型 Agent的完整套路并可直接改造该示例复用到游戏、检索、代码执行等需要工具反思的真实场景。1. 示例概览与学习目标python/samples/core_chess_game/目录仅包含三个文件main.py完整示例程序核心逻辑全部集中于此model_config_template.yml模型客户端配置模板README.md官方简短的运行说明。用官方 README 的一句话概括它An example with two chess player agents that executes its own tools to demonstrate tool use and reflection on tool use.——即两个会执行自己工具的棋手 Agent用来演示工具调用及对工具调用结果的反思。从这个示例可以学到三类关键技术点Agent 运行时的发布订阅编程模型两个PlayerAgent都通过default_subscription订阅默认主题default topic谁下完一步就把结果publish_message回默认主题另一方订阅接收后继续思考与落子形成回合制对弈环工具Function/Tool的声明式封装把普通的 Python 函数包装成FunctionTool为黑白双方各注册独立的ToolAgent工具调用循环tool-calling loopPlayerAgent并不直接执行工具而是通过tool_agent_caller_loop与模型的FunctionCall输出联动实现模型请求调用 → 工具 Agent 执行 → 结果回填模型 → 模型继续决策的反思式推理。2. 环境准备与安装依赖示例依赖 autogen-core 官方包、OpenAI/Azure 扩展、国际象棋规则库chess以及 YAML 解析库pyyaml。官方 README 给出的安装命令为pip install autogen-ext[openai,azure] chess pyyaml其中autogen-ext[openai,azure]为示例提供两种模型客户端实现OpenAIChatCompletionClient与AzureOpenAIChatCompletionClient。若希望在仓库内通过源码方式运行并验证实现细节可从仓库根目录按源码安装对应包python/packages/autogen-core与python/packages/autogen-ext分别位于 autogen-core 与 autogen-ext。chess库负责棋盘规则合法走法校验、落子、局面打印pyyaml用于解析模型配置。3. 模型配置从 YAML 到组件加载模型配置需要写在model_config.yml中可直接复制同目录的model_config_template.yml改名使用。模板给出三种主流配置方式OpenAIAPI Keyprovider: autogen_ext.models.openai.OpenAIChatCompletionClient config: model: gpt-4o api_key: REPLACE_WITH_YOUR_API_KEYAzure OpenAIAPI Keyprovider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient config: model: gpt-4o azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ azure_deployment: {your-azure-deployment} api_version: {your-api-version} api_key: REPLACE_WITH_YOUR_API_KEYAzure OpenAIMicrosoft Entra / AD Tokenprovider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient config: model: gpt-4o azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ azure_deployment: {your-azure-deployment} api_version: {your-api-version} azure_ad_token_provider: provider: autogen_ext.auth.azure.AzureTokenProvider config: provider_kind: DefaultAzureCredential scopes: - https://cognitiveservices.azure.com/.default配置文件遵循 AutoGen 的组件化声明component model约定顶层provider指向组件类全限定名config是其构造参数。示例代码通过ChatCompletionClient.load_component(model_config)定义于 autogen_core/models 的组件配置体系中把这份字典实例化为一个具体的模型客户端对象。这意味着换模型只需要换 YAML业务代码零改动——这也解释了为什么示例把模型能力抽象成接口而非直接依赖某个 SDK。4. 运行示例与命令行参数配置好model_config.yml后在示例目录下执行python main.py入口的argparse见 main.py支持两个可选参数--verbose开启后把autogen_core的日志级别调到 DEBUG并写入当前目录的chess_game.log文件便于观察消息流转与工具调用细节--model-config path指定模型配置文件路径默认model_config.yml。程序加载配置后调用asyncio.run(main(model_config))。运行时它会在终端持续打印对局过程每落一子会输出分隔线、当前玩家、UCI 格式走法、模型给出的思考内容以及带棋格边框的 Unicode 棋盘对应make_move内对board.unicode(bordersTrue)的调用。由于游戏在双方都无合法走法时由工具返回 No legal moves. The game is over. 而自然结束可观察SingleThreadedAgentRuntime在无新消息时自动退出。5. 双 Agent 对弈的通信架构示例的对弈结构由main()main.py驱动async def main(model_config: Dict[str, Any]) - None: runtime SingleThreadedAgentRuntime() model_client ChatCompletionClient.load_component(model_config) await chess_game(runtime, model_client) runtime.start() await runtime.send_message( TextMessage(contentGame started, white player your move., sourceSystem), AgentId(PlayerWhite, default), ) await runtime.stop_when_idle() await model_client.close()关键步骤包括创建单线程运行时、加载模型、注册 4 个 Agent、显式给PlayerWhite发开局消息、等待运行时空闲后停止。5.1 为什么用订阅发布而非定向直连实现回合制PlayerAgent类上用default_subscription装饰对应源码实现在 autogen_core/_default_subscription.py它在 Agent 注册时自动为该 Agent 创建一个DefaultSubscription默认订阅 topic type 为default的全局主题。而消息处理器末尾执行await self.publish_message(TextMessage(contentmessages[-1].content, sourceself.id.type), DefaultTopicId())DefaultTopicId()见 autogen_core/_default_topic.py不带参数时默认使用default主题类型若在消息处理上下文内创建其 source 会自动取当前处理者的 agent key否则为default。于是形成闭合回路系统向PlayerWhite发送开局消息白方推理并调用工具落子随后把结果文本publish_message到默认主题同样订阅了默认主题的PlayerBlack收到消息进入自己的思考—调用—落子循环黑方再发布白方接收……循环往复直至终局。这正是 AutoGen Core 推荐的全局事件总线用法Agent 之间解耦只对消息与主题负责新增对局方或旁听观察者只需同样订阅default主题即可无需修改任何对弈方的代码。5.2 回合顺序如何被约束AI 模型并不天然理解回合制示例通过两处硬约束杜绝抢走子工具内部校验validate_turn(board, player)main.py通过board.peek()查看最近一步被谁执行若上一步刚落的是白子却轮黑方调用、或棋盘为空却有非白方先行工具直接抛ValueError参数级人机分权白/黑各拥有一份独立的工具闭包get_legal_moves_white/get_legal_moves_black、make_move_white/make_move_black各自把player参数固定在闭包里模型只能在提示词instructions引导下使用属于自己颜色的工具。工具返回错误例如 It is not your turn to move...后tool_agent_caller_loop会把错误文本作为工具执行结果回喂给模型促使模型修正行为——这正是对工具调用的反思reflection on tool use的典型体现。6. 棋子工具的实现与封装工具层是对python-chess的薄封装四类函数如下均在 main.py函数用途关键行为validate_turn校验轮到谁走依据上一步落子颜色判断get_legal_moves(board, player)列出该方所有合法走法返回 UCI 格式串无走法时返回No legal moves. The game is over.get_board(board)返回当前局面直接str(board)make_move(board, player, thinking, move)落子先校验再Move.from_uci并board.push打印对局过程make_move特意接收一个thinking带Annotated[str, Thinking for the move.]描述参数把模型的内心独白随落子一起打印让开发者直观看到模型为何走出这一步——这是把推理过程暴露给用户的一种轻量方案。工具化封装发生在chess_game()中每个走法/查询函数都用局部闭包把共享的board与固定player绑定后再交给FunctionToolblack_tools: List[Tool] [ FunctionTool(get_legal_moves_black, nameget_legal_moves, descriptionGet legal moves.), FunctionTool(make_move_black, namemake_move, descriptionMake a move.), FunctionTool(get_board_text, nameget_board, descriptionGet the current board state.), ]注意黑白双方工具的名字刻意保持一致get_legal_moves/make_move/get_board只是各自引用不同的闭包实现这样模型提示词可以写成通用模板。FunctionTool的实现位于 autogen_core/tools/_function_tool.py它会通过类型签名自动推导参数模型并生成ToolSchema因此要求被包装函数必须带完整类型注解Annotated[str, ...]中的描述会进入 schema直接影响模型生成参数的质量。注册 Agent 时tool_schema[tool.schema for tool in black_tools]正是把 schema 集合传给PlayerAgent再透传给模型作为可调用工具的声明。7. ToolAgent 与 tool-calling 循环的工作原理7.1 分工PlayerAgent决策者与 ToolAgent执行者示例为黑白双方各注册了两个 AgentPlayerBlack/PlayerWhitePlayerAgent负责推理与PlayerBlackToolAgent/PlayerWhiteToolAgentToolAgent负责执行见 main.pyawait ToolAgent.register( runtime, PlayerBlackToolAgent, lambda: ToolAgent(descriptionTool agent for chess game., toolsblack_tools), ) await PlayerAgent.register( runtime, PlayerBlack, lambda: PlayerAgent( descriptionPlayer playing black., instructionsYou are a chess player and you play as black. Use the tool get_board and get_legal_moves to get the legal moves and make_move to make a move., model_clientmodel_client, model_contextBufferedChatCompletionContext(buffer_size10), tool_schema[tool.schema for tool in black_tools], tool_agent_typePlayerBlackToolAgent, ), )其中PlayerAgent在构造函数里保存tool_agent_type并为其生成同 key 的AgentId(tool_agent_type, self.id.key)——即每个棋手只与自己的工具 Agent 通信白方绝不会调用到黑方的工具执行器。ToolAgent本身的职责非常单一见 autogen_core/tool_agent/_tool_agent.py它的message_handler接收FunctionCall消息按message.name找到对应工具json.loads解析参数后调用tool.run_json(...)最终把结果封装成FunctionExecutionResult返回。找不到工具、参数解析失败、执行抛异常分别对应ToolNotFoundException、InvalidToolArgumentsException、ToolExecutionException三类可捕获异常。这把模型理解层和代码执行层彻底隔离——决策 Agent 无需 import 任何 python-chess 代码工具 Agent 也完全不感知 LLM。7.2 反思循环tool_agent_caller_loopPlayerAgent.handle_messagemain.py的核心只有一段messages await tool_agent_caller_loop( self, tool_agent_idself._tool_agent_id, model_clientself._model_client, input_messagesself._system_messages (await self._model_context.get_messages()), tool_schemaself._tool_schema, cancellation_tokenctx.cancellation_token, )tool_agent_caller_loop定义于 autogen_core/tool_agent/_caller_loop.py它把决策者 / 执行者 / 模型客户端三者串成一个循环调用model_client.create(input_messages, toolstool_schema, ...)让模型基于工具 schema 产出回复若返回内容全部是FunctionCall即模型想调用工具则并发caller.send_message把每个调用发给tool_agent_id指定的工具 Agent取回执行结果把FunctionExecutionResultMessage含各调用结果追加到消息列表再次调用模型——模型看到工具结果后会继续推理要么再发起新一轮工具调用如先看局面、再查合法走法、再落子要么给出最终文本答复循环直到模型的返回不再是纯工具调用此时把整个过程中产生的LLMMessage列表返回给调用方。回到PlayerAgent这段返回的所有消息会被逐条加入BufferedChatCompletionContext缓冲大小为 10即上下文窗口内近似保留最近 10 条消息保证下一轮收到对方走子消息时模型能记得局面演变的关键脉络随后messages[-1].content最终文本如走子汇报被打包为TextMessage发布到默认主题触发对方回合。一次handle_message之内模型可能经历获取棋盘 → 列举走法 → 思考 → 落子多次工具调用这正是示例想演示的对工具结果的反思式迭代。7.3 运行时的细节main()中的两个生命周期方法分别对应 SingleThreadedAgentRuntimeruntime.start()启动消息泵此后投递的消息会被依次调度处理runtime.stop_when_idle()等待当前消息队列清空后停止。由于象棋终局时工具返回游戏结束模型不会再落子、不会产生新消息运行时自然空闲退出。SingleThreadedAgentRuntime意味着所有 Agent 共享同一个事件循环串行处理消息配合共享的Board对象使用是线程安全的若要把这套模式扩展到多进程/多机例如让两位棋手分处不同主机可以替换为分布式GrpcWorkerAgentRuntime系列运行时Agent 与消息代码几乎可原样复用——这也是示例刻意保持Agent 只与消息、主题、工具打交道不持有运行时具体类型的原因。8. 对局中的关键数据流小结把前文串联起来完整的一步棋数据流如下对端 Player 发布 TextMessage │ 默认主题 / 本 Agent 订阅触发 ▼ PlayerAgent.handle_message 1) 把消息加入 BufferedChatCompletionContext 2) 组装 system instructions 会话历史 3) tool_agent_caller_loop 循环 ├─ 模型 create(...) ──► 返回 FunctionCall如 get_board → get_legal_moves → make_move │ │ 每个 FunctionCall 发送给同 key 的 ToolAgent │ ▼ │ ToolAgent 按 name 命中闭包函数 → validate_turn 校验 → 执行 → FunctionExecutionResult │ │ 错误如越序落子以 is_error 文本回喂模型 → 模型反思纠正 │ ▼ │ 结果拼入消息列表 → 再次调用模型 → 直至模型给出最终文本 4) 最终文本 publish_message 到默认主题 │ ▼ 对端 Player 被唤醒重复以上流程游戏结束则不再产生消息9. 把示例改造成自己的场景基于对以上结构与源码的分析迁移这套模式到新场景只需四个步骤定义消息模型仿照TextMessage(BaseModel)定义领域消息替换content语义编写纯函数工具保证所有参数与返回值都有类型注解需要说明意图处用Annotated[...]补充描述然后用FunctionTool包装并赋name/description注册 ToolAgent 决策 Agent决策 Agent 持有model_client、tool_schema、tool_agent_type处理器中调用tool_agent_caller_loop完成决策—执行—反思用instructions明确告知模型可用的工具名与调用顺序用主题连接多方默认default_subscriptionDefaultTopicId()足够支撑单局多方的回合循环当需要隔离对局如同时跑多盘棋时可改用具名 topic type 或带 source 的TopicId避免串场。如果尝试运行该示例建议先从--verbose开始python main.py --verbose随后查看chess_game.log中模型每次FunctionCall的工具名与参数你会清楚看到模型先看盘、再列合法走法、最后落子的工具使用轨迹这正是理解 Agentic 工具反思的最直观素材。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 AutoGen Chess Game 示例深度解析：用双 Agent 对弈演示工具调用与反思循环
AutoGen Chess Game 示例深度解析用双 Agent 对弈演示工具调用与反思循环【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogenAutoGen本仓库的python/侧实现即 autogen-core / autogen-agentchat 包族提供了一套基于事件与消息驱动的 Agent 运行时。python/samples/core_chess_game是一个极具教学价值的官方示例两个分别执白、执黑的棋手 Agent 借助同一块共享棋盘反复调用查询棋盘 / 列举合法走法 / 落子三类工具完成对弈。本篇指南以该示例为主线结合仓库源码逐步拆解它的运行环境、模型配置、双 Agent 消息拓扑、工具封装方式以及tool_agent_caller_loop背后的调用—执行—反思—再调用循环原理。读完本文你将掌握用 AutoGen Core 编排事件订阅型 Agent 工具执行型 Agent的完整套路并可直接改造该示例复用到游戏、检索、代码执行等需要工具反思的真实场景。1. 示例概览与学习目标python/samples/core_chess_game/目录仅包含三个文件main.py完整示例程序核心逻辑全部集中于此model_config_template.yml模型客户端配置模板README.md官方简短的运行说明。用官方 README 的一句话概括它An example with two chess player agents that executes its own tools to demonstrate tool use and reflection on tool use.——即两个会执行自己工具的棋手 Agent用来演示工具调用及对工具调用结果的反思。从这个示例可以学到三类关键技术点Agent 运行时的发布订阅编程模型两个PlayerAgent都通过default_subscription订阅默认主题default topic谁下完一步就把结果publish_message回默认主题另一方订阅接收后继续思考与落子形成回合制对弈环工具Function/Tool的声明式封装把普通的 Python 函数包装成FunctionTool为黑白双方各注册独立的ToolAgent工具调用循环tool-calling loopPlayerAgent并不直接执行工具而是通过tool_agent_caller_loop与模型的FunctionCall输出联动实现模型请求调用 → 工具 Agent 执行 → 结果回填模型 → 模型继续决策的反思式推理。2. 环境准备与安装依赖示例依赖 autogen-core 官方包、OpenAI/Azure 扩展、国际象棋规则库chess以及 YAML 解析库pyyaml。官方 README 给出的安装命令为pip install autogen-ext[openai,azure] chess pyyaml其中autogen-ext[openai,azure]为示例提供两种模型客户端实现OpenAIChatCompletionClient与AzureOpenAIChatCompletionClient。若希望在仓库内通过源码方式运行并验证实现细节可从仓库根目录按源码安装对应包python/packages/autogen-core与python/packages/autogen-ext分别位于 autogen-core 与 autogen-ext。chess库负责棋盘规则合法走法校验、落子、局面打印pyyaml用于解析模型配置。3. 模型配置从 YAML 到组件加载模型配置需要写在model_config.yml中可直接复制同目录的model_config_template.yml改名使用。模板给出三种主流配置方式OpenAIAPI Keyprovider: autogen_ext.models.openai.OpenAIChatCompletionClient config: model: gpt-4o api_key: REPLACE_WITH_YOUR_API_KEYAzure OpenAIAPI Keyprovider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient config: model: gpt-4o azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ azure_deployment: {your-azure-deployment} api_version: {your-api-version} api_key: REPLACE_WITH_YOUR_API_KEYAzure OpenAIMicrosoft Entra / AD Tokenprovider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient config: model: gpt-4o azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ azure_deployment: {your-azure-deployment} api_version: {your-api-version} azure_ad_token_provider: provider: autogen_ext.auth.azure.AzureTokenProvider config: provider_kind: DefaultAzureCredential scopes: - https://cognitiveservices.azure.com/.default配置文件遵循 AutoGen 的组件化声明component model约定顶层provider指向组件类全限定名config是其构造参数。示例代码通过ChatCompletionClient.load_component(model_config)定义于 autogen_core/models 的组件配置体系中把这份字典实例化为一个具体的模型客户端对象。这意味着换模型只需要换 YAML业务代码零改动——这也解释了为什么示例把模型能力抽象成接口而非直接依赖某个 SDK。4. 运行示例与命令行参数配置好model_config.yml后在示例目录下执行python main.py入口的argparse见 main.py支持两个可选参数--verbose开启后把autogen_core的日志级别调到 DEBUG并写入当前目录的chess_game.log文件便于观察消息流转与工具调用细节--model-config path指定模型配置文件路径默认model_config.yml。程序加载配置后调用asyncio.run(main(model_config))。运行时它会在终端持续打印对局过程每落一子会输出分隔线、当前玩家、UCI 格式走法、模型给出的思考内容以及带棋格边框的 Unicode 棋盘对应make_move内对board.unicode(bordersTrue)的调用。由于游戏在双方都无合法走法时由工具返回 No legal moves. The game is over. 而自然结束可观察SingleThreadedAgentRuntime在无新消息时自动退出。5. 双 Agent 对弈的通信架构示例的对弈结构由main()main.py驱动async def main(model_config: Dict[str, Any]) - None: runtime SingleThreadedAgentRuntime() model_client ChatCompletionClient.load_component(model_config) await chess_game(runtime, model_client) runtime.start() await runtime.send_message( TextMessage(contentGame started, white player your move., sourceSystem), AgentId(PlayerWhite, default), ) await runtime.stop_when_idle() await model_client.close()关键步骤包括创建单线程运行时、加载模型、注册 4 个 Agent、显式给PlayerWhite发开局消息、等待运行时空闲后停止。5.1 为什么用订阅发布而非定向直连实现回合制PlayerAgent类上用default_subscription装饰对应源码实现在 autogen_core/_default_subscription.py它在 Agent 注册时自动为该 Agent 创建一个DefaultSubscription默认订阅 topic type 为default的全局主题。而消息处理器末尾执行await self.publish_message(TextMessage(contentmessages[-1].content, sourceself.id.type), DefaultTopicId())DefaultTopicId()见 autogen_core/_default_topic.py不带参数时默认使用default主题类型若在消息处理上下文内创建其 source 会自动取当前处理者的 agent key否则为default。于是形成闭合回路系统向PlayerWhite发送开局消息白方推理并调用工具落子随后把结果文本publish_message到默认主题同样订阅了默认主题的PlayerBlack收到消息进入自己的思考—调用—落子循环黑方再发布白方接收……循环往复直至终局。这正是 AutoGen Core 推荐的全局事件总线用法Agent 之间解耦只对消息与主题负责新增对局方或旁听观察者只需同样订阅default主题即可无需修改任何对弈方的代码。5.2 回合顺序如何被约束AI 模型并不天然理解回合制示例通过两处硬约束杜绝抢走子工具内部校验validate_turn(board, player)main.py通过board.peek()查看最近一步被谁执行若上一步刚落的是白子却轮黑方调用、或棋盘为空却有非白方先行工具直接抛ValueError参数级人机分权白/黑各拥有一份独立的工具闭包get_legal_moves_white/get_legal_moves_black、make_move_white/make_move_black各自把player参数固定在闭包里模型只能在提示词instructions引导下使用属于自己颜色的工具。工具返回错误例如 It is not your turn to move...后tool_agent_caller_loop会把错误文本作为工具执行结果回喂给模型促使模型修正行为——这正是对工具调用的反思reflection on tool use的典型体现。6. 棋子工具的实现与封装工具层是对python-chess的薄封装四类函数如下均在 main.py函数用途关键行为validate_turn校验轮到谁走依据上一步落子颜色判断get_legal_moves(board, player)列出该方所有合法走法返回 UCI 格式串无走法时返回No legal moves. The game is over.get_board(board)返回当前局面直接str(board)make_move(board, player, thinking, move)落子先校验再Move.from_uci并board.push打印对局过程make_move特意接收一个thinking带Annotated[str, Thinking for the move.]描述参数把模型的内心独白随落子一起打印让开发者直观看到模型为何走出这一步——这是把推理过程暴露给用户的一种轻量方案。工具化封装发生在chess_game()中每个走法/查询函数都用局部闭包把共享的board与固定player绑定后再交给FunctionToolblack_tools: List[Tool] [ FunctionTool(get_legal_moves_black, nameget_legal_moves, descriptionGet legal moves.), FunctionTool(make_move_black, namemake_move, descriptionMake a move.), FunctionTool(get_board_text, nameget_board, descriptionGet the current board state.), ]注意黑白双方工具的名字刻意保持一致get_legal_moves/make_move/get_board只是各自引用不同的闭包实现这样模型提示词可以写成通用模板。FunctionTool的实现位于 autogen_core/tools/_function_tool.py它会通过类型签名自动推导参数模型并生成ToolSchema因此要求被包装函数必须带完整类型注解Annotated[str, ...]中的描述会进入 schema直接影响模型生成参数的质量。注册 Agent 时tool_schema[tool.schema for tool in black_tools]正是把 schema 集合传给PlayerAgent再透传给模型作为可调用工具的声明。7. ToolAgent 与 tool-calling 循环的工作原理7.1 分工PlayerAgent决策者与 ToolAgent执行者示例为黑白双方各注册了两个 AgentPlayerBlack/PlayerWhitePlayerAgent负责推理与PlayerBlackToolAgent/PlayerWhiteToolAgentToolAgent负责执行见 main.pyawait ToolAgent.register( runtime, PlayerBlackToolAgent, lambda: ToolAgent(descriptionTool agent for chess game., toolsblack_tools), ) await PlayerAgent.register( runtime, PlayerBlack, lambda: PlayerAgent( descriptionPlayer playing black., instructionsYou are a chess player and you play as black. Use the tool get_board and get_legal_moves to get the legal moves and make_move to make a move., model_clientmodel_client, model_contextBufferedChatCompletionContext(buffer_size10), tool_schema[tool.schema for tool in black_tools], tool_agent_typePlayerBlackToolAgent, ), )其中PlayerAgent在构造函数里保存tool_agent_type并为其生成同 key 的AgentId(tool_agent_type, self.id.key)——即每个棋手只与自己的工具 Agent 通信白方绝不会调用到黑方的工具执行器。ToolAgent本身的职责非常单一见 autogen_core/tool_agent/_tool_agent.py它的message_handler接收FunctionCall消息按message.name找到对应工具json.loads解析参数后调用tool.run_json(...)最终把结果封装成FunctionExecutionResult返回。找不到工具、参数解析失败、执行抛异常分别对应ToolNotFoundException、InvalidToolArgumentsException、ToolExecutionException三类可捕获异常。这把模型理解层和代码执行层彻底隔离——决策 Agent 无需 import 任何 python-chess 代码工具 Agent 也完全不感知 LLM。7.2 反思循环tool_agent_caller_loopPlayerAgent.handle_messagemain.py的核心只有一段messages await tool_agent_caller_loop( self, tool_agent_idself._tool_agent_id, model_clientself._model_client, input_messagesself._system_messages (await self._model_context.get_messages()), tool_schemaself._tool_schema, cancellation_tokenctx.cancellation_token, )tool_agent_caller_loop定义于 autogen_core/tool_agent/_caller_loop.py它把决策者 / 执行者 / 模型客户端三者串成一个循环调用model_client.create(input_messages, toolstool_schema, ...)让模型基于工具 schema 产出回复若返回内容全部是FunctionCall即模型想调用工具则并发caller.send_message把每个调用发给tool_agent_id指定的工具 Agent取回执行结果把FunctionExecutionResultMessage含各调用结果追加到消息列表再次调用模型——模型看到工具结果后会继续推理要么再发起新一轮工具调用如先看局面、再查合法走法、再落子要么给出最终文本答复循环直到模型的返回不再是纯工具调用此时把整个过程中产生的LLMMessage列表返回给调用方。回到PlayerAgent这段返回的所有消息会被逐条加入BufferedChatCompletionContext缓冲大小为 10即上下文窗口内近似保留最近 10 条消息保证下一轮收到对方走子消息时模型能记得局面演变的关键脉络随后messages[-1].content最终文本如走子汇报被打包为TextMessage发布到默认主题触发对方回合。一次handle_message之内模型可能经历获取棋盘 → 列举走法 → 思考 → 落子多次工具调用这正是示例想演示的对工具结果的反思式迭代。7.3 运行时的细节main()中的两个生命周期方法分别对应 SingleThreadedAgentRuntimeruntime.start()启动消息泵此后投递的消息会被依次调度处理runtime.stop_when_idle()等待当前消息队列清空后停止。由于象棋终局时工具返回游戏结束模型不会再落子、不会产生新消息运行时自然空闲退出。SingleThreadedAgentRuntime意味着所有 Agent 共享同一个事件循环串行处理消息配合共享的Board对象使用是线程安全的若要把这套模式扩展到多进程/多机例如让两位棋手分处不同主机可以替换为分布式GrpcWorkerAgentRuntime系列运行时Agent 与消息代码几乎可原样复用——这也是示例刻意保持Agent 只与消息、主题、工具打交道不持有运行时具体类型的原因。8. 对局中的关键数据流小结把前文串联起来完整的一步棋数据流如下对端 Player 发布 TextMessage │ 默认主题 / 本 Agent 订阅触发 ▼ PlayerAgent.handle_message 1) 把消息加入 BufferedChatCompletionContext 2) 组装 system instructions 会话历史 3) tool_agent_caller_loop 循环 ├─ 模型 create(...) ──► 返回 FunctionCall如 get_board → get_legal_moves → make_move │ │ 每个 FunctionCall 发送给同 key 的 ToolAgent │ ▼ │ ToolAgent 按 name 命中闭包函数 → validate_turn 校验 → 执行 → FunctionExecutionResult │ │ 错误如越序落子以 is_error 文本回喂模型 → 模型反思纠正 │ ▼ │ 结果拼入消息列表 → 再次调用模型 → 直至模型给出最终文本 4) 最终文本 publish_message 到默认主题 │ ▼ 对端 Player 被唤醒重复以上流程游戏结束则不再产生消息9. 把示例改造成自己的场景基于对以上结构与源码的分析迁移这套模式到新场景只需四个步骤定义消息模型仿照TextMessage(BaseModel)定义领域消息替换content语义编写纯函数工具保证所有参数与返回值都有类型注解需要说明意图处用Annotated[...]补充描述然后用FunctionTool包装并赋name/description注册 ToolAgent 决策 Agent决策 Agent 持有model_client、tool_schema、tool_agent_type处理器中调用tool_agent_caller_loop完成决策—执行—反思用instructions明确告知模型可用的工具名与调用顺序用主题连接多方默认default_subscriptionDefaultTopicId()足够支撑单局多方的回合循环当需要隔离对局如同时跑多盘棋时可改用具名 topic type 或带 source 的TopicId避免串场。如果尝试运行该示例建议先从--verbose开始python main.py --verbose随后查看chess_game.log中模型每次FunctionCall的工具名与参数你会清楚看到模型先看盘、再列合法走法、最后落子的工具使用轨迹这正是理解 Agentic 工具反思的最直观素材。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考