랭체인(LangChain)의 컴포넌트(Components) : Callbacks

랭체인의 컴포넌트에는 Model I/O, Retrieval, Composition, Memory, Callbacks 등이 있어요.
드디어 마지막 Component인 Callbacks에 대해서 학습해볼게요.
 
 
LangChain은 LLM 애플리케이션의 다양한 단계에 연결할 수 있는 콜백 시스템을 제공해요.
이 콜백 시스템은 로깅, 모니터링, 스트리밍 및 기타 작업에 유용합니다.
 
이 이벤트에 구독하려면 API에서 제공하는 Callbacks 인수(Argument)를 사용하면 되요.
이 인수(Argument)는 핸들러 객체들의 목록이며, 각각 하나 이상의 Method를 구현해야 되요.
각 Method에 대한 설명을 살펴볼게요.
 
 

# Callback handlers

 
CallbackHandler는 CallbackHandler 인터페이스를 구현한 객체로,
구독할 수 있는 각 이벤트에 대응하는 Method를 포함하고 있어요.
CallbackManager는 이벤트가 발생할 때마다 각 핸들러에서 적절한 Method를 호출한답니다.
 

class BaseCallbackHandler:
    """Base callback handler that can be used to handle callbacks from langchain."""

    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> Any:
        """Run when LLM starts running."""

    def on_chat_model_start(
        self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any
    ) -> Any:
        """Run when Chat Model starts running."""

    def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
        """Run on new LLM token. Only available when streaming is enabled."""

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        """Run when LLM ends running."""

    def on_llm_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """Run when LLM errors."""

    def on_chain_start(
        self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
    ) -> Any:
        """Run when chain starts running."""

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
        """Run when chain ends running."""

    def on_chain_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """Run when chain errors."""

    def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
    ) -> Any:
        """Run when tool starts running."""

    def on_tool_end(self, output: Any, **kwargs: Any) -> Any:
        """Run when tool ends running."""

    def on_tool_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """Run when tool errors."""

    def on_text(self, text: str, **kwargs: Any) -> Any:
        """Run on arbitrary text."""

    def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
        """Run on agent action."""

    def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
        """Run on agent end."""

 
 

# 시작하기

 
LangChain은 기본적으로 몇 가지 내장 핸들러를 제공하는데,
이 핸들러를 활용하면 쉽게 시작할 수 있어요.
이 핸들러는 langchain_core/callbacks 모듈에서 제공되요.
 
가장 기본적인 핸들러는 StdOutCallbackHandler로,
모든 이벤트를 표준 출력(stdout)에 로깅하는 역할을 해요.

* 참고 : 객체의 verbose 플래그가 true로 설정되어 있으면,
             StdOutCallbackHandler를 명시적으로 전달하지 않아도 자동으로 호출됩니다.
 

from langchain_core.callbacks import StdOutCallbackHandler
from langchain.chains import LLMChain
from langchain_openai import OpenAI
from langchain_core.prompts import PromptTemplate

handler = StdOutCallbackHandler()
llm = OpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

# Constructor callback: First, let's explicitly set the StdOutCallbackHandler when initializing our chain
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
chain.invoke({"number":2})

# Use verbose flag: Then, let's use the `verbose` flag to achieve the same result
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
chain.invoke({"number":2})

# Request callbacks: Finally, let's use the request `callbacks` to achieve the same result
chain = LLMChain(llm=llm, prompt=prompt)
chain.invoke({"number":2}, {"callbacks":[handler]})


# 결과

> Entering new LLMChain chain...
Prompt after formatting:
1 + 2 = 

> Finished chain.


> Entering new LLMChain chain...
Prompt after formatting:
1 + 2 = 

> Finished chain.


> Entering new LLMChain chain...
Prompt after formatting:
1 + 2 = 

> Finished chain.

 
 

# 콜백을 어디에 전달해야 할까?

 
API 대부분의 객체(체인, 모델, Tool, 에이전트 등)에서 콜백을 설정할 수 있으며,
이를 지정할 수 있는 위치는 두 가지가 있어요.
 

