怎么实现LLM根据用户的问题去动态调用智能体

[Taogang00]

请问怎么实现类似 《Dify-A2A Discovery 插件：调用外部智能体》下面流程的效果，就是不需要事先知道我目前注册了多少智能体，LLM根据用户的问题和注册上的智能体AgentCard 里面的技能，去动态调用智能体。

[xuhuafeifei]

我的个人想法，每次启动 agent 的时候，从注册中心拉取所有已注册的 agent 的描述与信息，注入到 systemprompt。
然后负责路由的 agent，就是通过这些外部 agent 的信息，决定调用哪个 agent。
我们可以给路由的 agent，提供 function tool，通过 tool 的形式与注册中心注册的 agent 进行 a2a通信

graph LR
    
    %% 用户层
    User[用户] --> RouterAgent[路由Agent]
    
    %% 注册中心
    Registry[注册中心<br/>Agent信息存储]
    RouterAgent -.-> Registry
    
    %% 执行Agent层
    RouterAgent --> A2A1[Agent 1<br/>领域专家1]
    RouterAgent --> A2A2[Agent 2<br/>领域专家2]
    RouterAgent --> A2A3[Agent 3<br/>领域专家3]
    
    %% 返回结果流
    A2A1 --> Results[处理结果]
    A2A2 --> Results
    A2A3 --> Results
    Results --> User
    
    class User user
    class RouterAgent router
    class Registry registry
    class A2A1,A2A2,A2A3 agent

Loading

graph LR
    %% 路由Agent核心模块（用子图表示内部结构）
    subgraph RouterAgent [路由Agent - 核心决策层]
        direction LR
        R1[拉取注册中心Agent信息]
        R2[注入System Prompt]
        R3[注册调用外部agent的function tool]
        R4[function tool -> A2A通信]
        
        R1 --> R2
        R2 --> R3
        R3 --> R4
    end

Loading

[xuhuafeifei]

这个是 demo 代码，单文件的。

startAgentServers方法，会在localhost:8081~8082启动 3 个 agent，并把它们写到一个 mock 的注册中心Registry

main函数，会从Registry中拉取startAgentServers注册的所有 agent 的元信息，然后构建Map<String, A2aAgent> agentMap，并将其传入自定义个匿名AgentTool中

匿名AgentTool对外暴露callAsync方法，该方法会通过 Agent 传入的agent_name和message，路由到对应的 Agent 中，以此实现动态调用Agent

package io.agentscope.examples.a2a;

import com.sun.net.httpserver.HttpServer;
import io.a2a.spec.AgentCard;
import io.a2a.spec.TransportProtocol;
import io.agentscope.core.a2a.agent.A2aAgent;
import io.agentscope.core.a2a.agent.card.WellKnownAgentCardResolver;
import io.agentscope.core.a2a.server.AgentScopeA2aServer;
import io.agentscope.core.a2a.server.card.ConfigurableAgentCard;
import io.agentscope.core.a2a.server.transport.DeploymentProperties;
import io.agentscope.core.a2a.server.transport.jsonrpc.JsonRpcTransportWrapper;
import io.agentscope.core.message.Msg;
import io.agentscope.core.message.ToolResultBlock;
import io.agentscope.core.model.DashScopeChatModel;
import io.agentscope.core.ReActAgent;
import io.agentscope.core.tool.AgentTool;
import io.agentscope.core.tool.ToolCallParam;
import io.agentscope.core.tool.Toolkit;

import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import reactor.core.publisher.Mono;

/**
 * 最小 A2A Demo - 单文件启动 3 个真实 A2A HTTP 服务器 + 路由客户端
 *
 * <p>架构：
 * <pre>
 *                    ┌─ :8081 英语老师 (A2A Server)
 *  用户 → 路由Agent ─┼─ :8082 数学老师 (A2A Server)
 *                    └─ :8083 语文老师 (A2A Server)
 * </pre>
 *
 * <p>核心流程：
 * <ol>
 *   <li>将所有 Agent 定义注册到本地 Registry</li>
 *   <li>启动 3 个独立的 A2A HTTP 服务器，每个监听不同端口</li>
 *   <li>从注册中心拉取所有 Agent 信息，注入路由 Agent 的 System Prompt</li>
 *   <li>路由 Agent 只有一个通用工具 call_agent(agent_name, message)</li>
 *   <li>通过 A2aAgent 连接远程 HTTP Agent</li>
 * </ol>
 */
public class MinimalA2aDemo {

