返回 思路一：定制
Autogen 进阶：Agent 如何动态选择并调用其他 Agent？
Ai
2025-04-14 22:05:32
Autogen 进阶：让 Agent 动态决定和谁“唠嗑”
咱们用 Autogen 搭多 Agent 系统的时候，经常遇到一个场景：一个主 Agent 在和用户聊着，聊到一半，发现，“哎呀，这事儿我一个人搞不定，得找个帮手问问”。比如，用户让主 Agent 写份报告，主 Agent 可能需要先让一个“调研员” Agent 去网上搜集资料，再让一个“图表师” Agent 画个图。
问题来了：主 Agent 怎么能在对话过程中，自己判断 啥时候该去找哪个帮手，并且跟帮手单独聊几句（可能是多轮对话），拿到结果后，再回来继续跟用户汇报？主 Agent 需要动态地决定下一条消息是发给用户，还是发给某个特定的帮手 Agent。
直接用 GroupChat 或者 Nested Chat 好像不太好直接解决这个“动态决策”的问题。GroupChat 的发言顺序要么固定，要么依赖预设的 speaker_selection_func，不够灵活。Nested Chat 虽然能处理子任务，但怎么让主 Agent 根据当前对话内容主动、临时起意 去触发一个嵌套对话，需要点儿技巧。
为啥这事儿有点绕？
Autogen 提供的基础组件，比如 AssistantAgent 和 UserProxyAgent，它们之间的交互通常是点对点的 (initiate_chat) 或者在预定义的群聊结构（GroupChat）里。
initiate_chat: 这个方法通常启动一个固定的、双向的对话。在main_agent和user_proxy的对话中，如果main_agent想临时找research_helper聊，直接在当前的对话循环里调用main_agent.initiate_chat(research_helper, ...)会开启一个全新的对话流程，不容易将结果自然地融入当前对话。GroupChat: 群聊管理器 (GroupChatManager) 控制着发言顺序。虽然可以自定义speaker_selection_func，但这个函数通常基于消息历史和 Agent 状态做决策，实现一个能精确理解“我（main_agent）现在需要暂停，找research_helper问个具体问题，拿到答案再回来”的逻辑，会比较复杂，而且可能污染群聊的整体对话流程。Nested Chats: 嵌套对话允许 Agent 在处理一个消息时，启动一个子对话来解决特定问题。这听起来很接近，但标准用法通常是响应方（比如UserProxyAgent或GroupChatManager）来触发，或者通过 function call 触发。如何让main_agent在生成回复的过程中，自主决定并执行一个对helper_agent的嵌套（或者说临时的）1:1 对话，是关键。
咱们的目标是让 main_agent 具备这种临场判断和“开小会”的能力。
解决方案：整活儿开始！
有几种思路可以实现这种动态路由：
思路一：定制 generate_reply 方法
这是最直接的方式。AssistantAgent 的核心是它生成回复的能力，这个能力主要由 generate_reply 方法提供（或者通过注册的 reply_func）。咱们可以重写或者扩展这个方法，在里面加入决策逻辑。
原理：
当轮到 main_agent 发言时，它的 generate_reply 方法会被调用。咱们在这个方法里：
- 分析收到的消息和当前的对话历史。
- 判断是否需要向某个
helper_agent请求信息。这个判断逻辑可以基于关键词、意图识别，甚至可以调用一个小型的 LLM 来做决策。 - 如果需要求助，就暂停对原始消息的回复。
- 直接调用 目标
helper_agent的generate_reply方法（或者更稳妥地，使用initiate_chat发起一个临时的、只有两方（main_agent和helper_agent）参与的短对话，并等待其完成）。这里要注意管理对话状态，避免混淆。 - 获取
helper_agent的回复（或者临时对话的结果）。 - 将这个结果整合到
main_agent最终要回复给原始发送者（比如user_proxy）的消息中。 - 返回整合后的最终回复。
代码示例：
这里展示一个简化的概念，通过注册自定义回复函数来实现。
import autogen
from typing import Dict, List, Optional, Union
# 假设已经配置好了 LLM
config_list = autogen.config_list_from_json("OAI_CONFIG_LIST")
llm_config = {"config_list": config_list, "cache_seed": 42}
# 定义 Agent
user_proxy = autogen.UserProxyAgent(
name="UserProxy",
human_input_mode="NEVER", # 改为 TERMINATE 或 ALWAYS 进行真实交互
max_consecutive_auto_reply=10,
is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
code_execution_config=False,
)
research_helper = autogen.AssistantAgent(
name="ResearchHelper",
llm_config=llm_config,
system_message="You are a web research expert. Given a query, find relevant information online and return a concise summary.",
)
diagram_helper = autogen.AssistantAgent(
name="DiagramHelper",
llm_config=llm_config,
system_message="You are a diagramming expert. Given data or description, create mermaid syntax for a diagram.",
)
# 主 Agent，带有自定义决策逻辑
class DynamicRoutingAssistantAgent(autogen.AssistantAgent):
def __init__(self, name, llm_config, research_agent, diagram_agent, **kwargs):
super().__init__(name, llm_config=llm_config, **kwargs)
self.research_agent = research_agent
self.diagram_agent = diagram_agent
# 注册自定义的回复生成函数
self.register_reply([autogen.Agent, None], DynamicRoutingAssistantAgent.generate_dynamic_reply)
def generate_dynamic_reply(
self,
messages: Optional[List[Dict]] = None,
sender: Optional[autogen.Agent] = None,
config: Optional[any] = None,
) -> Union[str, Dict, None]:
# 获取最后一条消息内容，用于决策
last_message = messages[-1].get("content", "").lower()
print(f"\n>> {self.name} received message from {sender.name}:\n{last_message}\n")
# 决策逻辑：是否需要调用 helper?
if "research" in last_message or "find information" in last_message:
print(f"--- {self.name} decides to talk to {self.research_agent.name} ---")
# 发起一个临时的 1:1 对话
# 为了简化，这里只做一次调用获取回复，实际可能需要多轮
# 注意：直接调用 generate_reply 可能不够鲁棒，更好的方式是用 initiate_chat 管理一个子对话
# 这里用一个模拟的交互
query_for_researcher = f"Based on the request: '{last_message}', perform the research task."
# **** * 实际实现中，推荐使用 client.initiate_chat 或 agent.initiate_chat 进行交互 **** *
# 以下为简化模拟：
print(f"--- {self.name} sending to {self.research_agent.name}: {query_for_researcher} ---")
# 这里应该有一个机制来真正调用 research_agent 并获取回复
# reply_from_researcher = self.research_agent.generate_reply(messages=[{"role":"user", "content": query_for_researcher}], sender=self)
# 模拟回复
reply_from_researcher = {"content": f"Research result: Found info about '{last_message[-20:]}'..."} # 模拟结果
print(f"--- {self.name} received from {self.research_agent.name}: {reply_from_researcher.get('content','')} ---")
# 构造给原始发送者的回复
final_reply = f"I consulted the Research Helper and found this:\n{reply_from_researcher.get('content', 'No result.')}\nIs there anything else?"
return final_reply
elif "diagram" in last_message or "draw a chart" in last_message:
print(f"--- {self.name} decides to talk to {self.diagram_agent.name} ---")
# 同样，发起临时对话
query_for_diagrammer = f"Create a diagram based on this request: '{last_message}'"
# **** * 同样建议用 initiate_chat **** *
# 模拟交互：
print(f"--- {self.name} sending to {self.diagram_agent.name}: {query_for_diagrammer} ---")
# 模拟回复
reply_from_diagrammer = {"content": "```mermaid\ngraph TD;\nA-->B;\n```"} # 模拟 Mermaid 代码
print(f"--- {self.name} received from {self.diagram_agent.name}: {reply_from_diagrammer.get('content','')} ---")
final_reply = f"Ok, I asked the Diagram Helper to create a diagram:\n{reply_from_diagrammer.get('content', 'Failed to create diagram.')}"
return final_reply
else:
# 不需要求助，正常回复
print(f"--- {self.name} generates reply normally ---")
# 调用父类的原始 generate_reply 或默认的 LLM 调用
# 如果注册了自定义函数，这里需要手动调用 OAIWrapper 或类似工具
# return super().generate_reply(messages=messages, sender=sender, config=config) # 如果是继承重写方式
# 如果是注册方式，需要自己完成 LLM 调用
client = self._clients[0] # 假设使用第一个 client
response = client.create(messages=self._oai_messages(messages))
extracted_response = client.extract_text_or_completion_object(response)[0]
return extracted_response
main_agent = DynamicRoutingAssistantAgent(
name="MainAssistant",
llm_config=llm_config,
system_message="You are a helpful assistant. You can coordinate with a ResearchHelper and a DiagramHelper.",
research_agent=research_helper,
diagram_agent=diagram_helper,
)
# 启动对话
user_proxy.initiate_chat(
main_agent,
message="Please research the topic 'AutoGen framework' and then draw a diagram of its basic components.",
)
进阶技巧与注意事项：
- 真正的 1:1 临时对话： 上面的代码用了模拟交互。实际中，在
generate_dynamic_reply内部发起一个真正的initiate_chat会更健壮，它能处理多轮对话。你需要创建一个临时的UserProxyAgent（或者让main_agent自己扮演临时 user proxy 的角色）来与helper_agent对话，捕获对话的最终结果。 - 状态管理： 要小心，避免无限循环（比如 main_agent 不断地去问 helper，helper 又触发 main_agent）。可以设置最大求助次数，或者在消息中加入标记，指明当前是主流程对话还是求助子流程。
- 错误处理： 如果
helper_agent求助失败，main_agent需要有备用策略，比如告知用户无法完成请求的部分。 - 上下文传递： 给
helper_agent的请求需要包含足够的上下文信息，这可能需要从主对话历史中提取相关部分。 - 性能： 每次决策和可能的子对话都会增加延迟和成本。对于简单的决策逻辑，可以直接写 Python 代码；复杂的判断可以交给 LLM，但这会更慢、更贵。
思路二：利用 Function Calling
如果底层的 LLM 支持 Function Calling（比如 OpenAI 的 GPT 模型），这可能是更优雅的方案。
原理：
- 为
main_agent定义几个 Python 函数，每个函数代表一种与helper_agent交互的操作，例如ask_research_helper(query: str)和ask_diagram_helper(description: str)。 - 将这些函数注册到
main_agent。这通常涉及到在llm_config中配置functions或tools，并提供函数的 Pydantic 模型或 JSON Schema。 - 修改
main_agent的系统提示（System Prompt），告知它拥有这些“工具”（即调用 helper 的函数），并鼓励它在需要时使用。例如：“你可以调用ask_research_helper来获取信息，调用ask_diagram_helper来创建图表。” - 当轮到
main_agent回复时，如果 LLM 判断需要调用某个 helper，它不会直接生成文本回复，而是生成一个特殊的“函数调用”请求。 - Autogen 框架（或者你自己写的包装逻辑）捕获这个函数调用请求。
- 执行对应的 Python 函数。这个函数内部会负责与相应的
helper_agent进行交互（比如通过initiate_chat发起临时对话），并返回结果。 - 将函数的执行结果（即
helper_agent的回复）传回给 LLM。 - LLM 拿到函数结果后，再生成最终的、面向用户的文本回复，其中包含了从 helper 获取的信息。
代码示例（概念）：
import autogen
import json
# ... (LLM config, user_proxy, helper agents defined as before) ...
# 定义代表与 Helper 交互的函数
def ask_research_helper(query: str) -> str:
"""Asks the ResearchHelper agent to find information about a given query."""
print(f"--- Function call: ask_research_helper(query='{query}') ---")
# 这里发起与 research_helper 的临时对话
# 使用一个新的 UserProxyAgent 来管理这个临时对话，或者简化处理
temp_chat_manager = autogen.UserProxyAgent(name="TempChatManager", human_input_mode="NEVER", max_consecutive_auto_reply=1) # 只期望 helper 回复一次
temp_chat_manager.initiate_chat(
research_helper,
message=f"Please research the following topic: {query}",
max_turns=2 # 限制对话轮次
)
# 假设最后一条消息是 research_helper 的回复
# 注意：真实的实现需要更精细地获取和处理对话历史中的结果
result = temp_chat_manager.last_message(research_helper)["content"]
print(f"--- ResearchHelper response: {result} ---")
return result if result else "Sorry, I couldn't get the information."
def ask_diagram_helper(description: str) -> str:
"""Asks the DiagramHelper agent to create a diagram based on a description."""
print(f"--- Function call: ask_diagram_helper(description='{description[:30]}...') ---")
# 同理，与 diagram_helper 交互
temp_chat_manager = autogen.UserProxyAgent(name="TempChatManager", human_input_mode="NEVER", max_consecutive_auto_reply=1)
temp_chat_manager.initiate_chat(
diagram_helper,
message=f"Please create a diagram for: {description}",
max_turns=2
)
result = temp_chat_manager.last_message(diagram_helper)["content"]
# 通常 diagram helper 返回代码块，这里直接返回
print(f"--- DiagramHelper response received ---")
return result if result else "Sorry, I couldn't create the diagram."
# 配置 main_agent 使用 function calling
llm_config_main = {
"config_list": config_list,
"cache_seed": 43, # Use a different seed if needed
# 重要：指定工具 (OpenAI "tools" 格式)
"tools": [
{
"type": "function",
"function": {
"name": "ask_research_helper",
"description": "Asks the ResearchHelper agent to find information about a given query.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The topic or question to research.",
}
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "ask_diagram_helper",
"description": "Asks the DiagramHelper agent to create a diagram based on a description.",
"parameters": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "The description of what the diagram should represent.",
}
},
"required": ["description"],
},
},
}
]
# 如果使用旧版 OpenAI API 或不同模型，可能是 "functions" 字段
# "functions": [...]
}
main_agent_fc = autogen.AssistantAgent(
name="MainAssistantFC",
system_message="You are a helpful assistant. You must use the provided tools 'ask_research_helper' for research tasks and 'ask_diagram_helper' for diagram requests. Respond to the user after getting results from the tools.",
llm_config=llm_config_main,
)
# 将 Function Calling 的执行逻辑注册给 UserProxyAgent 或 MainAssistantFC
# 通常，发起对话的 Agent（这里是 UserProxyAgent）需要知道如何执行这些函数调用
# 我们这里让 UserProxyAgent 能执行 MainAssistantFC 可能请求的函数
user_proxy.register_function(
function_map={
"ask_research_helper": ask_research_helper,
"ask_diagram_helper": ask_diagram_helper,
}
)
# 启动对话
user_proxy.initiate_chat(
main_agent_fc,
message="Please research the topic 'AutoGen framework components' and then draw a simple flowchart.",
)
进阶技巧与注意事项：
- Prompt Engineering:
main_agent的 System Prompt 很关键。要清晰地告诉 LLM 它有哪些工具，以及何时应该使用它们。 - Function 定义: 函数的（description）和参数说明要准确，这直接影响 LLM 是否能正确调用。
- 执行函数： 捕获并执行函数调用的逻辑通常放在能接收 LLM 回复的 Agent 里，比如
UserProxyAgent（因为它经常需要处理代码执行或函数调用结果），或者AssistantAgent自己也可以配置成能执行函数。上面的例子是注册在UserProxyAgent里。 - 多轮 Helper 对话：
ask_research_helper或ask_diagram_helper函数内部的initiate_chat可以设置为允许多于2轮的对话 (max_turns)，以处理与 helper 之间更复杂的交互。函数需要能够处理这种多轮对话，并汇总最终结果返回。 - 并发与异步： 如果
main_agent需要同时咨询多个 helper，或者 helper 的任务很耗时，可以考虑在执行函数时使用异步方式 (async def和asyncio)，但这会显著增加复杂性。 - 安全建议： 如果 helper agent 对应的函数会执行代码、访问网络或文件系统，务必确保有严格的安全措施，比如在沙箱环境中执行（如 Autogen 的
CodeExecutorDocker 支持），限制权限，对输入参数做校验，防止恶意调用。
选哪种方案？
-
定制
generate_reply:- 优点: 对决策逻辑有完全、显式的控制。如果决策规则比较简单（比如基于关键词），这种方式很直接。不需要依赖 LLM 的 function calling 能力。
- 缺点: 可能会让
generate_reply函数变得臃肿复杂，特别是决策逻辑或与 helper 的交互逻辑复杂时。可维护性可能下降。
-
Function Calling:
- 优点: 结构更清晰，将“调用 helper”视为一种工具。利用了 LLM 的理解和推理能力来决定何时使用工具。更符合 Agent 作为“智能体”的理念。易于扩展，增加新的 helper 只需增加新的函数/工具。
- 缺点: 强依赖 LLM 的 function calling 能力和效果。需要精心设计 Prompt 和函数定义。每次函数调用会增加 LLM 的交互次数，可能影响速度和成本。
一般建议： 如果 LLM 支持并且场景允许，优先考虑 Function Calling 。它更灵活、更具扩展性，并且能更好地利用 LLM 的智能。如果对执行逻辑需要极强的确定性控制，或者 LLM 不支持 function call，定制 generate_reply 是可行的替代方案。
通过这两种方法，就可以让 Autogen 里的 Agent 更加“聪明”，懂得在需要的时候主动找“外援”帮忙了。
