Agentic AI 系统设计：第三部分 Agent 之间的交互

在第二部分中，我们探讨了模块化的设计原则。我们讨论了通过借鉴微服务的有界上下文概念来分解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":"Handlesorder-relatedqueriessuchastrackingormanagingorders.","parameters":{"type":"object","properties":{"user_id":{"type":"string","description":"TheuniqueIDoftheuser."},"message":{"type":"string","description":"Theuser'squery."}},"required":["user_id","message"]}}},{"type":"function","function":{"name":"handoff_to_ReturnsAgent","description":"Handlesreturn-relatedtasks,suchasauthorizingortrackingareturn.","parameters":{"type":"object","properties":{"user_id":{"type":"string","description":"TheuniqueIDoftheuser."},"message":{"type":"string","description":"Theuser'squery."}},"required":["user_id","message"]}}}]

当大型语言模型（LLM）决定需要Agent与订单相关的问题时，它可以调用handoff_to_OrdersAgent，并附上必要的参数。然后，父Agent相应地分派请求：

defdispatch_request(self, function_call):  fn_name = function_call["name"]  arguments = json.loads(function_call["arguments"])
 iffn_name =="handoff_to_OrdersAgent":    result = self.child_agents["OrdersAgent"].process_request(arguments) eliffn_name =="handoff_to_ReturnsAgent":    result = self.child_agents["ReturnsAgent"].process_request(arguments) else:    result = {"status":"error","message":f"Unknown function{fn_name}"}
 returnresult

这种方法允许父Agent专注于路由，而每个子Agent专注于其特定领域（订单、退货、产品问题等）。

在子Agent内部，你可以定义与其特定任务相关的函数。例如，OrdersAgent可能会暴露lookupOrder或searchOrders函数。子Agent自身的推理循环被限制在该领域内，这有助于避免混淆和庞大的提示上下文。

classOrdersAgent: 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_inrange(3): # Limit recursive calls      response = self.run_llm_cycle(self.functions)
     if"function_call"inresponse:        function_call = response["function_call"]        result = self.handle_function_call(function_call)       ifresult["status"] =="success":         returnresult        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):   iffunction_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)ifresponse["status"]=="success":parent_agent.add_message(role="assistant",content=response["data"])elifresponse["status"]=="escalate":parent_agent.add_message(role="system",content="OrdersAgentcouldnotcompletetherequest.")#Optionallyretrywithanotheragent

在任何现实世界的系统中，某些查询可能会因为一些不可预见的原因而失败——比如API宕机、数据缺失，或者子Agent不支持的功能。当这种情况发生时，子Agent会返回一个“升级”状态：

defhandle_function_call(self,function_call):iffunction_call["name"]=="unsupported_function":return{"status":"escalate","message":"Unsupportedfunction"}

父Agent可以捕获这一状态，并决定是否重试、升级到另一个Agent，或者最终向用户返回错误消息。

展望未来

在第二部分和第三部分之间，我们可以看到Agent系统如何被分解为一系列具有统一通信模型的Agent/子Agent，以协调整个Agent层级中的交互。

然而，这些Agent并非孤立存在。它们需要访问外部工具，尤其是数据。在第四部分中，我们将探讨Agent系统数据检索的细微差别，并研究Agent系统独有的数据需求。