多 Agent 协作的 Java 后端实现——任务编排与上下文传递一、背景与问题大模型应用正从单 Agent 模式向多 Agent 协作模式演进。单一 Agent 处理复杂任务时存在角色混淆、知识过载、推理链过长的问题。多 Agent 系统将不同职能拆分为独立 Agent如规划 Agent、检索 Agent、验证 Agent各司其职通过任务编排协调执行顺序通过上下文传递共享中间结果。但在 Java 后端实现多 Agent 协作时工程挑战随之浮现如何编排 Agent 的执行顺序串行 vs 并行 vs 条件分支如何传递上下文而不让信息在链路中丢失或膨胀如何处理某个 Agent 执行失败时的降级与回退我们在企业智能审批系统中实现了 4-Agent 协作架构本文记录其 Java 后端实现的核心工程决策。二、方案设计多 Agent 协作的编排模式参考了 LangGraph 的图执行模型但用 Java 实现更贴近企业级工程需求。核心组件包括AgentNode每个 Agent 是一个独立的执行节点定义输入类型、输出类型、执行逻辑。WorkflowGraph定义 Agent 之间的连接关系与执行顺序支持串行、并行、条件分支。ContextBus上下文传递总线使用 ThreadLocal 消息队列实现跨 Agent 的数据共享与隔离。flowchart TD A[用户请求] -- B[规划Agent: PlannerAgent] B -- C{任务类型判断} C --|简单查询| D[检索Agent: RetrieverAgent] C --|复杂分析| E[并行执行] E -- F[检索Agent] E -- G[计算Agent] F -- H[合并上下文] G -- H H -- I[验证Agent: ValidatorAgent] D -- I I -- J{结果是否通过验证?} J --|是| K[输出结果] J --|否| L[回退到规划Agent重新分解]三、实战演示3.1 Agent 接口定义/** * Agent 基础接口——所有 Agent 必须实现此接口 * 定义统一的输入输出模型与执行契约 */ public interface AgentI, O { /** Agent 名称用于编排与日志追踪 */ String getName(); /** Agent 执行逻辑 */ O execute(I input, AgentContext context) throws AgentExecutionException; /** Agent 降级逻辑可选 */ default O fallback(I input, AgentContext context) { throw new AgentExecutionException( getName() 无降级策略执行失败); } }3.2 上下文总线/** * Agent 上下文总线——跨 Agent 的数据共享与隔离机制 * 每个 Workflow 执行创建一个独立的上下文实例 */ public class AgentContext { private final String workflowId; private final MapString, Object sharedData; // 共享数据跨 Agent 可读写 private final MapString, Object privateData; // 私有数据仅当前 Agent 可读写 private final ListContextEntry history; // 上下文传递历史审计追踪 public AgentContext(String workflowId) { this.workflowId workflowId; this.sharedData new ConcurrentHashMap(); this.privateData new ConcurrentHashMap(); this.history new ArrayList(); } /** * 写入共享数据——后续 Agent 可读取 * param key 数据键 * param value 数据值 * param source 写入的 Agent 名称 */ public void putShared(String key, Object value, String source) { sharedData.put(key, value); history.add(new ContextEntry(source, shared, key, value, Instant.now())); } /** * 读取共享数据——带类型检查 * param key 数据键 * param type 期望的类型 * return 类型安全的数据值 */ public T T getShared(String key, ClassT type) { Object value sharedData.get(key); if (value null) { return null; } if (!type.isInstance(value)) { throw new ContextTypeMismatchException( 共享数据类型不匹配: key key , 期望 type.getName() , 实际 value.getClass().getName()); } return type.cast(value); } /** * 获取上下文传递历史——用于审计与调试 */ public ListContextEntry getHistory() { return Collections.unmodifiableList(history); } }3.3 规划 Agent 实现/** * 规划 Agent——将用户请求分解为可执行的子任务序列 * 负责任务类型判断与编排路径选择 */ Service Slf4j public class PlannerAgent implements AgentUserRequest, PlanResult { private final ChatClient chatClient; public PlannerAgent(ChatClient chatClient) { this.chatClient chatClient; } Override public String getName() { return PlannerAgent; } Override public PlanResult execute(UserRequest input, AgentContext context) throws AgentExecutionException { try { String planPrompt 你是任务规划专家。根据用户请求制定执行计划。 输出格式JSON包含 tasks 数组每个 task 包含 name、type、dependsOn。 用户请求%s 可用 AgentRetrieverAgent检索、CalculatorAgent计算、ValidatorAgent验证 .formatted(input.getQuestion()); String response chatClient.call(planPrompt); PlanResult plan parsePlanResult(response); log.info(规划完成: workflowId{}, 任务数{}, context.getWorkflowId(), plan.getTasks().size()); // 将规划结果写入共享上下文 context.putShared(plan, plan, getName()); context.putShared(originalRequest, input, getName()); return plan; } catch (Exception e) { log.error(规划 Agent 执行失败: workflowId{}, context.getWorkflowId(), e); return fallback(input, context); } } Override public PlanResult fallback(UserRequest input, AgentContext context) { // 降级策略简单查询直接走检索路径 PlanResult simplePlan PlanResult.builder() .tasks(List.of( new Task(retrieval, RetrieverAgent, List.of()), new Task(validation, ValidatorAgent, List.of(retrieval)) )) .build(); context.putShared(plan, simplePlan, getName() -fallback); log.info(规划降级: workflowId{}, context.getWorkflowId()); return simplePlan; } }3.4 工作流编排引擎/** * 工作流编排引擎——执行 Agent 图的调度逻辑 * 支持串行、并行与条件分支三种执行模式 */ Service Slf4j public class WorkflowOrchestrator { private final MapString, Agent?, ? agentRegistry; private final ExecutorService parallelExecutor; public WorkflowOrchestrator(MapString, Agent?, ? agentRegistry) { this.agentRegistry agentRegistry; this.parallelExecutor Executors.newVirtualThreadPerTaskExecutor(); } /** * 执行工作流 * param workflowId 工作流唯一标识 * param plan 执行计划由 PlannerAgent 生成 * return 最终执行结果 */ public WorkflowResult execute(String workflowId, PlanResult plan) { AgentContext context new AgentContext(workflowId); MapString, Object taskResults new ConcurrentHashMap(); try { // 按依赖关系排序任务拓扑排序 ListTask sortedTasks topologicalSort(plan.getTasks()); for (Task task : sortedTasks) { Agent?, ? agent agentRegistry.get(task.getAgentName()); if (agent null) { throw new AgentNotFoundException( Agent 未注册: task.getAgentName()); } // 构造 Agent 输入从上下文中提取 Object input buildAgentInput(task, context, taskResults); // 执行 Agent log.info(开始执行: agent{}, task{}, workflowId{}, agent.getName(), task.getName(), workflowId); Object result agent.execute(input, context); taskResults.put(task.getName(), result); log.info(执行完成: agent{}, task{}, workflowId{}, agent.getName(), task.getName(), workflowId); } return WorkflowResult.success(workflowId, taskResults, context.getHistory()); } catch (Exception e) { log.error(工作流执行失败: workflowId{}, workflowId, e); return WorkflowResult.failure(workflowId, e.getMessage(), context.getHistory()); } } /** * 拓扑排序——根据 dependsOn 依赖关系排序任务 */ private ListTask topologicalSort(ListTask tasks) { MapString, Task taskMap tasks.stream() .collect(Collectors.toMap(Task::getName, t - t)); ListTask sorted new ArrayList(); SetString visited new HashSet(); for (Task task : tasks) { dfsSort(task, taskMap, visited, sorted); } return sorted; } private void dfsSort(Task task, MapString, Task taskMap, SetString visited, ListTask sorted) { if (visited.contains(task.getName())) return; visited.add(task.getName()); for (String dep : task.getDependsOn()) { Task depTask taskMap.get(dep); if (depTask ! null) dfsSort(depTask, taskMap, visited, sorted); } sorted.add(task); } }四、深度解析4.1 上下文传递的三种模式模式适用场景实现方式信息膨胀风险共享总线多 Agent 需读写同一数据AgentContext.sharedData中需定期清理链式传递串行 Agent后者只读前者的输出taskResults映射低消息队列并行 Agent异步通信内部 MQ需做消息压缩我们采用共享总线 链式传递的混合模式全局信息原始请求、规划结果写入共享总线局部信息单个 Agent 的输出通过链式传递。关键原则写入共享总线的数据必须有 TTL 或大小限制避免上下文随执行链路无限膨胀。4.2 降级策略的设计每个 Agent 的降级策略分为三个等级L1 轻量降级使用缓存结果或简化算法如检索 Agent 降级到关键词搜索。L2 跳过降级跳过该 Agent将前一步结果直接传给后续 Agent如验证 Agent 失败时直接输出。L3 重新规划回退到 PlannerAgent重新分解任务路径。降级等级的选择取决于 Agent 在链路中的角色核心 Agent如 Planner用 L1辅助 Agent如 Validator用 L2验证失败时用 L3。4.3 并行 Agent 的合并策略当 RetrieverAgent 和 CalculatorAgent 并行执行时它们的输出需要合并后传入 ValidatorAgent。合并策略有三种简单拼接Concat、结构化合并Merge by key、LLM 摘要合并Summarize via LLM。我们使用结构化合并按数据键将两个 Agent 的输出组合为统一对象避免信息冗余。4.4 工作流执行的幂等性保障多 Agent 工作流执行过程中某个 Agent 失败后的重试必须保证幂等——避免重复执行产生副作用。LLM 调用的重试通常是幂等的相同的 Prompt → 相同的推理结果但带有外部副作用的 Agent如发送通知、写入数据库必须加分布式锁 执行记录表保证至多一次。我们的实现方式是WorkflowOrchestrator 在调用每个 Agent 前检查agent_execution_log表中是否已有该 Agent 的完成记录若有则跳过执行直接取缓存结果实现断点续跑。五、总结与展望多 Agent 协作的 Java 后端实现核心工程挑战在编排与上下文两点编排引擎是骨架拓扑排序解决依赖关系并行执行利用虚拟线程降级策略保障鲁棒性。上下文总线是血脉共享数据 链式传递的混合模式兼顾灵活性与信息膨胀控制。每个 Agent 必须有降级方案生产环境中 LLM 调用随时可能失败降级不是可选功能而是必须项。后续方向基于 LangGraph Java 版的图定义 DSL替代手写拓扑排序、Agent 之间的协商协议而非硬编码的编排图、以及基于强化学习的动态编排策略——让系统根据历史执行数据自动调整 Agent 的执行顺序与参数。多 Agent 协作让 LLM 应用的能力边界大幅扩展但编排与上下文的工程复杂度也随之提升需要系统化的设计而非零散的拼凑。