[PYTHON/LANGCHAIN] AsyncCallbackHandler 클래스 : on_llm_start/on_llm_end 메소드를 재정의해서 커스텀 비동기 콜백 핸들러 만들기

■ AsyncCallbackHandler 클래스의 on_llm_start/on_llm_end 메소드를 재정의해서 커스텀 비동기 콜백 핸들러를 만드는 방법을 보여준다.

※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

▶ main.py


import asyncio

from dotenv                   import load_dotenv
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.callbacks import AsyncCallbackHandler
from typing                   import Dict
from typing                   import Any
from typing                   import List
from langchain_core.outputs   import LLMResult
from langchain_openai         import ChatOpenAI
from langchain_core.messages  import HumanMessage

load_dotenv()

class CustomSynchronousCallbackHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token : str, **keywordArgumentDictionary) -> None:
        print(f"on_llm_new_token : {token}")
        print()

class CustomAsynchronousCallbackHandler(AsyncCallbackHandler):
    """Async callback handler that can be used to handle callbacks from langchain."""

    async def on_llm_start(self, serializedDictionary : Dict[str, Any], promptList : List[str], **keywordArgumentDictionary: Any) -> None:
        """Run when chain starts running."""
        print("on_llm_start")
        await asyncio.sleep(0.3)
        className = serializedDictionary["name"]
        print(f"    [{className}] Hi! I just woke up. Your llm is starting")
        print()

    async def on_llm_end(self, result : LLMResult, **keywordArgumentDictionary: Any) -> None:
        """Run when chain ends running."""
        print("on_llm_end")
        await asyncio.sleep(0.3)
        print("    Hi! I just woke up. Your llm is ending")
        print()

chatOpenAI = ChatOpenAI(
    model      = "gpt-4o",
    max_tokens = 25,
    streaming  = True,
    callbacks  = [CustomSynchronousCallbackHandler(), CustomAsynchronousCallbackHandler()]
)

async def main():
    llmResult = await chatOpenAI.agenerate([[HumanMessage(content = "Tell me a joke")]])
    print(llmResult)

asyncio.run(main())

import asyncio

from dotenv import load_dotenv

from langchain_core.callbacks import BaseCallbackHandler

from langchain_core.callbacks import AsyncCallbackHandler

from typing import Dict

from typing import Any

from typing import List

from langchain_core.outputs import LLMResult

from langchain_openai import ChatOpenAI

from langchain_core.messages import HumanMessage

load_dotenv()

class CustomSynchronousCallbackHandler(BaseCallbackHandler):

def on_llm_new_token(self, token : str, **keywordArgumentDictionary) -> None:

print(f"on_llm_new_token : {token}")

print()

class CustomAsynchronousCallbackHandler(AsyncCallbackHandler):

"""Async callback handler that can be used to handle callbacks from langchain."""

async def on_llm_start(self, serializedDictionary : Dict[str, Any], promptList : List[str], **keywordArgumentDictionary: Any) -> None:

"""Run when chain starts running."""

print("on_llm_start")

await asyncio.sleep(0.3)

className = serializedDictionary["name"]

print(f" [{className}] Hi! I just woke up. Your llm is starting")

print()

async def on_llm_end(self, result : LLMResult, **keywordArgumentDictionary: Any) -> None:

"""Run when chain ends running."""

print("on_llm_end")

await asyncio.sleep(0.3)

print(" Hi! I just woke up. Your llm is ending")

print()

chatOpenAI = ChatOpenAI(

model = "gpt-4o",

max_tokens = 25,

streaming = True,

callbacks = [CustomSynchronousCallbackHandler(), CustomAsynchronousCallbackHandler()]

)

async def main():

llmResult = await chatOpenAI.agenerate([[HumanMessage(content = "Tell me a joke")]])

print(llmResult)

asyncio.run(main())

▶ requirements.txt


aiohappyeyeballs==2.4.3
aiohttp==3.11.6
aiosignal==1.3.1
annotated-types==0.7.0
anyio==4.6.2.post1
attrs==24.2.0
certifi==2024.8.30
charset-normalizer==3.4.0
colorama==0.4.6
distro==1.9.0
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.27.2
idna==3.10
jiter==0.7.1
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.7
langchain-core==0.3.19
langchain-openai==0.2.9
langchain-text-splitters==0.3.2
langsmith==0.1.143
multidict==6.1.0
numpy==1.26.4
openai==1.54.5
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.9.2
pydantic_core==2.23.4
python-dotenv==1.0.1
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.36
tenacity==9.0.0
tiktoken==0.8.0
tqdm==4.67.0
typing_extensions==4.12.2
urllib3==2.2.3
yarl==1.17.2

aiohappyeyeballs==2.4.3

aiohttp==3.11.6

aiosignal==1.3.1

annotated-types==0.7.0

anyio==4.6.2.post1

attrs==24.2.0

certifi==2024.8.30

charset-normalizer==3.4.0

colorama==0.4.6

distro==1.9.0

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.27.2

idna==3.10

jiter==0.7.1

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.7

langchain-core==0.3.19

langchain-openai==0.2.9

langchain-text-splitters==0.3.2

langsmith==0.1.143

multidict==6.1.0

numpy==1.26.4

openai==1.54.5

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.9.2

pydantic_core==2.23.4

python-dotenv==1.0.1

PyYAML==6.0.2

regex==2024.11.6

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.36

tenacity==9.0.0

tiktoken==0.8.0

tqdm==4.67.0

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.17.2

※ pip install python-dotenv langchain langchain-openai 명령을 실행했다.