在第二部分中，我们探讨了模块化的设计原则。我们讨论了通过借鉴微服务的有界上下文概念来分解Agent系统的策略，以确定每个子Agent的范围。
我们还暗示了模块化引入了需要深思熟虑的代理与子Agent之间的交互模型。
今天我们将深入探讨请求分派模式，这种模式可以帮助创建可预测的机制，以将请求分派给子Agent，并让这些Agent将结果反馈给分派者。
统一的分派/回调机制
当多个Agent需要在代理系统中协调工作时，你可能会创建出一系列临时的调用和不匹配的数据结构。通过标准化每个Agent如何调用（或分派给）其他Agent，以及这些代理如何响应，你可以减少混乱、错误和维护工作量。一个一致的接口迫使每个Agent在提出请求或返回结果时使用相同的“语言”。
统一接口的动机源于一个现实，即在一个复杂的Agent系统中，一个代理很少能够处理用户请求的所有方面。用户可能会在同一对话中询问跟踪包裹、发起退货和检查保修状态等问题。如果你的系统只是简单地委托给大型语言模型（LLM）选择的子Agent，你需要一种统一的方式来传递请求数据并检索结构化的响应。通过将这些Agent交接视为具有严格模式的函数调用，你可以确保每个Agent，无论是父代理还是子Agent，都以可预测的方式交换信息。
如果没有这种统一性，父Agent可能期望一种数据表示，而子Agent返回的却是完全不同的东西。或者你可能会发现在一个子Agent尝试调用另一个子Agent时出现不匹配。每个小的不一致都可能引发令人困惑的错误，这些错误很难调试，特别是在LLM驱动系统的动态行为下。数据形状和参数名称的一致性是使LLM能够可靠地推理要调用哪个函数以及必须提供哪些数据的关键。
Python示例
父代理需要知道如何正确地将任务委托给每个子Agent。你通过暴露负责特定领域的函数（在向LLM进行函数调用的意义上）来实现这一点。例如：
tools = [ { "type": "function", "function": { "name": "handoff_to_OrdersAgent", "description": "Handles order-related queries such as tracking or managing orders.", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "The unique ID of the user."}, "message": {"type": "string", "description": "The user's query."} }, "required": ["user_id", "message"] } } }, { "type": "function", "function": { "name": "handoff_to_ReturnsAgent", "description": "Handles return-related tasks, such as authorizing or tracking a return.", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "The unique ID of the user."}, "message": {"type": "string", "description": "The user's query."} }, "required": ["user_id", "message"] } } }]当大型语言模型（LLM）决定需要Agent与订单相关的问题时，它可以调用handoff_to_OrdersAgent，并附上必要的参数。然后，父Agent相应地分派请求：
def dispatch_request(self, function_call):fn_name = function_call["name"]arguments = json.loads(function_call["arguments"])if fn_name == "handoff_to_OrdersAgent":result = self.child_agents["OrdersAgent"].process_request(arguments)elif fn_name == "handoff_to_ReturnsAgent":result = self.child_agents["ReturnsAgent"].process_request(arguments)else:result = {"status": "error", "message": f"Unknown function {fn_name}"}return result
这种方法允许父Agent专注于路由，而每个子Agent专注于其特定领域（订单、退货、产品问题等）。
在子Agent内部，你可以定义与其特定任务相关的函数。例如，OrdersAgent 可能会暴露 lookupOrder 或 searchOrders 函数。子Agent自身的推理循环被限制在该领域内，这有助于避免混淆和庞大的提示上下文。
class OrdersAgent:def __init__(self):self.functions = [{"type": "function","function": {"name": "lookupOrder","parameters": {"type": "object","properties": {"order_id": {"type": "string", "description": "The order ID."}},"required": ["order_id"]}}},{"type": "function","function": {"name": "searchOrders","parameters": {"type": "object","properties": {"customer_id": {"type": "string", "description": "The customer ID."}},"required": ["customer_id"]}}}]def process_request(self, payload):self.message_history.append({"role": "user", "content": payload["message"]})for _ in range(3): # Limit recursive callsresponse = self.run_llm_cycle(self.functions)if "function_call" in response:function_call = response["function_call"]result = self.handle_function_call(function_call)if result["status"] == "success":return resultelif result["status"] == "escalate":return {"status": "escalate", "message": result["message"]}else:return {"status": "success", "data": response["content"]}return {"status": "error", "message": "Exceeded reasoning steps"}def handle_function_call(self, function_call):if function_call["name"] == "lookupOrder":return {"status": "success", "data": "Order details found..."}elif function_call["name"] == "searchOrders":return {"status": "success", "data": "Searching orders..."}else:return {"status": "escalate", "message": f"Function {function_call['name']} not supported"}
一旦子Agent完成任务，它会以一致的格式将结果发送回父Agent。这就是回调。然后父Agent可以：
如果一切顺利，将响应传递回用户。
使用另一个子Agent重新尝试请求。
如果系统无法自动处理，则将问题升级给人类Agent。
例如：
response = orders_agent.handle_request(payload)if response["status"] == "success": parent_agent.add_message(role="assistant", content=response["data"])elif response["status"] == "escalate": parent_agent.add_message(role="system", content="OrdersAgent could not complete the request.") # Optionally retry with another agent
在任何现实世界的系统中，某些查询可能会因为一些不可预见的原因而失败——比如API宕机、数据缺失，或者子Agent不支持的功能。当这种情况发生时，子Agent会返回一个“升级”状态：
def handle_function_call(self, function_call): if function_call["name"] == "unsupported_function": return {"status": "escalate", "message": "Unsupported function"}父Agent可以捕获这一状态，并决定是否重试、升级到另一个Agent，或者最终向用户返回错误消息。
展望未来
在第二部分和第三部分之间，我们可以看到Agent系统如何被分解为一系列具有统一通信模型的Agent/子Agent，以协调整个Agent层级中的交互。
然而，这些Agent并非孤立存在。它们需要访问外部工具，尤其是数据。在第四部分中，我们将探讨Agent系统数据检索的细微差别，并研究Agent系统独有的数据需求。
