CALLBACK Archives - icodebroker

[PYTHON/LANGCHAIN] BaseCallbackHandler 클래스 : on_custom_event 메소드를 재정의해서 커스텀 이벤트 처리하기

■ BaseCallbackHandler 클래스의 on_custom_event 메소드를 재정의해서 커스텀 이벤트를 처리하는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.callbacks         import BaseCallbackHandler
from typing                           import Any
from uuid                             import UUID
from typing                           import Optional
from typing                           import List
from typing                           import Dict
from langchain_core.callbacks.manager import dispatch_custom_event
from langchain_core.runnables         import RunnableLambda
from langchain_core.runnables.config  import RunnableConfig

class CustomCallbackHandler(BaseCallbackHandler):
    def on_custom_event(
        self,
        name     : str,
        data     : Any,
        *,
        run_id   : UUID,
        tags     : Optional[List[str]] = None,
        metadata : Optional[Dict[str, Any]] = None,
        **kwargs : Any
    ) -> None:
        print(f"Received event {name} with data : {data}, with tags : {tags}, with metadata : {metadata} and run_id : {run_id}")

@RunnableLambda
def testRunnableLambda(x : int, config : RunnableConfig) -> int:
    dispatch_custom_event("event1", {"x" : x})
    dispatch_custom_event("event2", {"x" : x})
    return x

customCallbackHandler = CustomCallbackHandler()

async def main():
    responseValue = testRunnableLambda.invoke(1, {"callbacks" : [customCallbackHandler], "tags" : ["foo", "bar"]})

    print(responseValue)

asyncio.run(main())

"""
Received event event1 with data : {'x': 1}, with tags : ['foo', 'bar'], with metadata : {} and run_id : 745a81f9-b39a-4003-8a09-f4b0592a5be8
Received event event2 with data : {'x': 1}, with tags : ['foo', 'bar'], with metadata : {} and run_id : 745a81f9-b39a-4003-8a09-f4b0592a5be8
1
"""

import asyncio

from langchain_core.callbacks import BaseCallbackHandler

from typing import Any

from uuid import UUID

from typing import Optional

from typing import List

from typing import Dict

from langchain_core.callbacks.manager import dispatch_custom_event

from langchain_core.runnables import RunnableLambda

from langchain_core.runnables.config import RunnableConfig

class CustomCallbackHandler(BaseCallbackHandler):

def on_custom_event(

self,

name : str,

data : Any,

run_id : UUID,

tags : Optional[List[str]] = None,

metadata : Optional[Dict[str, Any]] = None,

**kwargs : Any

) -> None:

print(f"Received event {name} with data : {data}, with tags : {tags}, with metadata : {metadata} and run_id : {run_id}")

@RunnableLambda

def testRunnableLambda(x : int, config : RunnableConfig) -> int:

dispatch_custom_event("event1", {"x" : x})

dispatch_custom_event("event2", {"x" : x})

return x

customCallbackHandler = CustomCallbackHandler()

async def main():

responseValue = testRunnableLambda.invoke(1, {"callbacks" : [customCallbackHandler], "tags" : ["foo", "bar"]})

print(responseValue)

asyncio.run(main())

"""

Received event event1 with data : {'x': 1}, with tags : ['foo', 'bar'], with metadata : {} and run_id : 745a81f9-b39a-4003-8a09-f4b0592a5be8

Received event event2 with data : {'x': 1}, with tags : ['foo', 'bar'], with metadata : {} and run_id : 745a81f9-b39a-4003-8a09-f4b0592a5be8

"""

▶ 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
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.27.2
idna==3.10
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.7
langchain-core==0.3.19
langchain-text-splitters==0.3.2
langsmith==0.1.143
multidict==6.1.0
numpy==1.26.4
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.9.2
pydantic_core==2.23.4
PyYAML==6.0.2
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.36
tenacity==9.0.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

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.27.2

idna==3.10

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.7

langchain-core==0.3.19

langchain-text-splitters==0.3.2

langsmith==0.1.143

multidict==6.1.0

numpy==1.26.4

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.9.2

pydantic_core==2.23.4

PyYAML==6.0.2

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.36

tenacity==9.0.0

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.17.2

※ pip install langchain 명령을

[PYTHON/LANGCHAIN] AsyncCallbackHandler 클래스 : on_custom_event 메소드를 재정의해서 커스텀 이벤트 처리하기

■ AsyncCallbackHandler 클래스의 on_custom_event 메소드를 재정의해서 커스텀 이벤트를 처리하는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.callbacks         import AsyncCallbackHandler
from typing                           import Any
from uuid                             import UUID
from typing                           import Optional
from typing                           import List
from typing                           import Dict
from langchain_core.runnables         import RunnableLambda
from langchain_core.callbacks.manager import adispatch_custom_event
from langchain_core.runnables.config  import RunnableConfig

