LangChain4j核心功能与响应式编程实践

1. LangChain4j 核心功能解析

1.1 @SystemMessage注解的深度应用

@SystemMessage注解是LangChain4j框架中用于定义AI系统角色的核心注解。它的主要作用是将预定义的系统提示词与用户输入进行组合，形成完整的对话上下文。在实际项目中，我们通常将系统提示词存储在外部文件中，通过fromResource属性引用：

java复制@SystemMessage(fromResource = "prompt/codegen-routing-system-prompt.txt")
CodeGenTypeEnum routeCodeGenType(String userPrompt);

这种设计有三大优势：

1. 提示词与代码分离：修改提示词无需重新编译代码
2. 多环境适配：不同环境可以使用不同的提示词文件
3. 版本控制友好：文本文件的diff更清晰

最佳实践：将系统提示词按功能模块分类存放，如prompt/codegen/目录下存放代码生成相关提示词，prompt/chat/目录下存放对话相关提示词。

1.2 JSON输出配置的注意事项

当需要LangChain4j输出结构化JSON数据时，必须进行特殊配置。以下是完整的配置方案：

yaml复制spring:
  web:
    flux:
      timeout: 30m  # 延长SSE连接超时时间

langchain4j:
  open-ai:
    streaming-chat-model:
      response-format: 
        type: "json_object"  # 强制JSON输出格式
      max-tokens: 8192       # 增加最大token限制
      temperature: 0.7       # 控制输出随机性

常见问题排查：

• 若JSON输出被截断：检查max-tokens是否足够
• 若JSON格式错误：确保提示词中包含明确的JSON格式要求
• 若响应超时：适当增加timeout值并检查网络延迟

2. Reactor与SSE深度集成

2.1 为什么选择Reactor

Reactor项目是Spring生态中的响应式编程核心库，与LangChain4j的集成解决了以下关键问题：

1. 流式数据传输：将LangChain的Token流转换为SSE兼容格式
2. 背压处理：自动调节数据流速，防止客户端过载
3. 线程效率：非阻塞IO模型，支持高并发场景

2.2 完整依赖配置

xml复制<!-- 响应式Web基础 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

<!-- LangChain4j核心 -->
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-core</artifactId>
    <version>0.34.0</version>
</dependency>

<!-- Reactor适配器 -->
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-reactor</artifactId>
    <version>0.34.0</version>
</dependency>

<!-- OpenAI适配器 -->
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-open-ai</artifactId>
    <version>0.34.0</version>
</dependency>

版本对齐原则：所有LangChain4j相关组件的版本号必须严格一致。

2.3 流式接口实现

java复制@GetMapping(value = "/ai/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamChat(String question) {
    return aiService.generateStream(question)
            .doOnNext(token -> log.debug("Sending token: {}", token))
            .doOnError(e -> log.error("Stream error", e))
            .doOnComplete(() -> log.info("Stream completed"));
}

关键点说明：

• produces = MediaType.TEXT_EVENT_STREAM_VALUE 声明SSE端点
• doOnNext 等操作符用于添加副作用逻辑
• 无需手动关闭流，Spring会自动管理生命周期

3. 生产级配置方案

3.1 多模型配置策略

实际项目中通常需要支持多个AI模型，推荐使用配置类管理：

java复制@Configuration
public class ModelConfig {
    
    @Bean
    @Primary
    public ChatModel defaultChatModel() {
        return OpenAiChatModel.builder()
                .apiKey("${langchain4j.open-ai.api-key}")
                .modelName("gpt-4")
                .build();
    }
    
    @Bean
    public ChatModel routingChatModel() {
        return OpenAiChatModel.builder()
                .apiKey("${langchain4j.open-ai.api-key}")
                .modelName("gpt-3.5-turbo")
                .temperature(0.3)
                .build();
    }
}

3.2 异常处理机制

健壮的AI应用需要完善的异常处理：

java复制@RestControllerAdvice
public class AiExceptionHandler {
    
    @ExceptionHandler(AiException.class)
    public ResponseEntity<ErrorResponse> handleAiException(AiException e) {
        return ResponseEntity.status(e.getStatusCode())
                .body(new ErrorResponse(e.getErrorCode(), e.getMessage()));
    }
    
    @ExceptionHandler(Exception.class)
    public ResponseEntity<ErrorResponse> handleGeneralException(Exception e) {
        return ResponseEntity.internalServerError()
                .body(new ErrorResponse("AI_SYSTEM_ERROR", "AI服务暂不可用"));
    }
}

4. 高级功能实现

4.1 记忆管理实现

LangChain4j通过@MemoryId实现对话记忆：

java复制public interface AiService {
    @SystemMessage("你是一个代码助手")
    String chat(@MemoryId String sessionId, @UserMessage String message);
}

