Popular multi-agent patterns and how we can support in AgentScope.

[chickenlj]

为什么要做 Multi-agen？无论 single agent 还是 multi-agent，本质的核心还是做好上下文管理。那么multi-agent在上下文管理上能带来哪些优势？

当大家聊 multi-agent 时，通常是需要解决认为复杂的问题时。比如我们可能认为multi-agent比较擅长实现以下几个方面：

• 上下文管理。通过 specialized agent 来减小延迟、降低token消耗等；隔离的上下文空间
• 子 agent 可以并行执行。

什么时候需要 multi-agent？

• 一个agent有太多tool时
• 当某些任务需要专业的、特有的知识或者上下文来解决问题时，可能是特定的工具或提示词
• 在一些需要特定条件判断的场景，比如当某个条件满足时，某个子agent定义的能力才被开启

智能体推理模式

1.1 Supervisor（子智能体）模式

核心特点：

• 主智能体（Supervisor）协调子智能体作为工具（Agent As Tool）调用
• 所有路由决策都通过主智能体
• 子智能体不直接与用户交互
• 支持并行执行多个子智能体
• 提供上下文隔离，防止上下文膨胀

适用场景：

• 当智能体处理的问题通常涉及多个不同领域时（如邮件、数据库不同系统）
• 子智能体不需要直接与用户交互
• 需要集中式工作流控制
• 需要并行执行多个任务

// 伪代码示例
@Tool
public String callSubagent(String agentName, String query) {
    ReActAgent subagent = subagentRegistry.get(agentName);
    // 子智能体无状态调用
    Msg result = subagent.invoke(Msg.builder()
        .content(query)
        .build());
    return result.getTextContent();
}

// 主智能体
ReActAgent supervisor = ReActAgent.builder()
    .toolkit(callSubagent, ...)  // 子智能体作为工具
    .build();

import io.agentscope.core.ReActAgent;
import io.agentscope.core.tool.Toolkit;
import io.agentscope.core.model.DashScopeChatModel;

// 创建子智能体的 Provider（工厂）
// 注意：必须使用 lambda 表达式，确保每次调用创建新实例
Toolkit toolkit = new Toolkit();
toolkit.registration()
        .subAgent(() -> ReActAgent.builder()
                .name("Expert")
                .sysPrompt("你是一个领域专家，负责回答专业问题。")
                .model(model)
                .build())
        .apply();

// 创建主智能体，配置工具
ReActAgent mainAgent = ReActAgent.builder()
        .name("Coordinator")
        .sysPrompt("你是一个协调员。当遇到专业问题时，调用 call_expert 工具咨询专家。")
        .model(model)
        .toolkit(toolkit)
        .build();

// 主智能体会在需要时自动调用专家智能体
Msg response = mainAgent.call(userMsg).block();

挑战：

• 子 Agent 是否支持 HITL 或中断
• 通过 ToolWrapper 模式支持传递额外的上下文到 subAgent，需要 Tool 中支持读取 supervisor agent 状态
• 控制 subAgent 到 supervisor Agent 的返回值
• 并行分支的可控性，完全靠模型决策
• 并行分支结果处理策略，等待所有结果处理完成再进入下一步推理？任一分支完成就继续执行？
• 验证最佳实践：子 Agent 不设置状态

Workflow 工作流模式

1.2 Handoffs（交接）模式

核心特点：

• 通过状态变量（如 current_step 或 active_agent）来动态改变Agent行为
• 通常是利用工具来更新状态变量，进而触发智能体转换
• 实现方案：单智能体中可通过Hook、多智能体通过嵌套
• 状态在多轮智能体交互之间持久化
• 每个状态的配置直接处理用户消息

适用场景：

• 需要一些判断逻辑来决策下一步动作的场景（即前一步完成后根据条件决策）
• 智能体需要在不同状态下直接与用户对话
• 构建多步骤的对话 workflow

典型场景：客户服务支持系统（如收集保修信息后再处理退款）

@Tool(name = "record_warranty_status", 
      description = "Record warranty status and transition to next step")
public class RecordWarrantyStatusTool {
    