class CustomAsyncCallbackHandler(AsyncCallbackHandler):
    async def on_custom_event(
        self,
        name     : str,
        data     : Any,
        *,
        run_id   : UUID,
        tags     : Optional[List[str]] = None,
        metadata : Optional[Dict[str, Any]] = None,
        **kwargs : Any,
    ) -> None:
        print(f"Received event {name} with data : {data}, with tags : {tags}, with metadata : {metadata} and run_id : {run_id}")

@RunnableLambda
async def testRunnableLambda(x : str, config : RunnableConfig) -> str:
    """An example that shows how to manually propagate config.

    You must do this if you're running python<=3.10.
    """
    await adispatch_custom_event("event1", {"x" : x}, config = config)
    await adispatch_custom_event("event2", 5        , config = config)
    return x

customAsyncCallbackHandler = CustomAsyncCallbackHandler()

async def main():
    responseValue = await testRunnableLambda.ainvoke(1, {"callbacks" : [customAsyncCallbackHandler], "tags" : ["foo", "bar"]})

    print(responseValue)

asyncio.run(main())

"""
Received event event1 with data: {'x': 1}, with tags: ['foo', 'bar'], with metadata: {} and run_id: 73b8e44b-38f8-4c08-a8c6-8e03c4a3d057
Received event event2 with data: 5, with tags: ['foo', 'bar'], with metadata: {} and run_id: 73b8e44b-38f8-4c08-a8c6-8e03c4a3d057
1
"""

import asyncio

from langchain_core.callbacks import AsyncCallbackHandler

from typing import Any

from uuid import UUID

from typing import Optional

from typing import List

from typing import Dict

from langchain_core.runnables import RunnableLambda

from langchain_core.callbacks.manager import adispatch_custom_event

from langchain_core.runnables.config import RunnableConfig

class CustomAsyncCallbackHandler(AsyncCallbackHandler):

async def on_custom_event(

self,

name : str,

data : Any,

run_id : UUID,

tags : Optional[List[str]] = None,

metadata : Optional[Dict[str, Any]] = None,

**kwargs : Any,

) -> None:

print(f"Received event {name} with data : {data}, with tags : {tags}, with metadata : {metadata} and run_id : {run_id}")

@RunnableLambda

async def testRunnableLambda(x : str, config : RunnableConfig) -> str:

"""An example that shows how to manually propagate config.

You must do this if you're running python<=3.10.

"""

await adispatch_custom_event("event1", {"x" : x}, config = config)

await adispatch_custom_event("event2", 5 , config = config)

return x

customAsyncCallbackHandler = CustomAsyncCallbackHandler()

async def main():

responseValue = await testRunnableLambda.ainvoke(1, {"callbacks" : [customAsyncCallbackHandler], "tags" : ["foo", "bar"]})

print(responseValue)

asyncio.run(main())

"""

Received event event1 with data: {'x': 1}, with tags: ['foo', 'bar'], with metadata: {} and run_id: 73b8e44b-38f8-4c08-a8c6-8e03c4a3d057

Received event event2 with data: 5, with tags: ['foo', 'bar'], with metadata: {} and run_id: 73b8e44b-38f8-4c08-a8c6-8e03c4a3d057

"""

▶ 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
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.27.2
idna==3.10
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.7
langchain-core==0.3.19
langchain-text-splitters==0.3.2
langsmith==0.1.143
multidict==6.1.0
numpy==1.26.4
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.9.2
pydantic_core==2.23.4
PyYAML==6.0.2
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.36
tenacity==9.0.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

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.27.2

idna==3.10

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.7

langchain-core==0.3.19

langchain-text-splitters==0.3.2

langsmith==0.1.143

multidict==6.1.0

numpy==1.26.4

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.9.2

pydantic_core==2.23.4

PyYAML==6.0.2

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.36

tenacity==9.0.0

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.17.2

※ pip install langchain 명령을

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

■ AsyncCallbackHandler 클래스의 on_llm_start/on_llm_end 메소드를 재정의해서 커스텀 비동기 콜백 핸들러를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶

[PYTHON/LANGCHAIN] BaseCallbackHandler 클래스 : on_llm_new_token 메소드를 재정의해서 커스텀 콜백 핸들러 만들기

■ BaseCallbackHandler 클래스의 on_llm_new_token 메소드를 재정의해서 커스텀 콜백 핸들러를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : 생성자에서 callbacks 인자를 사용해 콜백 핸들러 리스트 설정하기

■ ChatOpenAI 클래스의 생성자에서 callbacks 인자를 사용해 콜백 핸들러 리스트를 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶

[PYTHON/LANGCHAIN] RunnableSequence 클래스 : with_config 메소드에서 callbacks 인자를 사용해 콜백 핸들러 리스트 설정하기

