Spring AI - Advisor API

Spring Boot 이미지

 

Spring을 통해 개발하다 보면, Web 요청이나 응답을 중간에 가로채고 가공하여 전달하는 Interceptor가 존재합니다.

Spring AI에서도 AI 모델에 요청을 보내기 전/후로 작업을 진행하는 Interceptor와 비슷한 기능을 하는 Advisor가 존재하는데요, 오늘은 Advisor API에 대해 알아보려고 합니다.

 

Advisor API

Advisor APi는 AI 모델에 전달하기 전/후 중간에 요청 및 응답을 가로채어, payload를 가공하여 요청/응답을 가공하는 역할을 합니다.

이는 Spring의 Web에서 제공하는 Interceptor와 비슷한데요, 사실 이 도메인을 Interceptor로 만들면 어땠을까 하는 생각을 가지고 오늘 포스트를 진행하도록 하겠습니다.

CallAdvisor

Spring AI 에서 Advisor를 구현할 때 가장 먼저 떠올리는 항목이 될, CallAdvisor입니다. 왜냐... 해당 CallAdvisor의 경우 먼저 접하게 될(Spring을 사용하는 사람들이 MVC를 통해 가장 먼저 접하게 될 요소인..) 요청과 응답에 대한 가장 정답에 가까운 예시를 볼 수 있는 Advisor입니다.

 

하지만, 우리 서비스는 상황에 따라서 더 좋은 성능을 내야하기 때문에 해당 Advisor가 언제나 올바른 예시가 될 수는 없습니다. 다음 예시가 그에 대한 대답입니다.

 

StreamAdvisor

StreamAdvisor는 CallAdvisor와는 다르게 비동기식, 즉 Java Thread에 따라서 다르게 동작하는 Advisor입니다. 해당 Advisor는 WebFlux로 동작하게 되는데, 이는, 결국 요청과 응답에 대한 기존(동기식)과 다른 흐름을 만들어낼(비동기식) 수 있습니다.

 

왜 이 비동기식 방식이 중요할까요? 

• 우선, 서버는 하나의 서버가 아닌 여러 대의 서버로 동작하고,
• 각 서버의 상태를 보유하고 있지 않기 때문에 => staseless (상태 비저장)
• 이러한 비저장 상태의 서버를 관리할 수 있는 서로 다른 N개의 서버에 대해 각자의 서로 다른 비즈니스 로직을 가지고 관리할 수 있는 포인트가 필요합니다.
• 이를 기반으로, MSA 서비스에서는 서로 다른 서버 환경에서 서로 다른 상태를 관리할거라 생각하고 Stream이라는 API를 구현하게 되었습니다.

AdvisorChain

AdvisorChain은 각각의 CallAdviosr, StreamAdvisor interface를 구현한 클래스에 대해, 실행할 수 있도록 내부 필드와 advisor들을 실행할 수 있는 메서드를 제공할 수있는 interface입니다.

 

기본적으로, BaseAdvisorChain을 구현한 DefualtAroundAdvisorChain을 사용하고 있습니다.

public interface BaseAdvisorChain extends CallAdvisorChain, StreamAdvisorChain {
}

public class DefaultAroundAdvisorChain implements BaseAdvisorChain {
   ...
   public ChatClientResponse nextCall(ChatClientRequest chatClientRequest) {
        Assert.notNull(chatClientRequest, "the chatClientRequest cannot be null");
        if (this.callAdvisors.isEmpty()) {
            throw new IllegalStateException("No CallAdvisors available to execute");
        } else {
            CallAdvisor advisor = (CallAdvisor)this.callAdvisors.pop();
            AdvisorObservationContext observationContext = AdvisorObservationContext.builder().advisorName(advisor.getName()).chatClientRequest(chatClientRequest).order(advisor.getOrder()).build();
            return (ChatClientResponse)AdvisorObservationDocumentation.AI_ADVISOR.observation((ObservationConvention)null, DEFAULT_OBSERVATION_CONVENTION, () -> observationContext, this.observationRegistry).observe(() -> advisor.adviseCall(chatClientRequest, this));
        }
    }