    public String execute(String status, ToolContext context) {
        // 获取状态
        State state = context.getState();
        SupportState supportState = state.get("support", SupportState.class);
        
        // 更新状态变量
        supportState.warrantyStatus = status;
        supportState.currentStep = "specialist";  // 触发状态转换
        
        // 返回工具响应
        return String.format("Warranty status recorded: %s", status);
    }
}

@Tool(name = "provide_solution", 
      description = "Provide solutions based on warranty status")
public class ProvideSolutionTool {
    
    public String execute(String query, ToolContext context) {
        State state = context.getState();
        SupportState supportState = state.get("support", SupportState.class);
        
        String warrantyStatus = supportState.warrantyStatus;
        if ("in_warranty".equals(warrantyStatus)) {
            return "您的设备在保修期内，我们可以提供免费维修服务。";
        } else {
            return "您的设备已过保修期，我们可以提供付费维修服务。";
        }
    }
}

public class Hook implements Hook {
    public void onEvent(Event event) {
         String step = request.state.get("current_step", "triage")  
    
         Map<String, Config> configs = {
            "triage": {
                "prompt": "Collect warranty information...",
                "tools": [record_warranty_status]
            },
            "specialist": {
                "prompt": "Provide solutions based on warranty: {warranty_status}",
                "tools": [provide_solution, escalate]
            }
        }
    
        // update system prompt and tools 
    }
}

// 1. 定义状态
public class MultiAgentState {
    public String activeAgent;  // "sales_agent" 或 "support_agent"
    public List<Msg> messages = new ArrayList<>();
}

// 2. 创建交接工具
@Tool(name = "transfer_to_sales", 
      description = "Transfer to the sales agent")
public class TransferToSalesTool {
    
    public String execute(String reason, ToolContext context) {
        State state = context.getState();
        MultiAgentState multiState = state.get("multi_agent", MultiAgentState.class);
        
        // 更新活动智能体
        multiState.activeAgent = "sales_agent";
        
        return "Transferred to sales agent from support agent";
    }
}

@Tool(name = "transfer_to_support", 
      description = "Transfer to the support agent")
public class TransferToSupportTool {
    
    public String execute(String reason, ToolContext context) {
        State state = context.getState();
        MultiAgentState multiState = state.get("multi_agent", MultiAgentState.class);
        
        // 更新活动智能体
        multiState.activeAgent = "support_agent";
        
        return "Transferred to support agent from sales agent";
    }
}

ReActAgent salesAgent = ReActAgent.builder()
        .name("sales_agent")
        .sysPrompt("You are a sales agent. Help with sales inquiries. " +
                   "If asked about technical issues or support, transfer to the support agent.")
        .model(model)
        .memory(new InMemoryMemory())
        .tools(new TransferToSupportTool())
        .build();

ReActAgent supportAgent = ReActAgent.builder()
        .name("support_agent")
        .sysPrompt("You are a support agent. Help with technical issues. " +
                   "If asked about pricing or purchasing, transfer to the sales agent.")
        .model(model)
        .memory(new InMemoryMemory())
        .tools(new TransferToSalesTool())
        .build();

// 4. 简单的状态图路由器
public class SimpleWorkflow {
    
    private final Map<String, ReActAgent> agents;
    private final MultiAgentState state;
    
    public SimpleWorkflow(MultiAgentState state, 
                           ReActAgent salesAgent, 
                           ReActAgent supportAgent) {
        this.state = state;
        this.agents = new HashMap<>();
        this.agents.put("sales_agent", salesAgent);
        this.agents.put("support_agent", supportAgent);
    }
    
    // 初始路由：根据状态决定路由到哪个智能体
    private String routeInitial() {
        return state.activeAgent != null ? state.activeAgent : "sales_agent";
    }
    
    // 路由检查：智能体完成后决定下一步
    private String routeAfterAgent() {
        // 如果最后一条消息是 AI 消息且没有工具调用，则结束
        if (!state.messages.isEmpty()) {
            Msg lastMsg = state.messages.get(state.messages.size() - 1);
            if (lastMsg.getRole() == MsgRole.ASSISTANT && 
                (lastMsg.getToolCalls() == null || lastMsg.getToolCalls().isEmpty())) {
                return "__end__";  // 结束
            }
        }
        
        // 否则路由到当前活动智能体
        return state.activeAgent != null ? state.activeAgent : "sales_agent";
    }
    
