当前位置：首页 > news >正文

Spring AI Alibaba实战：构建Human-in-the-Loop人机协同系统

news 2026/8/10 9:07:03

1. 项目概述：当AI需要一双“人眼”

最近在折腾一个智能客服的POC项目，用上了Spring AI Alibaba。模型回答的流畅度没问题，但一到涉及具体业务规则、价格计算或者敏感信息确认时，就有点“放飞自我”，要么答非所问，要么给出一个模糊的、需要二次确认的答案。直接全自动吧，怕出错捅娄子；完全不让AI参与吧，又浪费了它的效率。这个矛盾点，就是“Human-in-the-Loop”（人机回环，简称HITL）要解决的核心问题。

简单来说，HITL不是取代AI，也不是让人工全盘接管，而是在AI决策流程的关键节点上，巧妙地插入人工审核或确认环节。让AI负责处理海量、重复、模式化的任务，而把那些需要经验、判断力、创造力和承担责任的“硬骨头”留给人类专家。Spring AI Alibaba作为一套企业级的AI应用开发框架，它提供的HITL能力，本质上是一套标准化的“拦截与转交”机制。当AI模型生成的回答触发了预设的规则（比如低置信度、涉及关键词、属于特定业务类别），流程会自动暂停，将当前上下文和AI的初步结果推送给指定的人工处理接口或界面，待人处理完毕后再将最终结果返回给用户。

这解决的远不止是“答案对不对”的问题。在风控场景，它能防止模型误批贷款或交易；在内容创作场景，它能确保文案风格和品牌调性一致；在代码生成场景，资深工程师可以审查AI生成的代码片段是否有安全漏洞或架构缺陷。它的价值在于，将AI的“广度”与人类的“深度”结合，构建出一个既高效又可靠的协同系统。无论你是开发AI增强型应用的工程师，还是负责AI落地的产品经理，理解并实现HITL，都是让AI从“玩具”走向“工具”的关键一步。

2. 核心设计：构建可插拔的“决策拦截器”

实现HITL，听起来像是在代码里到处写if-else来调用人工审核，但这会迅速导致代码臃肿且难以维护。Spring AI Alibaba的思路是提供一套声明式的、基于策略的拦截框架，让我们能像配置路由规则一样，定义哪些AI请求需要“过一遍人手”。

2.1 策略定义：何时需要人工介入？

介入的时机是设计的灵魂。Spring AI Alibaba允许我们通过实现HumanInterventionPolicy接口来定义策略。常见的策略维度有以下几个，你可以根据业务需求组合使用：

置信度阈值策略：这是最直接的策略。AI模型（尤其是大语言模型）在生成每个token或整个回答时，通常会有一个置信度分数。我们可以设定一个阈值（例如0.7），当模型对整个回答或其中关键实体（如金额、日期、产品型号）的置信度低于该阈值时，触发人工审核。

