AutoGen Chess Game 示例深度解析：用双 Agent 对弈演示工具调用与反思循环

AutoGen Chess Game 示例深度解析：用双 Agent 对弈演示工具调用与反思循环

【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen

AutoGen（本仓库的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，用来演示工具调用及对工具调用结果的反思。从这个示例可以学到三类关键技术点：

1. Agent 运行时的发布订阅编程模型：两个PlayerAgent都通过@default_subscription订阅默认主题（default topic），谁下完一步就把结果publish_message回默认主题，另一方订阅接收后继续思考与落子，形成回合制对弈环；
2. 工具（Function/Tool）的声明式封装：把普通的 Python 函数包装成FunctionTool，为黑白双方各注册独立的ToolAgent；
3. 工具调用循环（tool-calling loop）：PlayerAgent并不直接执行工具，而是通过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改名使用。模板给出三种主流配置方式：

OpenAI（API Key）

provider: autogen_ext.models.openai.OpenAIChatCompletionClient config: model: gpt-4o api_key: REPLACE_WITH_YOUR_API_KEY

Azure OpenAI（API Key）

provider: 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_KEY

Azure OpenAI（Microsoft Entra / AD Token）

provider: 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(borders=True)的调用）。由于游戏在双方都无合法走法时由工具返回 "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(content="Game started, white player your move.", source="System"), 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(content=messages[-1].content, source=self.id.type), DefaultTopicId())

DefaultTopicId()（见 autogen_core/_default_topic.py）不带参数时默认使用"default"主题类型；若在消息处理上下文内创建，其 source 会自动取当前处理者的 agent key，否则为"default"。于是形成闭合回路：

1. 系统向PlayerWhite发送开局消息；
2. 白方推理并调用工具落子，随后把结果文本publish_message到默认主题；
3. 同样订阅了默认主题的PlayerBlack收到消息，进入自己的"思考—调用—落子"循环；
4. 黑方再发布，白方接收……循环往复直至终局。

这正是 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绑定后，再交给FunctionTool：

black_tools: List[Tool] = [ FunctionTool(get_legal_moves_black, name="get_legal_moves", description="Get legal moves."), FunctionTool(make_move_black, name="make_move", description="Make a move."), FunctionTool(get_board_text, name="get_board", description="Get 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（执行者）

示例为黑白双方各注册了两个 Agent：PlayerBlack/PlayerWhite（PlayerAgent，负责推理）与PlayerBlackToolAgent/PlayerWhiteToolAgent（ToolAgent，负责执行），见 main.py：

await ToolAgent.register( runtime, "PlayerBlackToolAgent", lambda: ToolAgent(description="Tool agent for chess game.", tools=black_tools), ) await PlayerAgent.register( runtime, "PlayerBlack", lambda: PlayerAgent( description="Player playing black.", instructions="You 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_client=model_client, model_context=BufferedChatCompletionContext(buffer_size=10), tool_schema=[tool.schema for tool in black_tools], tool_agent_type="PlayerBlackToolAgent", ), )

其中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_loop

PlayerAgent.handle_message（main.py）的核心只有一段：

messages = await tool_agent_caller_loop( self, tool_agent_id=self._tool_agent_id, model_client=self._model_client, input_messages=self._system_messages + (await self._model_context.get_messages()), tool_schema=self._tool_schema, cancellation_token=ctx.cancellation_token, )

tool_agent_caller_loop定义于 autogen_core/tool_agent/_caller_loop.py，它把"决策者 / 执行者 / 模型客户端"三者串成一个循环：

1. 调用model_client.create(input_messages, tools=tool_schema, ...)让模型基于工具 schema 产出回复；
2. 若返回内容全部是FunctionCall（即模型想调用工具），则并发caller.send_message把每个调用发给tool_agent_id指定的工具 Agent，取回执行结果；
3. 把FunctionExecutionResultMessage（含各调用结果）追加到消息列表，再次调用模型——模型看到工具结果后会继续推理：要么再发起新一轮工具调用（如先看局面、再查合法走法、再落子），要么给出最终文本答复；
4. 循环直到模型的返回不再是纯工具调用，此时把整个过程中产生的LLMMessage列表返回给调用方。

回到PlayerAgent，这段返回的所有消息会被逐条加入BufferedChatCompletionContext（缓冲大小为 10，即上下文窗口内近似保留最近 10 条消息），保证下一轮收到对方走子消息时模型能"记得"局面演变的关键脉络；随后messages[-1].content（最终文本，如走子汇报）被打包为TextMessage发布到默认主题，触发对方回合。一次handle_message之内，模型可能经历"获取棋盘 → 列举走法 → 思考 → 落子"多次工具调用，这正是示例想演示的"对工具结果的反思式迭代"。

7.3 运行时的细节

main()中的两个生命周期方法分别对应 SingleThreadedAgentRuntime：

• runtime.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. 把示例改造成自己的场景

基于对以上结构与源码的分析，迁移这套模式到新场景只需四个步骤：

1. 定义消息模型：仿照TextMessage(BaseModel)定义领域消息，替换content语义；
2. 编写纯函数工具：保证所有参数与返回值都有类型注解，需要说明意图处用Annotated[...]补充描述，然后用FunctionTool包装并赋name/description；
3. 注册 ToolAgent + 决策 Agent：决策 Agent 持有model_client、tool_schema、tool_agent_type，处理器中调用tool_agent_caller_loop完成"决策—执行—反思"；用instructions明确告知模型可用的工具名与调用顺序；
4. 用主题连接多方：默认@default_subscription+DefaultTopicId()足够支撑单局多方的回合循环；当需要隔离对局（如同时跑多盘棋）时，可改用具名 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