    public Flux<ChatClientResponse> nextStream(ChatClientRequest chatClientRequest) {
        Assert.notNull(chatClientRequest, "the chatClientRequest cannot be null");
        return Flux.deferContextual((contextView) -> {
            if (this.streamAdvisors.isEmpty()) {
                return Flux.error(new IllegalStateException("No StreamAdvisors available to execute"));
            } else {
                StreamAdvisor advisor = (StreamAdvisor)this.streamAdvisors.pop();
                AdvisorObservationContext observationContext = AdvisorObservationContext.builder().advisorName(advisor.getName()).chatClientRequest(chatClientRequest).order(advisor.getOrder()).build();
                Observation observation = AdvisorObservationDocumentation.AI_ADVISOR.observation((ObservationConvention)null, DEFAULT_OBSERVATION_CONVENTION, () -> observationContext, this.observationRegistry);
                observation.parentObservation((Observation)contextView.getOrDefault("micrometer.observation", (Object)null)).start();
                return Flux.defer(() -> {
                    Flux var10000 = advisor.adviseStream(chatClientRequest, this);
                    Objects.requireNonNull(observation);
                    return var10000.doOnError(observation::error).doFinally((s) -> observation.stop()).contextWrite((ctx) -> ctx.put("micrometer.observation", observation));
                });
            }
        });
    }
   ...
}

 

전반적인 동작 흐름을 통해 알아보기

 

다음과 같은 prompt를 call 할 때, 어떻게 advisor가 호출되는지 알아보려고 합니다.

val chatClient: ChatClient = chatClientBuilder.build()

val prompt = Prompt("Spring AI에 대해서 알려줄래?")

val response = chatClient.prompt(prompt)
    .call()
    .content();

 

 

먼저 prompt를 호출하게 되면 다음과 같이 ChatClient를 구현한 DefaultChatClient를 통해 ChatClientRequestSpec을 만들어주는 것을 확인할 수 있습니다.

 

여기서 call을 호출하게 되면, ChatClient를 구현한 DefaultChatClient의 call() 메서드가 호출되게 됩니다.

public interface ChatClient {
ChatClientRequestSpec prompt();
    ...
    ChatClientRequestSpec prompt(String content);

    ChatClientRequestSpec prompt(Prompt prompt);
    
    CallResponseSpec call();
    ...
}

public class DefaultChatClient implements ChatClient {
    ...
    
    public ChatClient.ChatClientRequestSpec prompt(Prompt prompt) {
        Assert.notNull(prompt, "prompt cannot be null");
        DefaultChatClientRequestSpec spec = new DefaultChatClientRequestSpec(this.defaultChatClientRequest);
        if (prompt.getOptions() != null) {
            spec.options(prompt.getOptions());
        }

        if (prompt.getInstructions() != null) {
            spec.messages(prompt.getInstructions());
        }

        return spec;
    }
    
    public ChatClient.CallResponseSpec call() {
        BaseAdvisorChain advisorChain = this.buildAdvisorChain();
        return new DefaultCallResponseSpec(DefaultChatClientUtils.toChatClientRequest(this), advisorChain, this.observationRegistry, this.observationConvention);
    }
    ...
}

 

 

call 메서드 내에서 호출하는 buildAdvisorChain() 메서드를 다시 들여다 보면, 내부적으로 advisor 필드에 ChatModelCallAdvisor와 ChatModelStreamAdvisor를 asvisors 필드에 추가하고, DefaultAroundAdvisor를 생성하여 반환하는 것을 알 수 있습니다.

 

advisor를 추가했지만, 아직 호출하는 부분은 어딘지 잘 모르겠으니 조금 더 살펴보도록 하겠습니다.

