在第二部分中，我们探讨了模块化的设计原则。我们讨论了通过借鉴微服务的有界上下文概念来分解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 calls response = 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 result elif 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可以：
例如：
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系统独有的数据需求。
粤ICP备14082021号
免费POC，
零成本试错
