如何用 AutoGen Core 实现 Reflection 反思模式的 coder 与 reviewer 双智能体【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen这篇文章要解决的问题是如何在 AutoGen CorePython 端autogen-core包中用「先由一个 LLM 生成代码、再由另一个 LLM 审查代码」的 Reflection 反思模式搭出一对可循环协作的智能体——coder写代码和 reviewer审代码直到 reviewer 给出 APPROVE 才输出最终结果。实现方式基于 AutoGen Core 的 pub/sub 广播机制两个 agent 通过预定义的消息协议在同一个 topic 上互相发消息。适用前提以下均来自项目文档 installation.mdPython 3.10 或更高版本使用OpenAIChatCompletionClient作为模型客户端需要 OpenAI API Key设置环境变量OPENAI_API_KEY文档中该客户端示例注明 assuming OPENAI_API_KEY is set in the environment见 model-clients.ipynb参考示例使用模型gpt-4o-mini。安装依赖按 installation.md 的要求安装autogen-core和 OpenAI 扩展pip install autogen-core pip install autogen-ext[openai]文档建议先用虚拟环境隔离依赖venv 或 conda可选。定义消息协议Reflection 模式的第一步是先定义两个 agent 之间交换的消息类型。整个模式的流程是应用向 coder agent 发布CodeWritingTaskcoder 生成代码后发布CodeReviewTask给 reviewerreviewer 发布CodeReviewResult回给 coder如果approved为真coder 发布CodeWritingResult结束否则 coder 再发布新的CodeReviewTask循环继续。from dataclasses import dataclass dataclass class CodeWritingTask: task: str dataclass class CodeWritingResult: task: str code: str review: str dataclass class CodeReviewTask: session_id: str code_writing_task: str code_writing_scratchpad: str code: str dataclass class CodeReviewResult: review: str session_id: str approved: boolsession_id用于把同一任务的多次往返串起来coder 会为每个新任务生成一个uuid4并以此为 key 保存会话记忆使不同任务各自拥有独立的历史消息。实现 CoderAgentcoder agent 订阅CodeWritingTask和CodeReviewResult发布CodeReviewTask和CodeWritingResult。它的关键点系统提示词采用 chain-of-thought 风格要求把最终代码放在单个 Markdown 代码块中用Thoughts: ... Code: ...格式回复用re.search(r(\w)\n(.*?)\n, ...)从回复中提取代码块提取不到就抛ValueError(Code block not found.)收到CodeReviewResult且未通过时把本会话的全部消息历史任务、审查结果、草稿转成LLMMessage列表重新调用模型生成修订版代码再发一轮审查。import json import re import uuid from typing import Dict, List, Union from autogen_core import MessageContext, RoutedAgent, TopicId, default_subscription, message_handler from autogen_core.models import ( AssistantMessage, ChatCompletionClient, LLMMessage, SystemMessage, UserMessage, ) default_subscription class CoderAgent(RoutedAgent): An agent that performs code writing tasks. def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A code writing agent.) self._system_messages: List[LLMMessage] [ SystemMessage( contentYou are a proficient coder. You write code to solve problems. Work with the reviewer to improve your code. Always put all finished code in a single Markdown code block. For example: python def hello_world(): print(Hello, World!)Respond using the following format:Thoughts: Code: , ) ] self._model_client model_client self._session_memory: Dict[str, List[CodeWritingTask | CodeReviewTask | CodeReviewResult]] {}message_handler async def handle_code_writing_task(self, message: CodeWritingTask, ctx: MessageContext) - None: # Store the messages in a temporary memory for this request only. session_id str(uuid.uuid4()) self._session_memory.setdefault(session_id, []).append(message) # Generate a response using the chat completion API. response await self._model_client.create( self._system_messages [UserMessage(contentmessage.task, sourceself.metadata[type])], cancellation_tokenctx.cancellation_token, ) assert isinstance(response.content, str) # Extract the code block from the response. code_block self._extract_code_block(response.content) if code_block is None: raise ValueError(Code block not found.) # Create a code review task. code_review_task CodeReviewTask( session_idsession_id, code_writing_taskmessage.task, code_writing_scratchpadresponse.content, codecode_block, ) # Store the code review task in the session memory. self._session_memory[session_id].append(code_review_task) # Publish a code review task. await self.publish_message(code_review_task, topic_idTopicId(default, self.id.key)) message_handler async def handle_code_review_result(self, message: CodeReviewResult, ctx: MessageContext) - None: # Store the review result in the session memory. self._session_memory[message.session_id].append(message) # Obtain the request from previous messages. review_request next( m for m in reversed(self._session_memory[message.session_id]) if isinstance(m, CodeReviewTask) ) assert review_request is not None # Check if the code is approved. if message.approved: # Publish the code writing result. await self.publish_message( CodeWritingResult( codereview_request.code, taskreview_request.code_writing_task, reviewmessage.review, ), topic_idTopicId(default, self.id.key), ) print(Code Writing Result:) print(- * 80) print(fTask:\n{review_request.code_writing_task}) print(- * 80) print(fCode:\n{review_request.code}) print(- * 80) print(fReview:\n{message.review}) print(- * 80) else: # Create a list of LLM messages to send to the model. messages: List[LLMMessage] [*self._system_messages] for m in self._session_memory[message.session_id]: if isinstance(m, CodeReviewResult): messages.append(UserMessage(contentm.review, sourceReviewer)) elif isinstance(m, CodeReviewTask): messages.append(AssistantMessage(contentm.code_writing_scratchpad, sourceCoder)) elif isinstance(m, CodeWritingTask): messages.append(UserMessage(contentm.task, sourceUser)) else: raise ValueError(fUnexpected message type: {m}) # Generate a revision using the chat completion API. response await self._model_client.create(messages, cancellation_tokenctx.cancellation_token) assert isinstance(response.content, str) # Extract the code block from the response. code_block self._extract_code_block(response.content) if code_block is None: raise ValueError(Code block not found.) # Create a new code review task. code_review_task CodeReviewTask( session_idmessage.session_id, code_writing_taskreview_request.code_writing_task, code_writing_scratchpadresponse.content, codecode_block, ) # Store the code review task in the session memory. self._session_memory[message.session_id].append(code_review_task) # Publish a new code review task. await self.publish_message(code_review_task, topic_idTopicId(default, self.id.key)) def _extract_code_block(self, markdown_text: str) - Union[str, None]: pattern r(\w)\n(.*?)\n # Search for the pattern in the markdown text match re.search(pattern, markdown_text, re.DOTALL) # Extract the language and code block if a match is found if match: return match.group(2) return None## 实现 ReviewerAgent reviewer agent 只订阅 CodeReviewTask、发布 CodeReviewResult。与 coder 的两个差异值得注意 - 调用模型时传 json_outputTrue系统提示词要求以固定 JSON 结构输出包含 correctness、efficiency、safety、approvalAPPROVE 或 REVISE、suggested_changes 五个字段 - 解析返回的 JSON 后用 review[approval].lower().strip() approve 判定是否通过并把各字段拼成 review 文本写入 CodeReviewResult。 python default_subscription class ReviewerAgent(RoutedAgent): An agent that performs code review tasks. def __init__(self, model_client: ChatCompletionClient) - None: super().__init__(A code reviewer agent.) self._system_messages: List[LLMMessage] [ SystemMessage( contentYou are a code reviewer. You focus on correctness, efficiency and safety of the code. Respond using the following JSON format: { correctness: Your comments, efficiency: Your comments, safety: Your comments, approval: APPROVE or REVISE, suggested_changes: Your comments } , ) ] self._session_memory: Dict[str, List[CodeReviewTask | CodeReviewResult]] {} self._model_client model_client message_handler async def handle_code_review_task(self, message: CodeReviewTask, ctx: MessageContext) - None: # Format the prompt for the code review. # Gather the previous feedback if available. previous_feedback if message.session_id in self._session_memory: previous_review next( (m for m in reversed(self._session_memory[message.session_id]) if isinstance(m, CodeReviewResult)), None, ) if previous_review is not None: previous_feedback previous_review.review # Store the messages in a temporary memory for this request only. self._session_memory.setdefault(message.session_id, []).append(message) prompt fThe problem statement is: {message.code_writing_task} The code is:{message.code}Previous feedback: {previous_feedback} Please review the code. If previous feedback was provided, see if it was addressed. # Generate a response using the chat completion API. response await self._model_client.create( self._system_messages [UserMessage(contentprompt, sourceself.metadata[type])], cancellation_tokenctx.cancellation_token, json_outputTrue, ) assert isinstance(response.content, str) # TODO: use structured generation library e.g. guidance to ensure the response is in the expected format. # Parse the response JSON. review json.loads(response.content) # Construct the review text. review_text Code review:\n \n.join([f{k}: {v} for k, v in review.items()]) approved review[approval].lower().strip() approve result CodeReviewResult( reviewreview_text, session_idmessage.session_id, approvedapproved, ) # Store the review result in the session memory. self._session_memory[message.session_id].append(result) # Publish the review result. await self.publish_message(result, topic_idTopicId(default, self.id.key))运行注册、发布任务、等待空闲由于两个 agent 都带有default_subscription装饰器创建时会自动订阅默认 topic因此只需向默认 topic 发布一条CodeWritingTask就能启动整个反思循环。参考 reflection.ipynb运行部分如下from autogen_core import DefaultTopicId, SingleThreadedAgentRuntime from autogen_ext.models.openai import OpenAIChatCompletionClient runtime SingleThreadedAgentRuntime() model_client OpenAIChatCompletionClient(modelgpt-4o-mini) await ReviewerAgent.register(runtime, ReviewerAgent, lambda: ReviewerAgent(model_clientmodel_client)) await CoderAgent.register(runtime, CoderAgent, lambda: CoderAgent(model_clientmodel_client)) runtime.start() await runtime.publish_message( messageCodeWritingTask(taskWrite a function to find the sum of all even numbers in a list.), topic_idDefaultTopicId(), ) # Keep processing messages until idle. await runtime.stop_when_idle() # Close the model client. await model_client.close()注意原文档是 Jupyter notebookreflection.ipynb上面的顶层await依赖 notebook 的交互式执行环境在 notebook 中可直接运行task字符串就是你要交给 coder 的编程任务可替换为自己的任务描述。SingleThreadedAgentRuntime是文档推荐的本地嵌入式运行时stop_when_idle()会阻塞到运行时空闲即所有消息处理完毕。如果需要观察两个 agent 之间的消息往来按文档打开日志import logging logging.basicConfig(levellogging.WARNING) logging.getLogger(autogen_core).setLevel(logging.DEBUG)结果验证如何判断一次反思循环完成参考文档给出的示例结果文档示例实际任务的代码和 review 内容会不同任务「Write a function to find the sum of all even numbers in a list.」经过两轮往返后reviewer 第一次返回approval: REVISE建议补充输入类型校验coder 修订后第二次返回approval: APPROVE最终 stdout 打印出Code Writing Result:分节块依次是 Task、Code、Review 三段Code Writing Result: -------------------------------------------------------------------------------- Task: Write a function to find the sum of all even numbers in a list. -------------------------------------------------------------------------------- Code: def sum_of_even_numbers(numbers): if not isinstance(numbers, list) or not all(isinstance(num, int) for num in numbers): raise ValueError(Input must be a list of integers) return sum(num for num in numbers if num % 2 0) -------------------------------------------------------------------------------- Review: Code review: correctness: The function correctly sums all even numbers in the provided list. It raises a ValueError if the input is not a list of integers, which is a necessary check for correctness. efficiency: The function remains efficient with a time complexity of O(n) due to the use of a generator expression. There are no unnecessary intermediate lists created, so memory usage is optimal. safety: The function includes input validation, which enhances safety by preventing incorrect input types. It raises a ValueError for invalid inputs, making the function more robust against unexpected data. approval: APPROVE suggested_changes: No further changes are necessary as the previous feedback has been adequately addressed. --------------------------------------------------------------------------------打开日志后还能在autogen_core的 INFO 日志中核对完整消息流CodeWritingTask→CodeReviewTask→CodeReviewResult(approvedFalse)→CodeReviewTask→CodeReviewResult(approvedTrue)→CodeWritingResult其中LLMCall事件会记录每次模型调用的prompt_tokens/completion_tokens。示例日志中还有一条Unhandled message: CodeWritingTask(...)ReviewerAgent 收到 coder 的任务消息但无对应 handler这是广播模式下所有 agent 都会收到 topic 上全部消息、由各 agent 自行忽略不处理消息的正常现象参考文档中如实保留了这条日志。限制与后续改进点reviewer 依赖 JSON 模式json_outputTrue解析审查结果。模型必须支持 json_output 能力否则json.loads可能失败。文档中也留了一个 TODO「use structured generation library e.g. guidance to ensure the response is in the expected format」即当前示例的 JSON 解析尚未做结构化生成层面的保证。参考实现的停止条件是 reviewer 批准approvedTrue。文档在开头说明 Reflection 模式可以用「最大迭代次数或第二个 agent 的批准」作为停止条件本示例只实现了批准这一条如果要防止双方僵持不下需要自行加最大轮数约束。会话记忆存在 agent 实例的_session_memory字典中按session_id隔离进程内有效没有持久化。消息协议的细节广播、message_handler、topic 订阅可参考 message-and-communication.ipynb日志机制参考 logging.md完整可运行 notebook 见 reflection.ipynb。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考