    public Msg call(Msg userMessage) {
        // 添加用户消息
        state.messages.add(userMessage);
        
        // 初始路由
        String currentNode = routeInitial();
        state.activeAgent = currentNode;
        
        // 循环执行直到完成
        while (true) {
            // 获取当前节点对应的智能体
            ReActAgent currentAgent = agents.get(currentNode);
            
            // 调用智能体（智能体内部会执行工具调用，工具会更新 state.activeAgent）
            Msg response = currentAgent.call(userMessage).block();
            state.messages.add(response);
            
            // 路由决策
            String nextNode = routeAfterAgent();
            
            if ("__end__".equals(nextNode)) {
                return response;  // 完成
            }
            
            // 如果节点改变，继续循环
            if (!nextNode.equals(currentNode)) {
                currentNode = nextNode;
                state.activeAgent = currentNode;
            } else {
                return response;  // 没有变化，完成
            }
        }
    }
}

// 5. 使用状态
MultiAgentState state = new MultiAgentState();
SimpleWorkflow workflow = new SimpleWorkflow(state, salesAgent, supportAgent);

Msg userMsg = Msg.builder()
        .name("user")
        .role(MsgRole.USER)
        .content(TextBlock.builder()
                .text("Hi, I'm having trouble with my account login. Can you help?")
                .build())
        .build();

// 执行：自动路由到 support_agent，智能体可以调用 transfer_to_sales 切换
Msg result = workflow.call(userMsg);

// 输出消息历史
for (Msg msg : state.messages) {
    System.out.println(msg.getName() + ": " + msg.getTextContent());
}

挑战：

1. 单智能体：tool更新全局状态，状态能在tool和agent间流转
2. 多智能体：共享全局状态要自行管理、流程自行控制
3. 多智能体：自己控制上下文传递，full messages/summary?

1.3 Skills（技能）模式

核心特点：

• 将能力定位为可调用的 "技能"
• 主要是提示 Prompt 驱动的设计
• 按需渐进式的给到模型使用（Progressive Disclosure），降低token消耗、拓展规模
• 团队分布式开发和维护：不同团队可独立开发和维护技能
• 通常单智能体模式，skill 提供额外能力，比自定义 ReActAgent 子智能体更轻量

适用场景**：**

• 单个智能体需要多种可用的 “技能”
• 不需要在技能之间强制执行特定约束
• 不同团队需要独立开发 “技能”
• 常见例子：编码助手（不同语言/任务的技能）、知识库（不同领域的技能）

