张鹤臣1. Spring AI与自研DSL构建Multi-Agent规划层架构
在复杂业务场景中，Multi-Agent系统的规划层如同交响乐团的指挥家，负责将宏观目标分解为可执行的微观动作。基于Spring AI框架与自研DSL的规划层设计，能够实现从静态流程到动态决策的全覆盖。本文将深入解析三种核心规划模式及其实现方案。
关键提示：规划层的核心价值在于平衡确定性与灵活性。过度依赖静态规划会导致系统僵化，而完全动态规划则会引入不可控风险。
1.1 技术选型背景
Spring AI作为新兴的AI应用框架，其核心优势在于：
- 与Spring生态无缝集成，可利用现有IoC容器和AOP机制
- 提供标准化的ChatClient接口，支持多模型切换
- 内置Prompt模板管理，降低提示工程复杂度
自研DSL的设计考量：
- 采用YAML语法，业务人员可参与流程设计
- 支持条件路由、并行分支、循环控制等高级特性
- 通过SPEL表达式实现灵活的业务规则配置
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 三种规划模式深度解析
2.1 预定义静态计划模式
2.1.1 适用场景特征
- 流程节点和路径可预先枚举
- 执行逻辑不依赖运行时上下文
- 典型场景：审批流、ETL管道、报表生成
2.1.2 技术实现方案
DSL定义示例：
yaml复制workflow:
name: "订单履约流程"
type: "static"
nodes:
- id: "inventoryCheck"
type: "service"
class: "com.example.InventoryService#checkStock"
- id: "paymentCapture"
type: "service"
condition: "inventoryCheck.result == true"
edges:
- from: "start"
to: "inventoryCheck"
- from: "inventoryCheck"
to: "paymentCapture"
condition: "result == true"
Spring Graph实现要点：
java复制@Bean
public Graph<OrderContext> orderFulfillmentGraph(
InventoryNode inventoryNode,
PaymentNode paymentNode) {
return new StateGraph<>(OrderContext::new)
.addNode("inventoryCheck", inventoryNode)
.addNode("paymentCapture", paymentNode)
.addEdge(START, "inventoryCheck")
.addConditionalEdge("inventoryCheck",
ctx -> ctx.isInStock() ? "paymentCapture" : END)
.addEdge("paymentCapture", END)
.compile();
}
2.1.3 性能优化策略
- 预编译DAG结构到字节码
- 使用缓存避免重复解析DSL
- 并行执行独立节点分支
2.2 动态ReAct规划模式
2.2.1 核心运行机制
- Thought：LLM分析当前状态并制定行动计划
- Action：调用工具执行具体操作
- Observe：收集工具执行结果
- 循环直到任务完成或达到最大迭代次数
2.2.2 Spring AI实现方案
ReAct Agent核心逻辑：
java复制public class ResearchAgent {
private final ChatClient chatClient;
private final ToolExecutor toolExecutor;
public String execute(String researchTopic) {
List<Message> history = new ArrayList<>();
String prompt = """
你是一个研究助手，请用思考-行动-观察循环解决问题。
可用工具：
- webSearch(query): 网络搜索
- analyzeDocuments(docs): 文档分析""";
for (int i = 0; i < MAX_ITERATIONS; i++) {
String llmResponse = chatClient.prompt()
.system(prompt)
.messages(history)
.user(i == 0 ? researchTopic : "继续分析")
.call()
.content();
if (llmResponse.contains("最终答案:")) {
return extractAnswer(llmResponse);
}
ToolResult result = toolExecutor.execute(
parseToolCall(llmResponse));
history.add(new UserMessage("观察: " + result));
}
throw new MaxIterationException();
}
}
2.2.3 稳定性保障措施
- 设置最大迭代次数（通常5-10次）
- 工具调用超时控制（建议2-5秒）
- 输出格式严格校验
- 失败自动重试机制
2.3 层次分解规划模式
2.3.1 分层策略设计
- 战略层：宏观目标分解（LLM驱动）
- 战术层：子任务协调（规则引擎）
- 执行层：具体操作实施（预定义Agent）
2.3.2 混合执行方案
java复制public class HierarchicalPlanner {
public PlanResult execute(String objective) {
// 第一层：目标分解
List<SubGoal> subGoals = llmClient.decompose(objective);
// 第二层：并行执行子任务
List<CompletableFuture<SubResult>> futures = subGoals.stream()
.map(goal -> CompletableFuture.supplyAsync(() -> {
if (goal.isStatic()) {
return staticExecutor.execute(goal);
} else {
return dynamicAgent.execute(goal);
}
}))
.collect(Collectors.toList());
// 结果聚合
return aggregateResults(
futures.stream().map(CompletableFuture::join)
.collect(Collectors.toList()));
}
}
3. DSL高级特性实现
3.1 条件路由引擎
表达式语法设计：
yaml复制edges:
- from: "riskCheck"
to: "manualReview"
condition: |
state.riskScore > 0.7
AND state.customerLevel != 'VIP'
AND isBusinessDay(now())
SPEL扩展实现：
java复制public class CustomExpressionEvaluator {
private final SpelExpressionParser parser = new SpelExpressionParser();
public boolean evaluate(String expr, Map<String, Object> context) {
StandardEvaluationContext evalContext = new StandardEvaluationContext();
evalContext.setVariable("state", context);
evalContext.registerFunction("isBusinessDay",
DateUtils.class.getMethod("isBusinessDay", LocalDate.class));
return parser.parseExpression(expr)
.getValue(evalContext, Boolean.class);
}
}
3.2 并行控制原语
DSL配置示例：
yaml复制parallelGroup:
type: "fork-join"
branches:
- path: "legalReview"
agents: ["contractReviewer", "complianceCheck"]
- path: "techReview"
agents: ["architectReview"]
joinCondition: "allCompleted"
timeout: "PT1H"
Java执行逻辑：
java复制public Map<String, Object> executeParallel(ParallelConfig config) {
List<CompletableFuture<Map<String, Object>>> futures =
config.getBranches().stream()
.map(branch -> CompletableFuture.supplyAsync(() -> {
Map<String, Object> result = new HashMap<>();
for (String agent : branch.getAgents()) {
result.putAll(agentRegistry.get(agent).execute());
}
return result;
}))
.collect(Collectors.toList());
CompletableFuture<Void> allDone = CompletableFuture.allOf(
futures.toArray(new CompletableFuture[0]));
try {
allDone.get(config.getTimeout().toMillis(), TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
futures.forEach(f -> f.cancel(true));
throw new ParallelTimeoutException();
}
return futures.stream()
.map(CompletableFuture::join)
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.toMap(
Map.Entry::getKey,
Map.Entry::getValue));
}
4. 生产环境最佳实践
4.1 性能调优指标
| 场景类型 | 平均延迟 | 吞吐量(QPS) | 错误率 | 资源消耗 |
|---|---|---|---|---|
| 静态预定义流程 | <50ms | >1000 | <0.1% | 低 |
| 动态ReAct | 2-5s | 10-50 | 1-5% | 高 |
| 层次分解 | 5-10s | 5-20 | 3-8% | 非常高 |
4.2 稳定性保障方案
- 熔断机制：
java复制@CircuitBreaker(
failThreshold = 3,
resetTimeout = "PT1M",
exclude = {BusinessException.class})
public PlanResult executePlan(PlanRequest request) {
// 规划逻辑
}
- 降级策略：
java复制@Fallback(fallbackMethod = "staticFallback")
public PlanResult dynamicPlan(String goal) {
// 动态规划逻辑
}
private PlanResult staticFallback(String goal) {
return predefinedPlans.get(goal);
}
- 监控埋点：
java复制@Timed(value = "planning.time", description = "规划执行耗时")
@Counted(value = "planning.count", description = "规划执行次数")
public PlanResult monitorExecute(String input) {
// 业务逻辑
}
5. 典型场景实现：智能审批流
5.1 动态审批人分配
规则引擎配置：
yaml复制rules:
- name: "assignFinanceApprover"
when: "amount > 100000"
then: "approvers.add('CFO')"
- name: "requireLegalReview"
when: "contractType in ['M&A', 'IPO']"
then: "steps.add('legalReview')"
5.2 多级会签实现
并行会签逻辑：
java复制public CounterSignResult counterSign(List<String> approvers, Document doc) {
String batchId = UUID.randomUUID().toString();
List<ApprovalTask> tasks = approvers.stream()
.map(approver -> taskService.createTask(
batchId, approver, doc))
.collect(Collectors.toList());
CompletableFuture<List<ApprovalDecision>> future =
CompletableFuture.supplyAsync(() ->
tasks.stream()
.map(task -> taskService.awaitDecision(task.getId()))
.collect(Collectors.toList()));
try {
List<ApprovalDecision> decisions = future.get(2, TimeUnit.DAYS);
return new CounterSignResult(decisions);
} catch (TimeoutException e) {
taskService.cancelBatch(batchId);
throw new ApprovalTimeoutException();
}
}
5.3 状态恢复机制
持久化设计：
java复制@Entity
public class WorkflowState {
@Id
private String executionId;
@Enumerated(EnumType.STRING)
private WorkflowStatus status;
@Lob
@Convert(converter = JsonConverter.class)
private Map<String, Object> context;
@ElementCollection
private Set<String> completedNodes;
}
断点续跑逻辑：
java复制public void resumeWorkflow(String executionId) {
WorkflowState state = stateRepository.findById(executionId)
.orElseThrow();
CompiledGraph<?> graph = graphCache.get(state.getWorkflowId());
graph.resume(
state.getCurrentNode(),
state.getContext(),
state.getCompletedNodes());
}
在实际项目落地过程中，我们发现规划层的健壮性往往取决于异常处理的设计完备性。建议对每种规划模式都建立对应的错误处理子流程，例如当动态规划超过最大迭代次数时，自动转人工干预流程。同时要注意LLM生成内容的校验机制，避免错误的任务分解导致系统失控。