• Constructor Callbacks (생성자 콜백)
  • 생성자에 정의되며, LLMChain(callbacks=[handler], tags=['a-tag'])처럼 설정할 수 있어요.
  • 이 경우, 해당 객체에서 수행하는 모든 호출에 대해 콜백이 적용되며, 해당 객체에만 한정됩니다.
  • 즉, LLMChain 생성자에 핸들러를 전달하면 이 핸들러는 그 체인에 연결된 모델에서는 사용되지 않습니다.
• Request Callbacks (요청 콜백)
  • 요청을 실행할 때 사용하는 invoke Method에서 정의되요.
  • 이 경우 해당 요청과 그 요청에 포함된 모든 하위 요청에만 콜백이 적용됩니다.
  • 예를 들어 LLMChain을 호출하면 내부적으로 모델을 호출하게 되며,
  • 이 때 invoke() Method에 전달된 동일한 핸들러가 사용됩니다.
  • invoke() Method에서 콜백은 config 매개변수를 통해 전달됩니다.
  • 예제와 함께 살펴보겠습니다.
  • 참고 : 이 접근 방식은 batch, ainvoke, abatch Method에서 동일하게 적용됩니다.

handler = StdOutCallbackHandler()
llm = OpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

config = {
    'callbacks' : [handler]
}

chain = prompt | chain
chain.invoke({"number":2}, config=config)

 
* 참고 : chain = prompt | chain은 chain = LLMChain(llm=llm, prompt-prompt)와 동일한 의미에요.
 
verbose 인수(Argument)는 API의 대부분 객체(체인, 모델, Tool, 에이전트)에서 생성자 인수로 사용할 수 있습니다.
예를 들어 LLMChain(verbose=True)와 같이 설정할 수 있습니다.
이 설정은 해당 객체와 그 하위 객체의 callbacks 인수에 ConsoleCallbackHandler를 전달하는 것과 동일한 효과를 가집니다.
이를 활용하면 모든 이벤트가 콘솔에 로깅되므로 디버깅할 때 유용합니다.
 
 

각 방법은 언제 사용하는 것이 좋을까요?

 

• Constructor Callbacks (생성자 콜백)
  • 생성자 콜백은 특정 요청이 아니라 전체 체인과 관련된 로깅, 모니터링 등의 작업에 가장 유용합니다.
  • 예를 들어 LLMChain에서 수행된 모든 요청을 로깅하려면, 핸들러를 생성자에 전달하면 됩니다.
• Request Callbacks (요청 콜백)
  • 요청 콜백은 특정 요청의 출력을 특정 웹소켓 연결로 스트리밍하는 등의 작업에 가장 유용합니다.
  • 예를 들어 단일 요청의 출력을 웹소켓으로 스트리밍하려면 invoke() Method에 핸들러를 전달하면 됩니다.

 
 

# 마무리

 
Callbacks를 마지막으로 랭체인(LangChain)의 컴포넌트들에 대해 모두 알아봤어요.

공식 문서의 내용을 통해 Callbacks가 무엇인지 학습했는데 이해가 되나요?

다행히 저는 전체적인 개념을 이해할 수 있었어요.

 

물론 코드를 직접 작성해보면서 Callbacks의 개념을 더 가다듬고

어떻게 구현하는지, 어떻게 동작하는지 자세히 학습할 필요를 느꼈는데요.

그 전에!

컴포넌트들을 다시 한 번 더 복습하고!

그 다음에 코딩을 해볼까 해요.

 

이번 포스팅으로 Callbacks에 대해 자세히 이해할 수는 없지만

그래도 다른 사람들의 포스팅이나 자료를 찾아 더 자세히 공부하면

Callbacks를 효과적으로 사용할 수 있을 것 같아요!

 

이것으로 랭체인(LangChain)의 컴포넌트(Components) 포스팅은 끝!

 

글을 읽으면서 잘못된 점이 있으면 언제든 지적해주세요.
더욱 더 공부를 하고 수정을 해나갈게요.
 
 

# 참고

 

• LangChain Docs > Components : https://python.langchain.com/v0.1/docs/modules/

How-to guides | 🦜️🔗 LangChain

 

• LangChain Docs > Components > Callbacks : https://python.langchain.com/v0.1/docs/modules/callbacks/

How-to guides | 🦜️🔗 LangChain