SKILLS: list[Skill] = [
    {
        "name": "sales_analytics",
        "description": "Database schema and business logic for sales data analysis including customers, orders, and revenue.",
        "content": """# Sales Analytics Schema

## Tables

### customers
- customer_id (PRIMARY KEY)
- name
- email
- signup_date
- status (active/inactive)
- customer_tier (bronze/silver/gold/platinum)

### orders
- order_id (PRIMARY KEY)
- customer_id (FOREIGN KEY -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

### order_items
- item_id (PRIMARY KEY)
- order_id (FOREIGN KEY -> orders)
- product_id
- quantity
- unit_price
- discount_percent

## Business Logic

**Active customers**: status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

**Revenue calculation**: Only count orders with status = 'completed'. Use total_amount from orders table, which already accounts for discounts.

**Customer lifetime value (CLV)**: Sum of all completed order amounts for a customer.

**High-value orders**: Orders with total_amount > 1000

## Example Query

-- Get top 10 customers by revenue in the last quarter
SELECT
    c.customer_id,
    c.name,
    c.customer_tier,
    SUM(o.total_amount) as total_revenue
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
  AND o.order_date >= CURRENT_DATE - INTERVAL '3 months'
GROUP BY c.customer_id, c.name, c.customer_tier
ORDER BY total_revenue DESC
LIMIT 10;
""",
    },
    {
        "name": "inventory_management",
        "description": "Database schema and business logic for inventory tracking including products, warehouses, and stock levels.",
        "content": """# Inventory Management Schema

## Tables

### products
- product_id (PRIMARY KEY)
- product_name
- sku
- category
- unit_cost
- reorder_point (minimum stock level before reordering)
- discontinued (boolean)

### warehouses
- warehouse_id (PRIMARY KEY)
- warehouse_name
- location
- capacity

### inventory
- inventory_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- quantity_on_hand
- last_updated

### stock_movements
- movement_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (positive for inbound, negative for outbound)
- movement_date
- reference_number

## Business Logic

**Available stock**: quantity_on_hand from inventory table where quantity_on_hand > 0

**Products needing reorder**: Products where total quantity_on_hand across all warehouses is less than or equal to the product's reorder_point

**Active products only**: Exclude products where discontinued = true unless specifically analyzing discontinued items

**Stock valuation**: quantity_on_hand * unit_cost for each product

## Example Query

-- Find products below reorder point across all warehouses
SELECT
    p.product_id,
    p.product_name,
    p.reorder_point,
    SUM(i.quantity_on_hand) as total_stock,
    p.unit_cost,
    (p.reorder_point - SUM(i.quantity_on_hand)) as units_to_reorder
FROM products p
JOIN inventory i ON p.product_id = i.product_id
WHERE p.discontinued = false
GROUP BY p.product_id, p.product_name, p.reorder_point, p.unit_cost
HAVING SUM(i.quantity_on_hand) <= p.reorder_point
ORDER BY units_to_reorder DESC;
""",
    },
]

public static void main(String[] args) throws Exception {
    // Create toolkit and skillBox, configure skills
    Toolkit toolkit = new Toolkit();
    SkillBox skillBox = new SkillBox(toolkit);
    // 1. Create data analysis skill
    AgentSkill dataSkill = createDataAnalysisSkill();
    // 2. Register skill with data analysis tools to skillBox
    skillBox.registration().tool(new DataAnalysisTools()).skill(dataSkill).apply();

    // Create ReActAgent with data analysis skills
    ReActAgent agent =
            ReActAgent.builder()
                    .name("DataAnalyst")
                    .sysPrompt(buildSystemPrompt())
                    .model(
                            DashScopeChatModel.builder()
                                    .apiKey(apiKey)
                                    .modelName("qwen-max")
                                    .stream(true)
                                    .enableThinking(true)
                                    .formatter(new DashScopeChatFormatter())
                                    .build())
                    .toolkit(toolkit)
                    .skillBox(skillBox) // Automatically registers tools and hook
                    .memory(new InMemoryMemory())
                    .build();

    // Print example queries
    printExampleQueries();

    // Start interactive chat
    ExampleUtils.startChat(agent);
}

1.4 Routing（路由）模式

核心特点：

• 通常适合于回答涉及多个领域的问答，将问题按领域拆解并分派
• 支持单智能体路由或多智能体并行执行
• 汇总结果合
• 支持无状态和有状态两种实现。
  • 通常 router 是无状态的，routing 处理完成一次请求，下一次请求与上一次请求无关
  • 通过外围包一个 wrapper agent 来做记忆管理；让router中的各个agent存储记忆（会比较复杂）

适用场景：

• 每个子任务有明确的垂直领域（需要各个子智能体处理独立知识域）
• 支持并行执行多个子agent，如查询多个领域知识
• 需要汇总多个agent的响应

public class RoutingByToolCallsExample {

    public static void main(String[] args) {
        Toolkit toolkit = new Toolkit();
        toolkit.registerTool(new SimpleTools());
     
        

        // Example of implicit routing with tool calls.
        Msg userMsg =
                Msg.builder()
                        .role(MsgRole.USER)
                        .content(
                                TextBlock.builder()
                                        .text("Help me to generate a quick sort function in Python")
                                        .build())
                        .build();
        try {
            Msg response = routerImplicit.call(userMsg).block();
            if (response != null) {
                System.out.println("Agent> " + MsgUtils.getTextContent(response) + "\n");
            } else {
                System.out.println("Agent> [No response]\n");
            }
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
            e.printStackTrace();
        }
    }

