Agent链式调用原理与实战：从设计模式到性能优化

1次阅读

共计 1428 个字符，预计需要花费 4 分钟才能阅读完成。

在分布式系统中，Agent（智能体）协作常面临三大难题：

调用链路冗长：当业务逻辑涉及多个 Agent 时，传统嵌套调用会导致代码臃肿，例如订单处理需要依次调用库存检查 -> 支付验证 -> 物流调度 Agent
状态管理混乱：每个 Agent 可能修改共享状态，容易引发竞态条件（Race Condition）
错误溯源困难：当链路中某个环节失败时，缺乏统一的错误上下文（Context）传递机制

方案	时延	耦合度	可观测性
直接调用	最低	最高（强依赖）	最难（无中间态）
事件总线（EventBus）	中等	最低（解耦）	中等（需额外埋点）
责任链模式（Chain）	可控	灵活（动态组装）	最佳（天然链路）

from abc import ABC, abstractmethod
from typing import Optional

class AgentHandler(ABC):
    """处理器基类（Handler）"""
    _next_handler: Optional['AgentHandler'] = None

    def set_next(self, handler: 'AgentHandler') -> 'AgentHandler':
        self._next_handler = handler
        return handler  # 支持链式语法：a.set_next(b).set_next(c)

    @abstractmethod
    def handle(self, ctx: dict) -> bool:
        """
        返回 False 则终止链条
        ctx: 传递的上下文字典
        """
        if self._next_handler:
            return self._next_handler.handle(ctx)
        return True

type TimeoutHandler struct {
    next    Handler
    timeout time.Duration
}

func (h *TimeoutHandler) Handle(ctx Context) error {done := make(chan error, 1)
    go func() { done <- h.next.Handle(ctx) }()

    select {
    case err := <-done:
        return err
    case <-time.After(h.timeout):
        ctx.Log("timeout triggered")
        return errors.New("handler timeout")
    }
}