我要投稿

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

发布日期：2025-01-12 09:06:13 浏览次数： 1734 来源：红薯胡说

在第二部分中，我们探讨了模块化的设计原则。我们讨论了通过借鉴微服务的有界上下文概念来分解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可以：

如果一切顺利，将响应传递回用户。
使用另一个子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系统独有的数据需求。