■ RunnableSequence 클래스의 with_config 메소드에서 callbacks 인자를 사용해 콜백 핸들러 리스트를 설정하는 방법을 보여준다. ▶ main.py


from dotenv                   import load_dotenv
from langchain_core.callbacks import BaseCallbackHandler
from typing                   import Dict
from typing                   import Any
from typing                   import List
from langchain_core.messages  import BaseMessage
from langchain_core.outputs   import LLMResult
from langchain_core.prompts   import ChatPromptTemplate
from langchain_openai         import ChatOpenAI

load_dotenv()

class CustomCallbackHandler(BaseCallbackHandler):
    def on_chat_model_start(self, serializedDictionary : Dict[str, Any], messageListList : List[List[BaseMessage]], **keywordArgumentDictionary) -> None:
        print("on_chat_model_start")
        print()

    def on_llm_end(self, result : LLMResult, **keywordArgumentDictionary) -> None:
        print(f"on_llm_end : {result}")
        print()

    def on_chain_start(self, serializedDictionary: Dict[str, Any], inputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:
        if serializedDictionary:
            print(f"on_chain_start : {serializedDictionary.get("name")}")
        else:
            print(f"on_chain_start : {serializedDictionary}")
        print()

    def on_chain_end(self, outputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:
        print(f"on_chain_end : {outputDictionary}")
        print()

chatPromptTemplate = ChatPromptTemplate.from_template("What is 1 + {number}?")

chatOpenAI = ChatOpenAI(model = "gpt-4o")

runnableSequence = chatPromptTemplate | chatOpenAI

callBackHandlerList = [CustomCallbackHandler()]

runnableBinding = runnableSequence.with_config(callbacks = callBackHandlerList)

responseAIMessage = runnableBinding.invoke({"number" : "2"})

print(responseAIMessage)

"""
on_chain_start : None

on_chain_start : ChatPromptTemplate

on_chain_end : messages=[HumanMessage(content='What is 1 + 2?', additional_kwargs={}, response_metadata={})]

on_chat_model_start

on_llm_end : generations=[[ChatGeneration(text='1 + 2 equals 3.', generation_info={'finish_reason': 'stop', 'logprobs': None}, message=AIMessage(content='1 + 2 equals 3.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82', 'finish_reason': 'stop', 'logprobs': None}, id='run-a0d0dff3-4469-49a3-96ad-41ed409bf28d-0', usage_metadata={'input_tokens': 15, 'output_tokens': 8, 'total_tokens': 23, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}))]] llm_output={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82'} run=None type='LLMResult'

on_chain_end : content='1 + 2 equals 3.' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82', 'finish_reason': 'stop', 'logprobs': None} id='run-a0d0dff3-4469-49a3-96ad-41ed409bf28d-0' usage_metadata={'input_tokens': 15, 'output_tokens': 8, 'total_tokens': 23, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

content='1 + 2 equals 3.' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82', 'finish_reason': 'stop', 'logprobs': None} id='run-a0d0dff3-4469-49a3-96ad-41ed409bf28d-0' usage_metadata={'input_tokens': 15, 'output_tokens': 8, 'total_tokens': 23, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
"""

from dotenv import load_dotenv

from langchain_core.callbacks import BaseCallbackHandler

from typing import Dict

from typing import Any

from typing import List

from langchain_core.messages import BaseMessage

from langchain_core.outputs import LLMResult

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

load_dotenv()

class CustomCallbackHandler(BaseCallbackHandler):

def on_chat_model_start(self, serializedDictionary : Dict[str, Any], messageListList : List[List[BaseMessage]], **keywordArgumentDictionary) -> None:

print("on_chat_model_start")

print()

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

print(f"on_llm_end : {result}")

print()

def on_chain_start(self, serializedDictionary: Dict[str, Any], inputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:

if serializedDictionary:

print(f"on_chain_start : {serializedDictionary.get("name")}")

else:

print(f"on_chain_start : {serializedDictionary}")

print()

def on_chain_end(self, outputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:

print(f"on_chain_end : {outputDictionary}")

print()

chatPromptTemplate = ChatPromptTemplate.from_template("What is 1 + {number}?")

chatOpenAI = ChatOpenAI(model = "gpt-4o")

runnableSequence = chatPromptTemplate | chatOpenAI

callBackHandlerList = [CustomCallbackHandler()]

runnableBinding = runnableSequence.with_config(callbacks = callBackHandlerList)

responseAIMessage = runnableBinding.invoke({"number" : "2"})

print(responseAIMessage)

"""

on_chain_start : None

on_chain_start : ChatPromptTemplate

on_chain_end : messages=[HumanMessage(content='What is 1 + 2?', additional_kwargs={}, response_metadata={})]

on_chat_model_start

on_llm_end : generations=[[ChatGeneration(text='1 + 2 equals 3.', generation_info={'finish_reason': 'stop', 'logprobs': None}, message=AIMessage(content='1 + 2 equals 3.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82', 'finish_reason': 'stop', 'logprobs': None}, id='run-a0d0dff3-4469-49a3-96ad-41ed409bf28d-0', usage_metadata={'input_tokens': 15, 'output_tokens': 8, 'total_tokens': 23, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}))]] llm_output={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82'} run=None type='LLMResult'

on_chain_end : content='1 + 2 equals 3.' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82', 'finish_reason': 'stop', 'logprobs': None} id='run-a0d0dff3-4469-49a3-96ad-41ed409bf28d-0' usage_metadata={'input_tokens': 15, 'output_tokens': 8, 'total_tokens': 23, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

content='1 + 2 equals 3.' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 15, 'total_tokens': 23, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_831e067d82', 'finish_reason': 'stop', 'logprobs': None} id='run-a0d0dff3-4469-49a3-96ad-41ed409bf28d-0' usage_metadata={'input_tokens': 15, 'output_tokens': 8, 'total_tokens': 23, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

"""

▶ 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

[PYTHON/LANGCHAIN] RunnableSequence 클래스 : invoke 메소드에서 config 인자를 사용해 콜백 핸들러 설정하기

■ RunnableSequence 클래스의 invoke 메소드에서 config 인자를 사용해 콜백 핸들러를 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶

[PYTHON/LANGCHAIN] BaseCallbackHandler 클래스 : 커스텀 콜백 핸들러 만들기

■ BaseCallbackHandler 클래스를 사용해 커스텀 콜백 핸들러를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from dotenv                   import load_dotenv
from langchain_core.callbacks import BaseCallbackHandler
from typing                   import Dict
from typing                   import Any
from typing                   import List
from langchain_core.messages  import BaseMessage
from langchain_core.outputs   import LLMResult
from langchain_core.prompts   import ChatPromptTemplate
from langchain_openai         import ChatOpenAI

load_dotenv()

class CustomCallbackHandler(BaseCallbackHandler):
    def on_chat_model_start(self, serializedDictionary : Dict[str, Any], messageListList : List[List[BaseMessage]], **keywordArgumentDictionary) -> None:
        print("on_chat_model_start")
        print()

    def on_llm_end(self, result : LLMResult, **keywordArgumentDictionary) -> None:
        print(f"on_llm_end : {result}")
        print()

    def on_chain_start(self, serializedDictionary : Dict[str, Any], inputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:
        if serializedDictionary:
            print(f"on_chain_start : {serializedDictionary.get("name")}")
        else:
            print(f"on_chain_start : {serializedDictionary}")
        print()

    def on_chain_end(self, outputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:
        print(f"on_chain_end : {outputDictionary}")
        print()

chatPromptTemplate = ChatPromptTemplate.from_template("What is 1 + {number}?")

chatOpenAI = ChatOpenAI(model = "gpt-4o")

runnableSequence = chatPromptTemplate | chatOpenAI

callBackHandlerList = [CustomCallbackHandler()]

responseAIMessage  = runnableSequence.invoke({"number" : "2"}, config = {"callbacks" : callBackHandlerList})

print("-" * 100)
print(responseAIMessage)

from dotenv import load_dotenv

from langchain_core.callbacks import BaseCallbackHandler

from typing import Dict

from typing import Any

from typing import List

from langchain_core.messages import BaseMessage

from langchain_core.outputs import LLMResult

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

load_dotenv()

class CustomCallbackHandler(BaseCallbackHandler):

def on_chat_model_start(self, serializedDictionary : Dict[str, Any], messageListList : List[List[BaseMessage]], **keywordArgumentDictionary) -> None:

print("on_chat_model_start")

print()

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

print(f"on_llm_end : {result}")

print()

def on_chain_start(self, serializedDictionary : Dict[str, Any], inputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:

if serializedDictionary:

print(f"on_chain_start : {serializedDictionary.get("name")}")

else:

print(f"on_chain_start : {serializedDictionary}")

print()

def on_chain_end(self, outputDictionary : Dict[str, Any], **keywordArgumentDictionary) -> None:

print(f"on_chain_end : {outputDictionary}")

print()

chatPromptTemplate = ChatPromptTemplate.from_template("What is 1 + {number}?")

chatOpenAI = ChatOpenAI(model = "gpt-4o")

runnableSequence = chatPromptTemplate | chatOpenAI

callBackHandlerList = [CustomCallbackHandler()]

responseAIMessage = runnableSequence.invoke({"number" : "2"}, config = {"callbacks" : callBackHandlerList})

print("-" * 100)

print(responseAIMessage)

▶