    public static class SimpleTools {
        /**
         * Generate Python code based on the demand.
         *
         * @param demand The demand for the Python code.
         */
        @Tool(
                name = "generate_Python_code",
                description = "Generate Python code based on the demand")
        public Msg generatePython(String demand) {
            System.out.println("I am PythonAgent,now generating Python code for demand: " + demand);
            String apiKey = ExampleUtils.getDashScopeApiKey();
            ReActAgent agent =
                    ReActAgent.builder()
                            .name("PythonAgent")
                            .sysPrompt(
                                    "You're a Python expert, your target is to generate Python code"
                                            + " based on the demand.")
                            .model(
                                    DashScopeChatModel.builder()
                                            .apiKey(apiKey)
                                            .modelName("qwen-max")
                                            .stream(true)
                                            .enableThinking(true)
                                            .formatter(new DashScopeChatFormatter())
                                            .defaultOptions(
                                                    GenerateOptions.builder()
                                                            .thinkingBudget(1024)
                                                            .build())
                                            .build())
                            .memory(new InMemoryMemory())
                            .toolkit(new Toolkit())
                            .build();

            Msg userMsg =
                    Msg.builder()
                            .role(MsgRole.USER)
                            .content(TextBlock.builder().text(demand).build())
                            .build();
            return agent.call(userMsg).block();
        }

        /**
         * Generate a poem based on the demand.
         *
         * @param demand The demand for the poem.
         */
        @Tool(name = "generate_poem", description = "Generate a poem based on the demand")
        public Msg generatePoem(String demand) {
            System.out.println("I am PoemAgent,now generating a poem for demand: " + demand);
            String apiKey = ExampleUtils.getDashScopeApiKey();
            ReActAgent agent =
                    ReActAgent.builder()
                            .name("PoemAgent")
                            .sysPrompt(
                                    "You're a poet, your target is to generate poems based on the"
                                            + " demand.")
                            .model()
                            .memory(new InMemoryMemory())
                            .toolkit(new Toolkit())
                            .build();

            Msg userMsg =
                    Msg.builder()
                            .role(MsgRole.USER)
                            .content(TextBlock.builder().text(demand).build())
                            .build();
            return agent.call(userMsg).block();
        }
    }
}

与 Supervisor 的区别：supervisor是模型基于会话上下文进行子agent调度；routing是基于特定逻辑或非常独立的LLM意图识别进行分类调度（适用于更具确定性的场景）

1.5 Pipeline（管道）

输入 → Agent1 → Agent2 → Agent3 → 输出

         ┌→ Agent1 → Output1
输入 →──┼→ Agent2 → Output2
         └→ Agent3 → Output3

能力描述：

• 提供顺序管道（SequentialPipeline）和扇出管道（FanoutPipeline）
• 支持智能体的顺序执行和并行执行
• 提供静态工厂方法简化管道创建
• 支持结构化输出
• 支持管道组合

核心特性：

• 顺序执行：前一个智能体的输出成为下一个的输入
• 并行执行：多个智能体处理相同输入（并发或顺序）
• 管道组合：可以组合多个管道
• 结构化输出支持

1.6 Custom Workflow（自定义工作流）模式

核心特点：

• 通过编写代码流程自定义执行流
• 完全控制图结构（顺序步骤、条件分支、循环、并行执行）
• 混合确定性逻辑和智能体行为
• 可以将以上的其他模式作为节点嵌入 workflow

适用场景：

• 以上模式不符合需求
• 强调确定性逻辑和智能体混合的编排
• 用例需要复杂的路由或多阶段处理
• 需要构建 RAG 管道等复杂工作流

实现方式：

• 每个节点可以是简单函数、LLM 调用或智能体
• 可以在自定义工作流中嵌入多智能体系统作为单个节点

[junefsh]

你好，我使用主子的路由模式搭建了一套业务流程，发现存在下面的一个问题：
子Agent处理完后，返回的内容，主Agent依然还会再调LLM处理一次，但通过trace看子Agent返回的内容已经符合预期了，能否直接跳过主Agent那次的LLM处理，这样既能减少Token消耗，也能缩短请求处理的时间