[Camel in Action] 19-2. Camel과 AI/LLM 통합 – 지능형 통합 파이프라인 구축

Table of Contents

Camel + AI의 가능성

Apache Camel의 강점은 다양한 시스템 통합입니다. 여기에 LLM(Large Language Model)을 연결하면 기존 데이터 파이프라인에 자연어 처리, 분류, 요약 기능을 추가할 수 있습니다. 이메일 자동 분류, 문서 요약, 이상 감지 등 다양한 지능형 통합이 가능합니다.

OpenAI API와 Camel 통합

HTTP 컴포넌트로 OpenAI API를 직접 호출하는 방법입니다.

from("activemq:queue:customer-emails")
  .process(exchange -> {
    String emailContent = exchange.getIn().getBody(String.class);
    String prompt = String.format("""
      다음 고객 이메일을 분석해서 카테고리를 분류해줘.
      카테고리: 불만, 문의, 칭찬, 환불요청
      이메일: %s
      답변은 카테고리 이름만 써줘.
      """, emailContent);
    exchange.getIn().setBody(prompt);
  })
  .setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
  .setHeader("Authorization", simple("Bearer ${env.OPENAI_API_KEY}"))
  .process(exchange -> {
    String prompt = exchange.getIn().getBody(String.class);
    exchange.getIn().setBody(Map.of(
      "model", "gpt-4o-mini",
      "messages", List.of(Map.of("role", "user", "content", prompt))
    ));
  })
  .marshal().json()
  .to("https://api.openai.com/v1/chat/completions?bridgeEndpoint=true")
  .unmarshal().json(JsonNode.class)
  .process(exchange -> {
    JsonNode response = exchange.getIn().getBody(JsonNode.class);
    String category = response.at("/choices/0/message/content").asText();
    exchange.getIn().setHeader("emailCategory", category);
  })
  .toD("activemq:queue:${header.emailCategory}-emails");

LangChain4j Camel 컴포넌트 활용

Camel 4.x의 camel-langchain4j 컴포넌트를 사용하면 더 쉽게 LLM을 통합합니다.

<dependency>
  <groupId>org.apache.camel</groupId>
  <artifactId>camel-langchain4j-chat</artifactId>
</dependency>

// 라우트에서 직접 LLM 호출
from("direct:summarize")
  .to("langchain4j-chat:myModel?chatOperation=CHAT_SINGLE_MESSAGE")
  .log("요약: ${body}");

RAG(Retrieval Augmented Generation) 파이프라인

사내 문서를 기반으로 질문에 답하는 RAG 시스템을 Camel으로 구현합니다.

// 1. 문서 인덱싱 파이프라인
from("file:documents?recursive=true")
  .to("tika:parse")          // 텍스트 추출
  .split(body().tokenize("

")) // 청크 분할
  .to("langchain4j-embeddings:vectorDB") // 벡터 DB에 저장

// 2. 질문 답변 파이프라인
from("direct:askQuestion")
  .to("langchain4j-embeddings:vectorDB?operation=search") // 유사 문서 검색
  .process(exchange -> {
    String question = exchange.getIn().getHeader("question", String.class);
    String context = exchange.getIn().getBody(String.class);
    exchange.getIn().setBody(
      "다음 문서를 참고해서 질문에 답해줘.
문서: " + context + "
질문: " + question);
  })
  .to("langchain4j-chat:gpt4");

이상 감지 파이프라인

로그나 메트릭 데이터에서 이상을 LLM으로 감지하는 패턴입니다.

from("kafka:system-logs")
  .aggregate(constant(true), new GroupedBodyAggregationStrategy())
    .completionTimeout(60000) // 1분간 로그 수집
  .process(exchange -> {
    String logs = exchange.getIn().getBody(String.class);
    exchange.getIn().setBody(
      "다음 시스템 로그에서 이상 징후를 찾아줘. " +
      "이상하면 '이상 감지:' 로 시작하는 설명을, 정상이면 '정상'만 답해줘.
" + logs);
  })
  .to("langchain4j-chat:claude")
  .filter(body().startsWith("이상 감지:"))
    .to("activemq:queue:alerts");

zerg96