@Component public class ConfidenceThresholdPolicy implements HumanInterventionPolicy { private static final double THRESHOLD = 0.7; @Override public boolean requiresIntervention(AiResponse aiResponse, ConversationContext context) { // 假设AiResponse中能获取到整体或分段的置信度 Double overallConfidence = aiResponse.getMetadata().getConfidence(); return overallConfidence != null && overallConfidence < THRESHOLD; } }

关键词/正则匹配策略：适用于高风险或高确定性领域。例如，在客服场景中，一旦用户提问包含“投诉”、“赔偿”、“法律”等关键词，或符合“我要告你们”这类正则模式，无论AI回答得多好，都强制转人工。

@Component public class KeywordPolicy implements HumanInterventionPolicy { private final List<String> highRiskKeywords = Arrays.asList("投诉", "赔偿", "起诉", "监管"); @Override public boolean requiresIntervention(AiResponse aiResponse, ConversationContext context) { String userQuery = context.getLastUserMessage(); return highRiskKeywords.stream().anyMatch(userQuery::contains); } }

业务规则策略：这是最体现业务复杂性的地方。策略可能需要调用外部服务或数据库。例如，在金融问答中，如果用户询问的理财产品风险等级为R5（最高风险），或者查询的转账金额超过单日限额，则必须人工复核。

@Component @RequiredArgsConstructor public class BusinessRulePolicy implements HumanInterventionPolicy { private final ProductService productService; @Override public boolean requiresIntervention(AiResponse aiResponse, ConversationContext context) { String productCode = extractProductCode(context); // 从上下文中提取产品代码 ProductInfo product = productService.getProductInfo(productCode); return product != null && "R5".equals(product.getRiskLevel()); } }

输出格式/结构验证策略：当AI需要生成结构化数据（如JSON、XML）或特定格式的文本（如邮件标题、固定报告）时，可以用此策略验证其输出是否符合schema或模板要求，不符合则转人工修正。

设计心得：策略应该尽量保持单一职责。一个策略只判断一个维度的条件。然后通过一个PolicyAggregator（策略聚合器）来组合这些策略，聚合逻辑可以是“任一满足即触发”或“全部满足才触发”。这样便于独立测试、复用和动态调整。

2.2 流程编排：介入后发生了什么？

一旦策略判定需要人工介入，标准的HITL流程便开始了。Spring AI Alibaba的HumanInTheLoopInterceptor会接管后续流程：

流程挂起与状态保存：当前的AI对话上下文（包括历史消息、用户问题、AI的初步回答、置信度等元数据）会被完整地序列化并存储到一个持久化介质中，比如Redis或数据库。同时，生成一个唯一的interventionTicketId（介入工单ID）。
异步通知：系统通过预配置的渠道（如内部消息队列、Webhook、邮件、钉钉/飞书机器人）向人工处理平台或指定的处理人员发送通知，内容包含工单ID、问题摘要和快速链接。
人工处理：处理人员在专属的管理后台查看工单详情。后台界面会清晰地展示用户原始问题、AI的初步回答、以及相关的上下文。处理人员可以：
- 直接采纳：认为AI回答无误，点击确认。
- 编辑修正：在AI回答的基础上进行修改和完善。
- 完全重写：丢弃AI回答，提供全新的人工回答。
- 补充信息：可能要求AI根据补充信息重新生成（这需要更复杂的循环设计）。
结果回调与流程恢复：人工处理完成后，处理平台调用Spring AI应用提供的回调接口（通常是一个REST端点），传入工单ID和最终处理结果。拦截器根据工单ID找回挂起的上下文，用人工处理的结果替换掉原先的AI回答，然后将这个“增强后”的响应返回给最初的用户请求。
学习与反馈（可选但重要）：最终被采纳的结果（无论是AI原答案还是人工修改版）可以被标记为“优质答案”，并反哺到AI模型的微调数据集中，实现闭环学习，让AI在未来遇到类似问题时表现更好。

关键点：整个介入流程必须是异步和非阻塞的。用户的本次请求在触发介入后应立即得到一个友好的提示，如“您的问题已提交给专家处理，稍后将通过[消息]通知您结果”，而不是让用户前端一直等待。这关乎用户体验。

3. 实战集成：在Spring AI Alibaba中落地HITL

理论讲完，我们来看如何在一个Spring Boot应用中具体实现。假设我们有一个简单的智能问答服务。

3.1 环境与依赖准备

首先，确保你的pom.xml引入了必要的依赖。除了Spring AI Alibaba的核心starter，我们还需要持久化（如JPA + MySQL）和消息通知（如Spring for Apache Kafka或钉钉SDK）的相关依赖。

<dependency> <groupId>com.alibaba.cloud.ai</groupId> <artifactId>spring-ai-alibaba-spring-boot-starter</artifactId> <version>最新版本</version> <!-- 请替换为实际版本 --> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-jpa</artifactId> </dependency> <dependency> <groupId>com.mysql</groupId> <artifactId>mysql-connector-j</artifactId> <scope>runtime</scope> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <!-- 用于异步通知 --> </dependency>

3.2 定义数据模型与仓储

我们需要一个实体来保存每次人工介入的工单状态。

@Entity @Table(name = "human_intervention_ticket") @Data public class InterventionTicket { @Id private String ticketId; // 工单唯一ID private String sessionId; // 对话会话ID @Lob private String originalContext; // 序列化后的完整对话上下文 private String userQuery; private String aiInitialResponse; private String status; // PENDING, PROCESSING, RESOLVED, CANCELLED private String assignedTo; // 分配的处理人 private String finalResponse; // 人工处理后的最终回答 private LocalDateTime createdAt; private LocalDateTime resolvedAt; }

对应的JpaRepositoryInterventionTicketRepository用于工单的增删改查。

3.3 实现核心拦截器与策略

这是最核心的部分。我们需要实现HumanInterventionInterceptor接口，并在其中注入我们定义的策略。

@Component @Slf4j @RequiredArgsConstructor public class CustomHumanInterventionInterceptor implements HumanInterventionInterceptor { private final List<HumanInterventionPolicy> policies; // 所有策略Bean会自动注入 private final InterventionTicketRepository ticketRepository; private final KafkaTemplate<String, InterventionAlert> kafkaTemplate; // 通知用 @Override public AiResponse intercept(AiResponse aiResponse, ConversationContext context) { // 1. 聚合策略判断是否需要介入 boolean needsIntervention = policies.stream() .anyMatch(policy -> policy.requiresIntervention(aiResponse, context)); if (!needsIntervention) { return aiResponse; // 无需介入，直接返回AI响应 } log.info("Human intervention triggered for session: {}", context.getSessionId()); // 2. 创建介入工单 InterventionTicket ticket = new InterventionTicket(); ticket.setTicketId(UUID.randomUUID().toString()); ticket.setSessionId(context.getSessionId()); ticket.setOriginalContext(serializeContext(context)); // 序列化方法 ticket.setUserQuery(context.getLastUserMessage()); ticket.setAiInitialResponse(aiResponse.getContent()); ticket.setStatus("PENDING"); ticket.setCreatedAt(LocalDateTime.now()); ticketRepository.save(ticket); // 3. 发送异步通知（例如到Kafka，由另一个服务消费并发送钉钉消息） InterventionAlert alert = new InterventionAlert(ticket.getTicketId(), ticket.getUserQuery()); kafkaTemplate.send("human-intervention-alerts", alert); // 4. 抛出特定异常或返回一个等待中的响应，告知上游需要人工处理 throw new HumanInterventionRequiredException("Request requires human review.", ticket.getTicketId()); // 注意：更优雅的方式是修改AiResponse，返回一个提示信息，而不是抛异常。 // 这取决于你的上游如何处理。这里用异常示意流程中断。 } }

注意事项：在实际项目中，intercept方法可能不会直接抛异常，而是返回一个特殊的AiResponse，其内容为“您的问题已提交审核，工单号：XXX”。这需要前后端协议配合。抛异常是一种让全局异常处理器统一处理并转换响应的方式。

3.4 构建人工处理后台与回调接口

人工处理后台可以是一个独立的Web应用。它提供一个列表页展示所有PENDING状态的工单，以及一个详情页供处理人员操作。

核心的回调接口（由AI服务提供）可能如下：

@RestController @RequestMapping("/api/intervention") @RequiredArgsConstructor public class InterventionCallbackController { private final InterventionTicketRepository ticketRepository; private final ConversationService conversationService; // 假设有服务能恢复对话 @PostMapping("/resolve/{ticketId}") public ResponseEntity<String> resolveTicket(@PathVariable String ticketId, @RequestBody ResolutionRequest request) { InterventionTicket ticket = ticketRepository.findById(ticketId) .orElseThrow(() -> new RuntimeException("Ticket not found")); // 更新工单状态和最终答案 ticket.setFinalResponse(request.getFinalResponse()); ticket.setStatus("RESOLVED"); ticket.setResolvedAt(LocalDateTime.now()); ticketRepository.save(ticket); // 关键：恢复原对话流程。这里需要将最终答案“注入”回原会话。 // 一种方式是将答案存入一个临时存储（如Redis），键为sessionId，原服务轮询或通过事件获取。 conversationService.completePendingResponse(ticket.getSessionId(), request.getFinalResponse()); return ResponseEntity.ok("Ticket resolved successfully."); } }

处理后台的设计要点：界面应把AI的初步回答和用户问题并排显示，高亮显示可能有问题或低置信度的部分。提供便捷的编辑工具和预设的常用修正短语按钮，提升处理效率。

3.5 配置与启用拦截器

最后，在配置类中，将我们的自定义拦截器注册到Spring AI的对话链中。

@Configuration @EnableAiClients public class AiConfig { @Bean public ChatClient chatClient(AiClient aiClient, CustomHumanInterventionInterceptor interceptor) { // 假设使用流式ChatClient return ChatClient.builder(aiClient) .interceptors(interceptor) // 注册HITL拦截器 .build(); } }

4. 进阶考量与性能优化

实现基础功能后，我们需要关注一些进阶问题，以确保系统在生产环境稳定可靠。

4.1 超时、降级与熔断

人工处理超时：不能无限期等待人工处理。需要为每个工单设置超时时间（如30分钟）。超时后，系统可以执行降级策略：例如，自动发送一条“问题已升级，请稍后”的消息给用户，或者尝试用一个更保守、安全的AI预设答案进行回复。
拦截器本身熔断：如果策略判断或工单创建服务出现故障（如数据库连接超时），拦截器应有熔断机制。可以记录错误日志并放行本次AI回答（可能伴随告警），而不是阻塞所有请求。这符合“Fail-Open”（失败时开放）的设计原则，保证核心问答功能不中断。

4.2 上下文管理与序列化

对话上下文可能很大，特别是支持长上下文模型后。完整序列化存储成本高。

选择性存储：并非所有历史消息都需要。可以只存储最近N轮对话，或者只存储与触发策略强相关的消息。
压缩与清理：对存储的上下文进行压缩。工单解决后，根据数据保留策略定期清理旧的上下文数据，避免存储膨胀。

4.3 策略的动态配置

将策略的阈值（如置信度）、关键词列表等配置外置到配置中心（如Nacos、Apollo）。这样可以在不重启应用的情况下，动态调整介入的敏感度。例如，大促期间客服压力大，可以临时调高置信度阈值，减少人工介入量；在模型刚上线或更新后，可以调低阈值，加强人工复核。

4.4 监控与度量

必须建立完善的监控体系：

介入率：触发人工介入的请求占总请求的比例。这是衡量AI模型在该场景下成熟度的关键指标。
平均处理时间：从触发介入到人工解决的平均耗时。用于评估人工处理团队的效率和用户体验。
采纳率与修改率：人工直接采纳AI答案的比例 vs 需要修改的比例。高采纳率说明AI质量好；高修改率则指明了模型需要优化的具体方向。
策略触发统计：每个策略分别触发了多少次。这能帮你分析哪些规则最常被触发，是否合理，是否需要优化。

5. 避坑指南与常见问题

在实际开发和运维中，我踩过不少坑，这里总结几个关键点：

策略过载导致“拦截风暴”：初期由于担心出错，设置了过多、过严的策略，导致超过50%的请求都走了人工，完全失去了AI提效的意义。建议：从小范围、高风险场景开始试点，逐步增加策略。密切监控介入率，目标是将其控制在一个可接受的较低水平（例如5%-10%），并持续通过反馈优化模型来降低这个比例。
上下文丢失或错乱：在异步回调恢复对话时，如果会话管理不当，可能出现“张冠李戴”，把工单A的答案回复给了用户B。解决方案：确保ticketId与sessionId的强关联，并在恢复时做严格校验。使用线程局部存储或显式的会话存储管理器来隔离上下文。
人工处理体验差：如果处理后台加载慢、信息展示不全、编辑困难，会严重影响处理人员的效率和意愿。心得：把处理后台当成一个重要的产品来设计。提供全文搜索、批量操作、常用语模板、与内部知识库联动等功能。处理效率直接关系到系统的整体响应时间。
忽略反馈闭环：人工处理完就结束了，没有把修正后的优质数据系统性地收集起来用于模型优化。建议：建立数据管道，将(用户问题，AI原始回答，人工修正后答案)这样的三元组自动存入特定数据集，定期用于模型的监督微调（SFT）。这是HITL长期价值最大化的关键。
对用户体验的冲击：如果每次介入都让用户等待很久，体验会很差。优化：对于明确需要较长时间处理的问题，在触发介入时立即给用户一个预期（“您的问题需要专家核实，预计30分钟内通过短信答复您”），并提供工单号供查询。对于可以快速判断的，尝试实现“实时协作”，即AI给出答案的同时，在后台同步发送给人工复核，如果几秒内人工无异议，则答案不变；如有修改，可通过推送等方式及时更新给用户（适用于IM场景）。