内存记忆配置：

java复制@Bean
public ChatMemoryProvider chatMemoryProvider() {
    return memoryId -> MessageWindowChatMemory.builder()
            .maxMessages(20)
            .id(memoryId)
            .build();
}

4.2 工具调用集成

定义工具接口：

java复制@Tool("获取当前天气")
public String getWeather(@P("城市") String city) {
    return weatherService.getCurrentWeather(city);
}

在AI服务中启用工具：

java复制AiServices.builder(MyAiService.class)
        .tools(new MyTools())
        .chatModel(chatModel)
        .build();

5. 性能优化实践

5.1 流式响应优化

java复制return aiService.generateStream(prompt)
        .bufferTimeout(50, Duration.ofMillis(100))  // 批量发送减少网络开销
        .map(list -> String.join("", list));

5.2 缓存策略

java复制@Bean
public CacheManager cacheManager() {
    return new CaffeineCacheManager("ai-responses") {
        @Override
        protected Cache<Object, Object> createNativeCache(String name) {
            return Caffeine.newBuilder()
                    .maximumSize(1000)
                    .expireAfterWrite(1, TimeUnit.HOURS)
                    .build();
        }
    };
}

6. 监控与日志

6.1 监控指标暴露

java复制@Bean
public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() {
    return registry -> registry.config().commonTags(
            "application", "ai-service",
            "region", System.getenv("REGION"));
}

6.2 结构化日志

java复制@Slf4j
@Aspect
public class AiLoggingAspect {
    
    @Around("execution(* com..ai..*(..))")
    public Object logAiCall(ProceedingJoinPoint pjp) throws Throwable {
        long start = System.currentTimeMillis();
        try {
            Object result = pjp.proceed();
            log.info("AI_CALL|success|method={}|duration={}ms", 
                    pjp.getSignature().getName(),
                    System.currentTimeMillis() - start);
            return result;
        } catch (Exception e) {
            log.error("AI_CALL|failed|method={}|error={}", 
                    pjp.getSignature().getName(),
                    e.getMessage());
            throw e;
        }
    }
}

7. 安全实践

7.1 输入验证

java复制@Validated
public interface AiService {
    String chat(@NotBlank @Size(max = 1000) String message);
}

7.2 敏感数据过滤

java复制@Component
public class SensitiveDataFilter implements Function<Prompt, Prompt> {
    
    @Override
    public Prompt apply(Prompt prompt) {
        String filteredText = sensitiveFilter.filter(prompt.text());
        return Prompt.from(filteredText);
    }
}

在AI服务配置中注册过滤器：

java复制AiServices.builder(MyAiService.class)
        .promptTemplateCustomizer(sensitiveDataFilter)
        .build();

8. 测试策略

8.1 单元测试示例

java复制@ExtendWith(MockitoExtension.class)
class AiServiceTest {
    
    @Mock
    private ChatModel chatModel;
    
    @InjectMocks
    private MyAiService aiService;
    
    @Test
    void shouldReturnResponse() {
        when(chatModel.generate(any())).thenReturn("Mocked response");
        
        String result = aiService.chat("test");
        
        assertEquals("Mocked response", result);
    }
}

8.2 集成测试

java复制@SpringBootTest
@AutoConfigureWebTestClient
class AiControllerIT {
    
    @Autowired
    private WebTestClient webClient;
    
    @Test
    void shouldStreamResponse() {
        webClient.get()
                .uri("/ai/chat?question=test")
                .exchange()
                .expectStatus().isOk()
                .expectHeader().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM)
                .expectBody(String.class).consumeWith(response -> {
                    assertNotNull(response.getResponseBody());
                });
    }
}

9. 部署方案

9.1 容器化配置

dockerfile复制FROM eclipse-temurin:17-jre
COPY target/ai-service.jar /app/
EXPOSE 8080
ENTRYPOINT ["java", "-jar", "/app/ai-service.jar"]

9.2 健康检查

yaml复制management:
  endpoint:
    health:
      probes:
        enabled: true
  endpoints:
    web:
      exposure:
        include: health,info,metrics

10. 经验总结

在实际项目中使用LangChain4j时，我们总结了以下关键经验：

1. 提示词工程：系统提示词的质量直接影响输出效果，需要反复调试
2. 流控策略：对于付费API，必须实现请求限流和失败重试
3. 监控指标：必须监控API调用延迟、费用和错误率
4. 本地缓存：对频繁查询的内容实现本地缓存，减少API调用
5. 版本隔离：不同环境的模型版本应该隔离，避免相互影响

一个典型的性能优化案例：通过将bufferTimeout从默认值调整为100ms，我们的API吞吐量提升了40%，同时保持了良好的实时性。