private BaseAdvisorChain buildAdvisorChain() {
    this.advisors.add(ChatModelCallAdvisor.builder().chatModel(this.chatModel).build());
    this.advisors.add(ChatModelStreamAdvisor.builder().chatModel(this.chatModel).build());
    return DefaultAroundAdvisorChain.builder(this.observationRegistry).pushAll(this.advisors).templateRenderer(this.templateRenderer).build();
}

 

DefaultAroundAdvisor 내부를 살펴보면, 우리가 찾고 있던 advisor를 호출하는 부분을 볼 수 있습니다.

public interface CallAdvisorChain extends AdvisorChain {
    ChatClientResponse nextCall(ChatClientRequest chatClientRequest);

    List<CallAdvisor> getCallAdvisors();
}

public interface BaseAdvisorChain extends CallAdvisorChain, StreamAdvisorChain {
}

public class DefaultAroundAdviosrChain implements BaseAdvisor {
	public ChatClientResponse nextCall(ChatClientRequest chatClientRequest) {
        Assert.notNull(chatClientRequest, "the chatClientRequest cannot be null");
        if (this.callAdvisors.isEmpty()) {
            throw new IllegalStateException("No CallAdvisors available to execute");
        } else {
            CallAdvisor advisor = (CallAdvisor)this.callAdvisors.pop();
            AdvisorObservationContext observationContext = AdvisorObservationContext.builder().advisorName(advisor.getName()).chatClientRequest(chatClientRequest).order(advisor.getOrder()).build();
            return (ChatClientResponse)AdvisorObservationDocumentation.AI_ADVISOR.observation((ObservationConvention)null, DEFAULT_OBSERVATION_CONVENTION, () -> observationContext, this.observationRegistry).observe(() -> advisor.adviseCall(chatClientRequest, this));
        }
    }
}

 

 

 

위 동작 흐름으로 보아, 기본으로 사용하는 Advisor는 CalAdvisor와 StreamAdvisor가 존재하는 것으로 보입니다.

 

나만의 Advisor를 만들어보자!

나만의 커스텀한 Advisor를 만들기 위해서는 먼저 CallAdvisor와 StreamAdvisor를 구현해야 합니다. (DefaultAdvisor를 하나 만들어주면 얼마나 좋을까요..)

 

Adviosr와 CallAdvisor, StreamAdvisor 인터페이스 구조를 통해 어떤 메서드를 구현해야 하는지 다시 살펴봅시다.

public interface Advisor extends Ordered {
    int DEFAULT_CHAT_MEMORY_PRECEDENCE_ORDER = -2147482648;

    String getName();
}

public interface CallAdvisor extends Advisor {
    ChatClientResponse adviseCall(ChatClientRequest chatClientRequest, CallAdvisorChain callAdvisorChain);
}

public interface StreamAdvisor extends Advisor {
    Flux<ChatClientResponse> adviseStream(ChatClientRequest chatClientRequest, StreamAdvisorChain streamAdvisorChain);
}

 

 

위 interface를 구현한 나만의 CustomAdisor인 MyAdvisor입니다.

class MyAdvisor: CallAdvisor, StreamAdvisor {
    override fun getOrder(): Int {
        return 1
    }

    override fun getName(): String {
        return "my advisor"
    }

    override fun adviseStream(
        chatClientRequest: ChatClientRequest,
        streamAdvisorChain: StreamAdvisorChain
    ): Flux<ChatClientResponse> {
        val chatResponse = streamAdvisorChain.nextStream(chatClientRequest)
        return (ChatClientMessageAggregator()).aggregateChatClientResponse(
            chatResponse
        ) { println("advise stream") }
    }

    override fun adviseCall(
        chatClientRequest: ChatClientRequest,
        callAdvisorChain: CallAdvisorChain
    ): ChatClientResponse {
        println("advise call")
        return callAdvisorChain.nextCall(chatClientRequest)
    }
}