    private static final ObjectMapper JSON = new ObjectMapper().registerModule(new JavaTimeModule()).disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);

    // ==================== 1. Agent 定义 ====================

    /**
     * 本地注册中心 - 存储所有已注册的 Agent
     */
    static class Registry {
        private static final Map<String, AgentDef> registeredAgents = new HashMap<>();

        static void registerAgent(AgentDef agent) {
            registeredAgents.put(agent.name(), agent);
        }

        static List<AgentDef> allAgents() {
            return List.copyOf(registeredAgents.values());
        }
    }

    record AgentDef(String name, String displayName, String description, String url, int port, String sysPrompt) {
    }

    // ==================== 2. 启动 A2A HTTP 服务器 ====================

    /**
     * 启动多个独立的 A2A HTTP 服务器，每个 Agent 监听不同的端口
     */
    static void startAgentServers(DashScopeChatModel model) throws IOException {
        // 定义并注册所有 Agent
        List<AgentDef> agentDefs = List.of(new AgentDef("english-agent", "英语老师", "专业英语翻译，支持中英互译", "http://localhost:8081/", 8081, "你是一个专业的英语翻译老师，支持中文和英文互译。请用简洁准确的语言进行翻译，并给出必要的语法解释。"), new AgentDef("math-agent", "数学老师", "专业数学辅导，解答代数/几何/微积分", "http://localhost:8082/", 8082, "你是一个专业的数学老师，擅长代数、几何、微积分等数学问题。请分步骤详细解答学生的数学问题。"), new AgentDef("chinese-agent", "语文老师", "专业语文教学，擅长古诗创作和文学赏析", "http://localhost:8083/", 8083, "你是一个专业的语文老师，擅长古诗创作和文学赏析。请帮助学生学习古诗文。"));

        for (AgentDef def : agentDefs) {
            Registry.registerAgent(def);
        }

        // 为每个 Agent 创建独立的 HTTP 服务器
        for (AgentDef def : Registry.allAgents()) {
            ReActAgent.Builder agentBuilder = ReActAgent.builder().name(def.name()).sysPrompt(def.sysPrompt()).model(model).maxIters(10);

            AgentScopeA2aServer server = AgentScopeA2aServer.builder(agentBuilder).agentCard(new ConfigurableAgentCard.Builder().name(def.displayName()).description(def.description()).url(def.url()).build()).deploymentProperties(new DeploymentProperties.Builder().host("localhost").port(def.port()).build()).build();

            JsonRpcTransportWrapper wrapper = server.getTransportWrapper(TransportProtocol.JSONRPC.asString(), JsonRpcTransportWrapper.class);
            AgentCard card = server.getAgentCard();

            HttpServer httpServer = HttpServer.create(new InetSocketAddress("localhost", def.port()), 0);
            httpServer.setExecutor(Executors.newCachedThreadPool());

            // POST / → JSON-RPC 端点
            httpServer.createContext("/", exchange -> {
                if (!"POST".equalsIgnoreCase(exchange.getRequestMethod())) {
                    exchange.sendResponseHeaders(405, -1);
                    exchange.close();
                    return;
                }
                byte[] body = exchange.getRequestBody().readAllBytes();
                String requestBody = new String(body, StandardCharsets.UTF_8);

                try {
                    Map<String, String> headers = new HashMap<>();
                    exchange.getRequestHeaders().forEach((k, v) -> {
                        if (v != null && !v.isEmpty()) {
                            headers.put(k, v.get(0));
                        }
                    });
                    Object response = wrapper.handleRequest(requestBody, headers, Map.of());

                    if (response instanceof reactor.core.publisher.Flux<?> fluxResponse) {
                        exchange.getResponseHeaders().set("Content-Type", "text/event-stream; charset=UTF-8");
                        exchange.sendResponseHeaders(200, 0);
                        try (OutputStream os = exchange.getResponseBody()) {
                            fluxResponse.toStream().forEach(item -> {
                                try {
                                    String data = JSON.writeValueAsString(item);
                                    os.write(("event: jsonrpc\ndata: " + data + "\n\n").getBytes(StandardCharsets.UTF_8));
                                    os.flush();
                                } catch (IOException e) {
                                    throw new RuntimeException(e);
                                }
                            });
                        }
                    } else {
                        String json = JSON.writeValueAsString(response);
                        exchange.getResponseHeaders().set("Content-Type", "application/json");
                        byte[] resp = json.getBytes(StandardCharsets.UTF_8);
                        exchange.sendResponseHeaders(200, resp.length);
                        try (OutputStream os = exchange.getResponseBody()) {
                            os.write(resp);
                        }
                    }
                } catch (Exception e) {
                    String error = "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"" + e.getMessage() + "\"},\"id\":null}";
                    exchange.getResponseHeaders().set("Content-Type", "application/json");
                    byte[] resp = error.getBytes(StandardCharsets.UTF_8);
                    exchange.sendResponseHeaders(200, resp.length);
                    try (OutputStream os = exchange.getResponseBody()) {
                        os.write(resp);
                    }
                }
                exchange.close();
            });

            httpServer.createContext("/.well-known/agent-card.json", exchange -> {
                System.err.println(card.name() + " 被请求");
                String json = JSON.writeValueAsString(card);
                exchange.getResponseHeaders().set("Content-Type", "application/json");
                byte[] resp = json.getBytes(StandardCharsets.UTF_8);
                exchange.sendResponseHeaders(200, resp.length);
                try (OutputStream os = exchange.getResponseBody()) {
                    os.write(resp);
                }
                exchange.close();
            });

            httpServer.start();
            server.postEndpointReady();
        }
    }

    // ==================== 3. 主程序 ====================

    public static void main(String[] args) throws Exception {
        String apiKey = System.getenv("DASHSCOPE_API_KEY");
        if (apiKey == null || apiKey.isEmpty()) {
            System.err.println("请设置环境变量 DASHSCOPE_API_KEY");
            System.exit(1);
        }

        DashScopeChatModel model = DashScopeChatModel.builder().apiKey(apiKey).modelName("qwen3.5-27b").build();

        startAgentServers(model);

        List<AgentDef> agents = Registry.allAgents();


        // 构建 agentMap
        Map<String, A2aAgent> agentMap = new HashMap<>();
        for (AgentDef info : agents) {
            A2aAgent a2aAgent = A2aAgent.builder().name(info.name()).agentCardResolver(WellKnownAgentCardResolver.builder().baseUrl(info.url()).build()).build();
            agentMap.put(info.name(), a2aAgent);
        }

        // 创建 Toolkit + 通用工具
        Toolkit toolkit = new Toolkit();
        toolkit.registerTool(new AgentTool() {
            @Override
            public String getName() {
                return "call_agent";
            }

            @Override
            public String getDescription() {
                return "调用远程 Agent。参数: agent_name (选择哪个Agent), message (要问的问题)";
            }

            @Override
            public Map<String, Object> getParameters() {
                Map<String, Object> schema = new HashMap<>();
                schema.put("type", "object");
                schema.put("properties", Map.of("agent_name", Map.of("type", "string", "description", "要调用的 Agent 名称"), "message", Map.of("type", "string", "description", "要发送给 Agent 的消息")));
                schema.put("required", List.of("agent_name", "message"));
                return schema;
            }

            @Override
            public Mono<ToolResultBlock> callAsync(ToolCallParam param) {
                Map<String, Object> input = param.getToolUseBlock().getInput();
                String agentName = (String) input.get("agent_name");
                String message = (String) input.get("message");

                A2aAgent agent = agentMap.get(agentName);
                if (agent == null) {
                    return Mono.just(ToolResultBlock.text("错误：找不到 Agent - " + agentName));
                }

                return agent.call(Msg.builder().textContent(message).build()).timeout(java.time.Duration.ofSeconds(45)).map(r -> ToolResultBlock.text(r.getTextContent())).onErrorResume(e -> Mono.just(ToolResultBlock.text("调用远程 Agent 失败: " + agentName + ", 原因: " + e.getMessage())));
            }
        });

        // 创建路由 Agent
        String systemPrompt = buildSystemPrompt(agents);
        ReActAgent routerAgent = ReActAgent.builder().name("Dispatcher").sysPrompt(systemPrompt).model(model).toolkit(toolkit).maxIters(5).build();

        System.out.println("\n=== 请输入问题 (quit 退出) ===\n");
        java.util.Scanner scanner = new java.util.Scanner(System.in);
        while (true) {
            System.out.println("用户: ");
            String input = scanner.nextLine().trim();
            if (input.isEmpty() || input.equalsIgnoreCase("quit")) break;

            try {
                var result = routerAgent.call(Msg.builder().textContent(input).build()).timeout(java.time.Duration.ofSeconds(60)).block();
                System.out.println("回复: " + result.getTextContent() + "\n");
            } catch (Exception e) {
                System.err.println("错误: " + e.getMessage() + "\n");
            }
        }
    }

    private static String buildSystemPrompt(List<AgentDef> agents) {
        StringBuilder sb = new StringBuilder();
        sb.append("你是一个智能路由调度员。分析用户问题后，选择合适的专业 Agent 来处理。\n\n");
        sb.append("## 可用的外部 Agent\n\n");
        for (int i = 0; i < agents.size(); i++) {
            AgentDef a = agents.get(i);
            sb.append(String.format("%d. **%s** (agent_name: `%s`)\n   - 描述: %s\n\n", i + 1, a.displayName(), a.name(), a.description()));
        }
        sb.append("## 工作方式\n");
        sb.append("调用 `call_agent` 工具，传入 agent_name 和 message 即可。\n");
        return sb.toString();
    }
}