LANGCHAIN Archives - 26 중 2 번째 페이지

[PYTHON/LANGCHAIN] BaseBlobParser 클래스 : 커스텀 BLOB 파서 만들기

■ BaseBlobParser 클래스를 사용해 커스텀 BLOB 파서를 만드는 방법을 보여준다. ▶ main.py


from langchain_core.document_loaders import BaseBlobParser
from langchain_core.document_loaders import Blob
from typing                          import Iterator
from langchain_core.documents        import Document

class CustomBlobParser(BaseBlobParser):
    """A simple parser that creates a document from each line."""

    def lazy_parse(self, blob : Blob) -> Iterator[Document]:
        """Parse a blob into a document line by line."""
        lineNumber = 0
        with blob.as_bytes_io() as bufferedReader:
            for line in bufferedReader:
                lineNumber += 1
                yield Document(
                    page_content = line.strip(),
                    metadata     = {"line_number" : lineNumber, "source" : blob.source}
                )

blob1 = Blob.from_path("meow.txt")

customBlobParser = CustomBlobParser()

generator1 = customBlobParser.lazy_parse(blob1)

documentList1 = list(generator1)

for document in documentList1:
    print(document)
print("-" * 50)

blob2 = Blob(data = b"some data from memory\nmeow")

generator2 = customBlobParser.lazy_parse(blob2)

documentList2 = list(generator2)

for document in documentList2:
    print(document)

"""
page_content='meow meow1' metadata={'line_number': 1, 'source': './meow.txt'}
page_content='meow meow2' metadata={'line_number': 2, 'source': './meow.txt'}
page_content='meow meow3' metadata={'line_number': 3, 'source': './meow.txt'}
--------------------------------------------------
page_content='some data from memory' metadata={'line_number': 1, 'source': None}
page_content='meow' metadata={'line_number': 2, 'source': None}
"""

from langchain_core.document_loaders import BaseBlobParser

from langchain_core.document_loaders import Blob

from typing import Iterator

from langchain_core.documents import Document

class CustomBlobParser(BaseBlobParser):

"""A simple parser that creates a document from each line."""

def lazy_parse(self, blob : Blob) -> Iterator[Document]:

"""Parse a blob into a document line by line."""

lineNumber = 0

with blob.as_bytes_io() as bufferedReader:

for line in bufferedReader:

lineNumber += 1

yield Document(

page_content = line.strip(),

metadata = {"line_number" : lineNumber, "source" : blob.source}

)

blob1 = Blob.from_path("meow.txt")

customBlobParser = CustomBlobParser()

generator1 = customBlobParser.lazy_parse(blob1)

documentList1 = list(generator1)

for document in documentList1:

print(document)

print("-" * 50)

blob2 = Blob(data = b"some data from memory\nmeow")

generator2 = customBlobParser.lazy_parse(blob2)

documentList2 = list(generator2)

for document in documentList2:

print(document)

"""

page_content='meow meow1' metadata={'line_number': 1, 'source': './meow.txt'}

page_content='meow meow2' metadata={'line_number': 2, 'source': './meow.txt'}

page_content='meow meow3' metadata={'line_number': 3, 'source': './meow.txt'}

--------------------------------------------------

page_content='some data from memory' metadata={'line_number': 1, 'source': None}

page_content='meow' metadata={'line_number': 2, 'source': None}

"""

▶ requirements.txt


aiohappyeyeballs==2.4.3
aiohttp==3.11.7
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.8
langchain-core==0.3.21
langchain-text-splitters==0.3.2
langsmith==0.1.145
multidict==6.1.0
numpy==1.26.4
orjson==3.10.12
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic_core==2.27.1
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.18.0

aiohappyeyeballs==2.4.3

aiohttp==3.11.7

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.8

langchain-core==0.3.21

langchain-text-splitters==0.3.2

langsmith==0.1.145

multidict==6.1.0

numpy==1.26.4

orjson==3.10.12

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic_core==2.27.1

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.18.0

※ pip install langchain 명령을 실행했다.

[PYTHON/LANGCHAIN] BaseLoader 클래스 : 커스텀 문서 로더 만들기

■ BaseLoader 클래스를 사용해 커스텀 문서 로더를 만드는 방법을 보여준다. ▶ main.py


import aiofiles
import asyncio

from langchain_core.document_loaders import BaseLoader
from typing                          import Iterator
from langchain_core.documents        import Document
from typing                          import AsyncIterator

class CustomDocumentLoader(BaseLoader):
    """An example document loader that reads a file line by line."""

    def __init__(self, filePath : str) -> None:
        """Initialize the loader with a file path.

        Args :
            file_path: The path to the file to load.
        """
        self.filePath = filePath

    def lazy_load(self) -> Iterator[Document]:
        """A lazy loader that reads a file line by line.

        When you're implementing lazy load methods, you should use a generator to yield documents one by one.
        """
        with open(self.filePath, encoding = "utf-8") as textIOWrapper:
            lineNumber = 0
            for line in textIOWrapper:
                yield Document(
                    page_content = line.strip(),
                    metadata     = {"line_number" : lineNumber, "source" : self.filePath}
                )
                lineNumber += 1

    # alazy_load는 선택 사항이다.
    # 구현을 생략하면 lazy_load에 위임하는 기본 구현이 사용된다!
    async def alazy_load(self) -> AsyncIterator[Document]:
        """An async lazy loader that reads a file line by line."""

        async with aiofiles.open(self.filePath, encoding = "utf-8") as asyncTextIOWrapper:
            lineNumber = 0
            async for line in asyncTextIOWrapper:
                yield Document(
                    page_content = line.strip(),
                    metadata     = {"line_number" : lineNumber, "source" : self.filePath}
                )
                lineNumber += 1

with open("./meow.txt", "w", encoding = "utf-8") as textIOWrapper:
    fileContent = "meow meow1\nmeow meow2\nmeow meow3"
    textIOWrapper.write(fileContent)

customDocumentLoader = CustomDocumentLoader("./meow.txt")

for document in customDocumentLoader.lazy_load():
    print(document)
print("-" * 50)

documentList = customDocumentLoader.load()
for document in documentList:
    print(document)
print("-" * 50)

async def main():
    async for document in customDocumentLoader.alazy_load():
        print(document)
    print("-" * 50)

asyncio.run(main())

"""
page_content='meow meow1' metadata={'line_number': 0, 'source': './meow.txt'}
page_content='meow meow2' metadata={'line_number': 1, 'source': './meow.txt'}
page_content='meow meow3' metadata={'line_number': 2, 'source': './meow.txt'}
--------------------------------------------------
page_content='meow meow1' metadata={'line_number': 0, 'source': './meow.txt'}
page_content='meow meow2' metadata={'line_number': 1, 'source': './meow.txt'}
page_content='meow meow3' metadata={'line_number': 2, 'source': './meow.txt'}
--------------------------------------------------
page_content='meow meow1' metadata={'line_number': 0, 'source': './meow.txt'}
page_content='meow meow2' metadata={'line_number': 1, 'source': './meow.txt'}
page_content='meow meow3' metadata={'line_number': 2, 'source': './meow.txt'}
--------------------------------------------------
"""

import aiofiles

import asyncio

from langchain_core.document_loaders import BaseLoader

from typing import Iterator

from langchain_core.documents import Document

from typing import AsyncIterator

class CustomDocumentLoader(BaseLoader):

"""An example document loader that reads a file line by line."""

def __init__(self, filePath : str) -> None:

"""Initialize the loader with a file path.

Args :

file_path: The path to the file to load.

"""

self.filePath = filePath

def lazy_load(self) -> Iterator[Document]:

"""A lazy loader that reads a file line by line.

When you're implementing lazy load methods, you should use a generator to yield documents one by one.

"""

with open(self.filePath, encoding = "utf-8") as textIOWrapper:

lineNumber = 0

for line in textIOWrapper:

yield Document(

page_content = line.strip(),

metadata = {"line_number" : lineNumber, "source" : self.filePath}

)

lineNumber += 1

# alazy_load는 선택 사항이다.

# 구현을 생략하면 lazy_load에 위임하는 기본 구현이 사용된다!

async def alazy_load(self) -> AsyncIterator[Document]:

"""An async lazy loader that reads a file line by line."""

async with aiofiles.open(self.filePath, encoding = "utf-8") as asyncTextIOWrapper:

lineNumber = 0

async for line in asyncTextIOWrapper:

yield Document(

page_content = line.strip(),

metadata = {"line_number" : lineNumber, "source" : self.filePath}

)

lineNumber += 1

with open("./meow.txt", "w", encoding = "utf-8") as textIOWrapper:

fileContent = "meow meow1\nmeow meow2\nmeow meow3"

textIOWrapper.write(fileContent)

customDocumentLoader = CustomDocumentLoader("./meow.txt")

for document in customDocumentLoader.lazy_load():

print(document)

print("-" * 50)

documentList = customDocumentLoader.load()

for document in documentList:

print(document)

print("-" * 50)

async def main():

async for document in customDocumentLoader.alazy_load():

print(document)

print("-" * 50)

asyncio.run(main())

"""

page_content='meow meow1' metadata={'line_number': 0, 'source': './meow.txt'}

page_content='meow meow2' metadata={'line_number': 1, 'source': './meow.txt'}

page_content='meow meow3' metadata={'line_number': 2, 'source': './meow.txt'}

--------------------------------------------------

page_content='meow meow1' metadata={'line_number': 0, 'source': './meow.txt'}

page_content='meow meow2' metadata={'line_number': 1, 'source': './meow.txt'}

page_content='meow meow3' metadata={'line_number': 2, 'source': './meow.txt'}

--------------------------------------------------

page_content='meow meow1' metadata={'line_number': 0, 'source': './meow.txt'}

page_content='meow meow2' metadata={'line_number': 1, 'source': './meow.txt'}

page_content='meow meow3' metadata={'line_number': 2, 'source': './meow.txt'}

--------------------------------------------------

"""

▶ requirements.txt


aiofiles==24.1.0
aiohappyeyeballs==2.4.3
aiohttp==3.11.7
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.8
langchain-core==0.3.21
langchain-text-splitters==0.3.2
langsmith==0.1.145
multidict==6.1.0
numpy==1.26.4
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic_core==2.27.1
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.18.0

aiofiles==24.1.0

aiohappyeyeballs==2.4.3

aiohttp==3.11.7

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.8

langchain-core==0.3.21

langchain-text-splitters==0.3.2

langsmith==0.1.145

multidict==6.1.0

numpy==1.26.4

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic_core==2.27.1

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.18.0

※ pip install aiofiles langchain 명령을

[PYTHON/LANGCHAIN] BaseRetriever 클래스 : 커스텀 검색기 만들기

■ BaseRetriever 클래스를 사용해 커스텀 검색기를 만드는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.retrievers import BaseRetriever
from typing                    import List
from langchain_core.documents  import Document
from langchain_core.callbacks  import CallbackManagerForRetrieverRun

class CustomRetriever(BaseRetriever):
    """A toy retriever that contains the top k documents that contain the user query.

    This retriever only implements the sync method _get_relevant_documents.
    If the retriever were to involve file access or network access, it could benefit from a native async implementation of `_aget_relevant_documents`.
    As usual, with Runnables, there's a default async implementation that's provided that delegates to the sync implementation running on another thread.
    """

    documentList : List[Document]
    """List of documents to retrieve from."""

    k : int
    """Number of top results to return"""

    def _get_relevant_documents(self, query : str, *, run_manager : CallbackManagerForRetrieverRun) -> List[Document]:
        """Sync implementations for retriever."""
        matchingDocumentList = []
        for document in self.documentList:
            if len(matchingDocumentList) > self.k:
                return matchingDocumentList

            if query.lower() in document.page_content.lower():
                matchingDocumentList.append(document)
        return matchingDocumentList

    # 선택 사항 : _aget_relevant_documents를 재정의하여 보다 효율적인 네이티브 구현을 제공한다.
    # async def _aget_relevant_documents(self, query : str, *, run_manager : AsyncCallbackManagerForRetrieverRun) -> List[Document]:
    #     """Asynchronously get documents relevant to a query.

    #     Args :
    #         query       : String to find relevant documents for
    #         run_manager : The callbacks handler to use

    #     Returns :
    #         List of relevant documents
    #     """

documentList = [
    Document(
        page_content = "Dogs are great companions, known for their loyalty and friendliness.",
        metadata     = {"type" : "dog", "trait" : "loyalty"}
    ),
    Document(
        page_content = "Cats are independent pets that often enjoy their own space.",
        metadata     = {"type" : "cat", "trait" : "independence"}
    ),
    Document(
        page_content = "Goldfish are popular pets for beginners, requiring relatively simple care.",
        metadata     = {"type": "fish", "trait": "low maintenance"}
    ),
    Document(
        page_content = "Parrots are intelligent birds capable of mimicking human speech.",
        metadata     = {"type": "bird", "trait": "intelligence"}
    ),
    Document(
        page_content = "Rabbits are social animals that need plenty of space to hop around.",
        metadata     = {"type": "rabbit", "trait": "social"}
    ),
]

customRetriever = CustomRetriever(documentList = documentList, k = 3)

responseDocumentList = customRetriever.invoke("that")

print(responseDocumentList)
print("-" * 50)

responseDocumentListList = customRetriever.batch(["dog", "cat"])

print(responseDocumentListList)
print("-" * 50)

async def main():
    responseDocumentList = await customRetriever.ainvoke("that")

    print(responseDocumentList)
    print("-" * 50)

    async for eventDictionary in customRetriever.astream_events("bar", version = "v2"):
        print(eventDictionary)
    print("-" * 50)

asyncio.run(main())

"""
[Document(metadata={'type': 'cat', 'trait': 'independence'}, page_content='Cats are independent pets that often enjoy their own space.'), Document(metadata={'type': 'rabbit', 'trait': 'social'}, page_content='Rabbits are social animals that need plenty of space to hop around.')]
---------------------------------------------------
[[Document(metadata={'type': 'dog', 'trait': 'loyalty'}, page_content='Dogs are great companions, known for their loyalty and friendliness.')], [Document(metadata={'type': 'cat', 'trait': 'independence'}, page_content='Cats are independent pets that often enjoy their own space.')]]
---------------------------------------------------
[Document(metadata={'type': 'cat', 'trait': 'independence'}, page_content='Cats are independent pets that often enjoy their own space.'), Document(metadata={'type': 'rabbit', 'trait': 'social'}, page_content='Rabbits are social animals that need plenty of space to hop around.')]
---------------------------------------------------
{'event': 'on_retriever_start', 'data': {'input': 'bar'}, 'name': 'ToyRetriever', 'tags': [], 'run_id': '359f2805-45a0-422a-b8d9-ccc9067ea7de', 'metadata': {'ls_retriever_name': 'toy'}, 'parent_ids': []}
{'event': 'on_retriever_end', 'data': {'output': []}, 'run_id': '359f2805-45a0-422a-b8d9-ccc9067ea7de', 'name': 'ToyRetriever', 'tags': [], 'metadata': {'ls_retriever_name': 'toy'}, 'parent_ids': []}
---------------------------------------------------
"""

100

101

102

103

104

import asyncio

from langchain_core.retrievers import BaseRetriever

from typing import List

from langchain_core.documents import Document

from langchain_core.callbacks import CallbackManagerForRetrieverRun

class CustomRetriever(BaseRetriever):

"""A toy retriever that contains the top k documents that contain the user query.

This retriever only implements the sync method _get_relevant_documents.

If the retriever were to involve file access or network access, it could benefit from a native async implementation of `_aget_relevant_documents`.

As usual, with Runnables, there's a default async implementation that's provided that delegates to the sync implementation running on another thread.

"""

documentList : List[Document]

"""List of documents to retrieve from."""

k : int

"""Number of top results to return"""

def _get_relevant_documents(self, query : str, *, run_manager : CallbackManagerForRetrieverRun) -> List[Document]:

"""Sync implementations for retriever."""

matchingDocumentList = []

for document in self.documentList:

if len(matchingDocumentList) > self.k:

return matchingDocumentList

if query.lower() in document.page_content.lower():

matchingDocumentList.append(document)

return matchingDocumentList

# 선택 사항 : _aget_relevant_documents를 재정의하여 보다 효율적인 네이티브 구현을 제공한다.

# async def _aget_relevant_documents(self, query : str, *, run_manager : AsyncCallbackManagerForRetrieverRun) -> List[Document]:

# """Asynchronously get documents relevant to a query.

# Args :

# query : String to find relevant documents for

# run_manager : The callbacks handler to use

# Returns :

# List of relevant documents

# """

documentList = [

Document(

page_content = "Dogs are great companions, known for their loyalty and friendliness.",

metadata = {"type" : "dog", "trait" : "loyalty"}

Document(

page_content = "Cats are independent pets that often enjoy their own space.",

metadata = {"type" : "cat", "trait" : "independence"}

Document(

page_content = "Goldfish are popular pets for beginners, requiring relatively simple care.",

metadata = {"type": "fish", "trait": "low maintenance"}

Document(

page_content = "Parrots are intelligent birds capable of mimicking human speech.",

metadata = {"type": "bird", "trait": "intelligence"}

Document(

page_content = "Rabbits are social animals that need plenty of space to hop around.",

metadata = {"type": "rabbit", "trait": "social"}

]

customRetriever = CustomRetriever(documentList = documentList, k = 3)

responseDocumentList = customRetriever.invoke("that")

print(responseDocumentList)

print("-" * 50)

responseDocumentListList = customRetriever.batch(["dog", "cat"])

print(responseDocumentListList)

print("-" * 50)

async def main():

responseDocumentList = await customRetriever.ainvoke("that")

print(responseDocumentList)

print("-" * 50)

async for eventDictionary in customRetriever.astream_events("bar", version = "v2"):

print(eventDictionary)

print("-" * 50)

asyncio.run(main())

"""

[Document(metadata={'type': 'cat', 'trait': 'independence'}, page_content='Cats are independent pets that often enjoy their own space.'), Document(metadata={'type': 'rabbit', 'trait': 'social'}, page_content='Rabbits are social animals that need plenty of space to hop around.')]

---------------------------------------------------

[[Document(metadata={'type': 'dog', 'trait': 'loyalty'}, page_content='Dogs are great companions, known for their loyalty and friendliness.')], [Document(metadata={'type': 'cat', 'trait': 'independence'}, page_content='Cats are independent pets that often enjoy their own space.')]]

---------------------------------------------------

{'event': 'on_retriever_start', 'data': {'input': 'bar'}, 'name': 'ToyRetriever', 'tags': [], 'run_id': '359f2805-45a0-422a-b8d9-ccc9067ea7de', 'metadata': {'ls_retriever_name': 'toy'}, 'parent_ids': []}

{'event': 'on_retriever_end', 'data': {'output': []}, 'run_id': '359f2805-45a0-422a-b8d9-ccc9067ea7de', 'name': 'ToyRetriever', 'tags': [], 'metadata': {'ls_retriever_name': 'toy'}, 'parent_ids': []}

---------------------------------------------------

"""

▶ requirements.txt


aiohappyeyeballs==2.4.3
aiohttp==3.11.7
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.144
multidict==6.1.0
numpy==1.26.4
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic_core==2.27.1
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.18.0

aiohappyeyeballs==2.4.3

aiohttp==3.11.7

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.144

multidict==6.1.0

numpy==1.26.4

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic_core==2.27.1

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.18.0

※ pip install langchain 명령을 실행했다.

[PYTHON/LANGCHAIN] LLM 클래스 : 커스텀 모델 만들기

■ LLM 클래스를 사용해 커스텀 모델을 만드는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.language_models.llms import LLM
from typing                              import Optional
from typing                              import List
from langchain_core.callbacks.manager    import CallbackManagerForLLMRun
from typing                              import Any
from typing                              import Iterator
from langchain_core.outputs              import GenerationChunk
from typing                              import Dict
from langchain_core.prompts              import ChatPromptTemplate

class CustomLLM(LLM):
    """A custom chat model that echoes the first `n` characters of the input.

    When contributing an implementation to LangChain,
    carefully document the model including the initialization parameters,
    include an example of how to initialize the model and include any relevant links to the underlying models documentation or API.

    Example:

        .. code-block:: python

            customLLM = CustomLLM(n = 2)
            result = model.invoke([HumanMessage(content = "hello")])
            result = model.batch([[HumanMessage(content = "hello")], [HumanMessage(content = "world")]])
    """

    n : int
    """The number of characters from the last message of the prompt to be echoed."""

    def _call(
        self,
        prompt      : str,
        stop        : Optional[List[str]] = None,
        run_manager : Optional[CallbackManagerForLLMRun] = None,
        **kwargs    : Any,
    ) -> str:
        """Run the LLM on the given input.

        Override this method to implement the LLM logic.

        Args:
            prompt: The prompt to generate from.
            stop: Stop words to use when generating. Model output is cut off at the
                first occurrence of any of the stop substrings.
                If stop tokens are not supported consider raising NotImplementedError.
            run_manager: Callback manager for the run.
            **kwargs: Arbitrary additional keyword arguments. These are usually passed
                to the model provider API call.

        Returns:
            The model output as a string. Actual completions SHOULD NOT include the prompt.
        """
        if stop is not None:
            raise ValueError("stop kwargs are not permitted.")
        return prompt[:self.n]

    def _stream(
        self,
        prompt      : str,
        stop        : Optional[List[str]] = None,
        run_manager : Optional[CallbackManagerForLLMRun] = None,
        **kwargs    : Any,
    ) -> Iterator[GenerationChunk]:
        """Stream the LLM on the given prompt.

        This method should be overridden by subclasses that support streaming.

        If not implemented, the default behavior of calls to stream will be to
        fallback to the non-streaming version of the model and return
        the output as a single chunk.

        Args:
            prompt: The prompt to generate from.
            stop: Stop words to use when generating. Model output is cut off at the
                first occurrence of any of these substrings.
            run_manager: Callback manager for the run.
            **kwargs: Arbitrary additional keyword arguments. These are usually passed
                to the model provider API call.

        Returns:
            An iterator of GenerationChunks.
        """
        for character in prompt[:self.n]:
            generationChunk = GenerationChunk(text = character)
            if run_manager:
                run_manager.on_llm_new_token(generationChunk.text, chunk = generationChunk)

            yield generationChunk

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Return a dictionary of identifying parameters."""
        return {
            # The model name allows users to specify custom token counting
            # rules in LLM monitoring applications (e.g., in LangSmith users
            # can provide per token pricing for their model and monitor
            # costs for the given LLM.)
            "model_name": "CustomChatModel",
        }

    @property
    def _llm_type(self) -> str:
        """Get the type of language model used by this chat model. Used for logging purposes only."""
        return "custom"

customLLM = CustomLLM(n = 5)

responseString = customLLM.invoke("This is a foobar thing")

print(responseString)
print("-" * 50)

responseStringList = customLLM.batch(["woof woof woof", "meow meow meow"])

print(responseStringList)
print("-" * 50)

async def main():
    responseString = await customLLM.ainvoke("world")
    print(responseString)
    print("-" * 50)

    responseStringList = await customLLM.abatch(["woof woof woof", "meow meow meow"])
    print(responseStringList)
    print("-" * 50)

    async for stringChunk in customLLM.astream("hello"):
        print(stringChunk, end = "|", flush = True)
    print()
    print("-" * 50)

    chatPromptTemplate = ChatPromptTemplate.from_messages([("system", "you are a bot"), ("human", "{input}")])

    runnableSequence = chatPromptTemplate | customLLM

    index = 0

    async for eventDictionary in runnableSequence.astream_events({"input" : "hello there!"}, version = "v2"):
        print(eventDictionary)
        index += 1
        if index > 7:
            break
    print("-" * 50)

asyncio.run(main())

"""
This
---------------------------------------------------
['woof ', 'meow ']
---------------------------------------------------
world
---------------------------------------------------
['woof ', 'meow ']
---------------------------------------------------
h|e|l|l|o|
---------------------------------------------------
{'event': 'on_chain_start', 'data': {'input': {'input': 'hello there!'}}, 'name': 'RunnableSequence', 'tags': [], 'run_id': 'b74d2aa5-6202-4792-9d24-bec5c1abd446', 'metadata': {}, 'parent_ids': []}
{'event': 'on_prompt_start', 'data': {'input': {'input': 'hello there!'}}, 'name': 'ChatPromptTemplate', 'tags': ['seq:step:1'], 'run_id': '8303741b-eb65-4ba3-b415-4235c0dff8db', 'metadata': {}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}
{'event': 'on_prompt_end', 'data': {'output': ChatPromptValue(messages=[SystemMessage(content='you are a bot', additional_kwargs={}, response_metadata={}), HumanMessage(content='hello there!', additional_kwargs={}, response_metadata={})]), 'input': {'input': 'hello there!'}}, 'run_id': '8303741b-eb65-4ba3-b415-4235c0dff8db', 'name': 'ChatPromptTemplate', 'tags': ['seq:step:1'], 'metadata': {}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}
{'event': 'on_llm_start', 'data': {'input': {'prompts': ['System: you are a bot\nHuman: hello there!']}}, 'name': 'CustomLLM', 'tags': ['seq:step:2'], 'run_id': '1f28dce6-8326-443f-8f09-2ca6967d6171', 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}
{'event': 'on_llm_stream', 'data': {'chunk': GenerationChunk(text='S')}, 'run_id': '1f28dce6-8326-443f-8f09-2ca6967d6171', 'name': 'CustomLLM', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}
{'event': 'on_chain_stream', 'run_id': 'b74d2aa5-6202-4792-9d24-bec5c1abd446', 'name': 'RunnableSequence', 'tags': [], 'metadata': {}, 'data': {'chunk': 'S'}, 'parent_ids': []}
{'event': 'on_llm_stream', 'data': {'chunk': GenerationChunk(text='y')}, 'run_id': '1f28dce6-8326-443f-8f09-2ca6967d6171', 'name': 'CustomLLM', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}
{'event': 'on_chain_stream', 'run_id': 'b74d2aa5-6202-4792-9d24-bec5c1abd446', 'name': 'RunnableSequence', 'tags': [], 'metadata': {}, 'data': {'chunk': 'y'}, 'parent_ids': []}
---------------------------------------------------
"""

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

import asyncio

from langchain_core.language_models.llms import LLM

from typing import Optional

from typing import List

from langchain_core.callbacks.manager import CallbackManagerForLLMRun

from typing import Any

from typing import Iterator

from langchain_core.outputs import GenerationChunk

from typing import Dict

from langchain_core.prompts import ChatPromptTemplate

class CustomLLM(LLM):

"""A custom chat model that echoes the first `n` characters of the input.

When contributing an implementation to LangChain,

carefully document the model including the initialization parameters,

include an example of how to initialize the model and include any relevant links to the underlying models documentation or API.

Example:

.. code-block:: python

customLLM = CustomLLM(n = 2)

result = model.invoke([HumanMessage(content = "hello")])

result = model.batch([[HumanMessage(content = "hello")], [HumanMessage(content = "world")]])

"""

n : int

"""The number of characters from the last message of the prompt to be echoed."""

def _call(

self,

prompt : str,

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

run_manager : Optional[CallbackManagerForLLMRun] = None,

**kwargs : Any,

) -> str:

"""Run the LLM on the given input.

Override this method to implement the LLM logic.

Args:

prompt: The prompt to generate from.

stop: Stop words to use when generating. Model output is cut off at the

first occurrence of any of the stop substrings.

If stop tokens are not supported consider raising NotImplementedError.

run_manager: Callback manager for the run.

**kwargs: Arbitrary additional keyword arguments. These are usually passed

to the model provider API call.

Returns:

The model output as a string. Actual completions SHOULD NOT include the prompt.

"""

if stop is not None:

raise ValueError("stop kwargs are not permitted.")

return prompt[:self.n]

def _stream(

self,

prompt : str,

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

run_manager : Optional[CallbackManagerForLLMRun] = None,

**kwargs : Any,

) -> Iterator[GenerationChunk]:

"""Stream the LLM on the given prompt.

This method should be overridden by subclasses that support streaming.

If not implemented, the default behavior of calls to stream will be to

fallback to the non-streaming version of the model and return

the output as a single chunk.

Args:

prompt: The prompt to generate from.

stop: Stop words to use when generating. Model output is cut off at the

first occurrence of any of these substrings.

run_manager: Callback manager for the run.

**kwargs: Arbitrary additional keyword arguments. These are usually passed

to the model provider API call.

Returns:

An iterator of GenerationChunks.

"""

for character in prompt[:self.n]:

generationChunk = GenerationChunk(text = character)

if run_manager:

run_manager.on_llm_new_token(generationChunk.text, chunk = generationChunk)

yield generationChunk

@property

def _identifying_params(self) -> Dict[str, Any]:

"""Return a dictionary of identifying parameters."""

return {

# The model name allows users to specify custom token counting

# rules in LLM monitoring applications (e.g., in LangSmith users

# can provide per token pricing for their model and monitor

# costs for the given LLM.)

"model_name": "CustomChatModel",

}

@property

def _llm_type(self) -> str:

"""Get the type of language model used by this chat model. Used for logging purposes only."""

return "custom"

customLLM = CustomLLM(n = 5)

responseString = customLLM.invoke("This is a foobar thing")

print(responseString)

print("-" * 50)

responseStringList = customLLM.batch(["woof woof woof", "meow meow meow"])

print(responseStringList)

print("-" * 50)

async def main():

responseString = await customLLM.ainvoke("world")

print(responseString)

print("-" * 50)

responseStringList = await customLLM.abatch(["woof woof woof", "meow meow meow"])

print(responseStringList)

print("-" * 50)

async for stringChunk in customLLM.astream("hello"):

print(stringChunk, end = "|", flush = True)

print()

print("-" * 50)

chatPromptTemplate = ChatPromptTemplate.from_messages([("system", "you are a bot"), ("human", "{input}")])

runnableSequence = chatPromptTemplate | customLLM

index = 0

async for eventDictionary in runnableSequence.astream_events({"input" : "hello there!"}, version = "v2"):

print(eventDictionary)

index += 1

if index > 7:

break

print("-" * 50)

asyncio.run(main())

"""

This

---------------------------------------------------

['woof ', 'meow ']

---------------------------------------------------

world

---------------------------------------------------

['woof ', 'meow ']

---------------------------------------------------

h|e|l|l|o|

---------------------------------------------------

{'event': 'on_chain_start', 'data': {'input': {'input': 'hello there!'}}, 'name': 'RunnableSequence', 'tags': [], 'run_id': 'b74d2aa5-6202-4792-9d24-bec5c1abd446', 'metadata': {}, 'parent_ids': []}

{'event': 'on_prompt_start', 'data': {'input': {'input': 'hello there!'}}, 'name': 'ChatPromptTemplate', 'tags': ['seq:step:1'], 'run_id': '8303741b-eb65-4ba3-b415-4235c0dff8db', 'metadata': {}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}

{'event': 'on_prompt_end', 'data': {'output': ChatPromptValue(messages=[SystemMessage(content='you are a bot', additional_kwargs={}, response_metadata={}), HumanMessage(content='hello there!', additional_kwargs={}, response_metadata={})]), 'input': {'input': 'hello there!'}}, 'run_id': '8303741b-eb65-4ba3-b415-4235c0dff8db', 'name': 'ChatPromptTemplate', 'tags': ['seq:step:1'], 'metadata': {}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}

{'event': 'on_llm_start', 'data': {'input': {'prompts': ['System: you are a bot\nHuman: hello there!']}}, 'name': 'CustomLLM', 'tags': ['seq:step:2'], 'run_id': '1f28dce6-8326-443f-8f09-2ca6967d6171', 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}

{'event': 'on_llm_stream', 'data': {'chunk': GenerationChunk(text='S')}, 'run_id': '1f28dce6-8326-443f-8f09-2ca6967d6171', 'name': 'CustomLLM', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}

{'event': 'on_chain_stream', 'run_id': 'b74d2aa5-6202-4792-9d24-bec5c1abd446', 'name': 'RunnableSequence', 'tags': [], 'metadata': {}, 'data': {'chunk': 'S'}, 'parent_ids': []}

{'event': 'on_llm_stream', 'data': {'chunk': GenerationChunk(text='y')}, 'run_id': '1f28dce6-8326-443f-8f09-2ca6967d6171', 'name': 'CustomLLM', 'tags': ['seq:step:2'], 'metadata': {'ls_provider': 'custom', 'ls_model_type': 'llm'}, 'parent_ids': ['b74d2aa5-6202-4792-9d24-bec5c1abd446']}

{'event': 'on_chain_stream', 'run_id': 'b74d2aa5-6202-4792-9d24-bec5c1abd446', 'name': 'RunnableSequence', 'tags': [], 'metadata': {}, 'data': {'chunk': 'y'}, 'parent_ids': []}

---------------------------------------------------

"""

▶ requirements.txt


aiohappyeyeballs==2.4.3
aiohttp==3.11.7
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.144
multidict==6.1.0
numpy==1.26.4
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic_core==2.27.1
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.18.0

aiohappyeyeballs==2.4.3

aiohttp==3.11.7

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.144

multidict==6.1.0

numpy==1.26.4

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic_core==2.27.1

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.18.0

※ pip install langchain 명령을 실행했다.

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : model_kwargs 속성을 사용해 함수 설정하기

■ ChatOpenAI 클래스의 model_kwargs 속성을 사용해 함수를 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


import json

from dotenv           import load_dotenv
from langchain_openai import ChatOpenAI
from typing           import Callable
from langchain.schema import HumanMessage
from langchain.schema import FunctionMessage

load_dotenv()

def getWeather(location : str) -> str:
    # 실제로는 날씨 API를 호출하거나 DB를 조회하겠지만, 예시를 위해 하드코딩된 값을 반환한다.
    weatherDictionary = {
        "location"    : location,
        "temperature" : "20도",
        "condition"   : "맑음"
    }
    return json.dumps(weatherDictionary)

functionList = [
    {
        "name"        : "getWeather",
        "description" : "특정 도시의 현재 날씨 정보를 조회합니다",
        "parameters"  : {
            "type"       : "object",
            "properties" : {
                "location" : {
                    "type"        : "string",
                    "description" : "도시 이름 (예 : 서울, 부산)"
                }
            },
            "required" : ["location"]
        }
    }
]

chatOpenAI = ChatOpenAI(temperature = 0)

chatOpenAI.model_kwargs = {"functions" : functionList}

def processConversation(humanMessageContent : str, availableFunctionDictionary : dict[str, Callable]):
    messageList = [HumanMessage(content = humanMessageContent)]

    while True:
        response = chatOpenAI.generate([messageList])

        responseAIMessage = response.generations[0][0].message

        if responseAIMessage.additional_kwargs.get("function_call"):
            functionName               = responseAIMessage.additional_kwargs["function_call"]["name"]
            functionArgumentDictionary = json.loads(responseAIMessage.additional_kwargs["function_call"]["arguments"])

            functionToCall   = availableFunctionDictionary[functionName]
            functionResponse = functionToCall(**functionArgumentDictionary)

            messageList.append(responseAIMessage)
            messageList.append(FunctionMessage(name = functionName, content = functionResponse))
        else:
            messageList.append(responseAIMessage)
            break

    return messageList

availableFunctionDictionary = {"getWeather" : getWeather}

responseMessageList = processConversation("서울의 현재 날씨를 알려줘", availableFunctionDictionary)

for responseMessage in responseMessageList:
    print(f"Type    : {type(responseMessage).__name__}")
    print(f"Content : {responseMessage.content}")
    if isinstance(responseMessage, FunctionMessage):
        print(f"Function : {responseMessage.name}")
    print()

"""
Type    : HumanMessage
Content : 서울의 현재 날씨를 알려줘

Type    : AIMessage
Content :

Type    : FunctionMessage
Content : {"location": "\uc11c\uc6b8", "temperature": "20\ub3c4", "condition": "\ub9d1\uc74c"}
Function : getWeather

Type    : AIMessage
Content : 서울의 현재 날씨는 20도이며, 맑은 날씨입니다.
"""

import json

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from typing import Callable

from langchain.schema import HumanMessage

from langchain.schema import FunctionMessage

load_dotenv()

def getWeather(location : str) -> str:

# 실제로는 날씨 API를 호출하거나 DB를 조회하겠지만, 예시를 위해 하드코딩된 값을 반환한다.

weatherDictionary = {

"location" : location,

"temperature" : "20도",

"condition" : "맑음"

}

return json.dumps(weatherDictionary)

functionList = [

{

"name" : "getWeather",

"description" : "특정 도시의 현재 날씨 정보를 조회합니다",

"parameters" : {

"type" : "object",

"properties" : {

"location" : {

"type" : "string",

"description" : "도시 이름 (예 : 서울, 부산)"

}

"required" : ["location"]

}

]

chatOpenAI = ChatOpenAI(temperature = 0)

chatOpenAI.model_kwargs = {"functions" : functionList}

def processConversation(humanMessageContent : str, availableFunctionDictionary : dict[str, Callable]):

messageList = [HumanMessage(content = humanMessageContent)]

while True:

response = chatOpenAI.generate([messageList])

responseAIMessage = response.generations[0][0].message

if responseAIMessage.additional_kwargs.get("function_call"):

functionName = responseAIMessage.additional_kwargs["function_call"]["name"]

functionArgumentDictionary = json.loads(responseAIMessage.additional_kwargs["function_call"]["arguments"])

functionToCall = availableFunctionDictionary[functionName]

functionResponse = functionToCall(**functionArgumentDictionary)

messageList.append(responseAIMessage)

messageList.append(FunctionMessage(name = functionName, content = functionResponse))

else:

messageList.append(responseAIMessage)

break

return messageList

availableFunctionDictionary = {"getWeather" : getWeather}

responseMessageList = processConversation("서울의 현재 날씨를 알려줘", availableFunctionDictionary)

for responseMessage in responseMessageList:

print(f"Type : {type(responseMessage).__name__}")

print(f"Content : {responseMessage.content}")

if isinstance(responseMessage, FunctionMessage):

print(f"Function : {responseMessage.name}")

print()

"""

Type : HumanMessage

Content : 서울의 현재 날씨를 알려줘

Type : AIMessage

Content :

Type : FunctionMessage

Content : {"location": "\uc11c\uc6b8", "temperature": "20\ub3c4", "condition": "\ub9d1\uc74c"}

Function : getWeather

Type : AIMessage

Content : 서울의 현재 날씨는 20도이며, 맑은 날씨입니다.

"""

▶

[PYTHON/LANGCHAIN] BaseChatModel 클래스 : 커스텀 채팅 모델 만들기

■ BaseChatModel 클래스를 사용해 커스텀 채팅 모델을 만드는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.language_models import BaseChatModel
from typing                         import List
from langchain_core.messages        import BaseMessage
from typing                         import Optional
from langchain_core.callbacks       import CallbackManagerForLLMRun
from typing                         import Any
from langchain_core.outputs         import ChatResult
from langchain_core.messages        import AIMessage
from langchain_core.outputs         import ChatGeneration
from typing                         import Iterator
from langchain_core.outputs         import ChatGenerationChunk
from langchain_core.messages        import AIMessageChunk
from typing                         import Dict
from langchain_core.messages        import HumanMessage

class CustomChatModel(BaseChatModel):
    """A custom chat model that echoes the first `n` characters of the input.

    When contributing an implementation to LangChain, carefully document
    the model including the initialization parameters, include
    an example of how to initialize the model and include any relevant
    links to the underlying models documentation or API.

    Example:

        .. code-block:: python

            customChatModel = CustomChatModel(n = 2)

            responseAIMessage     = customChatModel.invoke([HumanMessage(content = "hello")])
            responseAIMessageList = customChatModel.batch([[HumanMessage(content = "hello")], [HumanMessage(content = "world")]])
    """

    model_name : str
    """The name of the model"""
    n : int
    """The number of characters from the last message of the prompt to be echoed."""

    def _generate(
        self,
        messages    : List[BaseMessage],
        stop        : Optional[List[str]] = None,
        run_manager : Optional[CallbackManagerForLLMRun] = None,
        **kwargs    : Any,
    ) -> ChatResult:
        """Override the _generate method to implement the chat model logic.

        This can be a call to an API, a call to a local model, or any other
        implementation that generates a response to the input prompt.

        Args:
            messages : the prompt composed of a list of messages.
            stop : a list of strings on which the model should stop generating.
                   If generation stops due to a stop token, the stop token itself
                   SHOULD BE INCLUDED as part of the output. This is not enforced
                   across models right now, but it's a good practice to follow since
                   it makes it much easier to parse the output of the model
                   downstream and understand why generation stopped.
            run_manager : A run manager with callbacks for the LLM.
        """
        # 이것을 실제 논리로 바꿔서 메시지 목록에서 응답을 생성한다.
        lastMessage = messages[-1]
        tokenCount  = lastMessage.content[: self.n]
        aiMessage   = AIMessage(
            content           = tokenCount,
            additional_kwargs = {}, # 추가 페이로드(예 : 함수 호출 요청)를 추가하는 데 사용한다.
            response_metadata = {   # 응답 메타 데이터에 사용한다.
                "time_in_seconds" : 3
            }
        )

        chatGeneration = ChatGeneration(message = aiMessage)

        return ChatResult(generations = [chatGeneration])

    def _stream(
        self,
        messages    : List[BaseMessage],
        stop        : Optional[List[str]] = None,
        run_manager : Optional[CallbackManagerForLLMRun] = None,
        **kwargs    : Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Stream the output of the model.

        This method should be implemented if the model can generate output in a streaming fashion.
        If the model does not support streaming, do not implement it.
        In that case streaming requests will be automatically handled by the _generate method.

        Args :
            messages : the prompt composed of a list of messages.
            stop : a list of strings on which the model should stop generating.
                   If generation stops due to a stop token, the stop token itself
                   SHOULD BE INCLUDED as part of the output. This is not enforced
                   across models right now, but it's a good practice to follow since
                   it makes it much easier to parse the output of the model
                   downstream and understand why generation stopped.
            run_manager : A run manager with callbacks for the LLM.
        """
        lastMessage = messages[-1]
        tokenCount  = lastMessage.content[: self.n]

        for token in tokenCount:
            chatGenerationChunk = ChatGenerationChunk(message = AIMessageChunk(content = token))

            if run_manager:
                # 이것은 LangChain의 최신 버전에서는 선택 사항이다.
                # on_llm_new_token이 자동으로 호출된다.
                run_manager.on_llm_new_token(token, chunk = chatGenerationChunk)

            yield chatGenerationChunk

        # 다른 정보(예 : 응답 메타 데이터)를 추가한다.
        chatGenerationChunk = ChatGenerationChunk(message = AIMessageChunk(content = "", response_metadata = {"time_in_sec" : 3}))

        if run_manager:
            # 이것은 LangChain의 새로운 버전에서는 선택 사항이다.
            # on_llm_new_token이 자동으로 호출된다.
            run_manager.on_llm_new_token(token, chunk = chatGenerationChunk)
        yield chatGenerationChunk

    @property
    def _llm_type(self) -> str:
        """Get the type of language model used by this chat model."""
        return "echoing-chat-model-advanced"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        """Return a dictionary of identifying parameters.

        This information is used by the LangChain callback system, which
        is used for tracing purposes make it possible to monitor LLMs.
        """
        return {
            # 모델 이름을 사용하면 사용자가 사용자 정의 토큰 계산을 지정할 수 있다.
            # LLM 모니터링 애플리케이션의 규칙(예 : LangSmith에서 사용자는 모델에 대한 토큰당 가격을 제공하고 지정된 LLM에 대한 비용을 모니터링할 수 있다)
            "model_name": self.model_name,
        }

customChatModel = CustomChatModel(n = 3, model_name = "custom_chat_model")

print("-" * 50)

responseAIMessage1 = customChatModel.invoke(
    [
        HumanMessage(content = "hello!"),
        AIMessage(content = "Hi there human!"),
        HumanMessage(content = "Meow!")
    ]
)

print(responseAIMessage1)
print("-" * 50)

responseAIMessage2 = customChatModel.invoke("hello")

print(responseAIMessage2)
print("-" * 50)

responseAIMessageList = customChatModel.batch(["hello", "goodbye"])

print(responseAIMessageList)
print("-" * 50)

for responseAIMessageChunk in customChatModel.stream("cat"):
    print(responseAIMessageChunk.content, end = "|")
print()
print("-" * 50)

async def main():
    async for eventDictionary in customChatModel.astream_events("cat", version = "v2"):
        print(eventDictionary)
    print("-" * 50)

asyncio.run(main())

"""
---------------------------------------------------
content='Meo' additional_kwargs={} response_metadata={'time_in_seconds': 3} id='run-dc5a9152-b42f-4767-b788-be2d845e5b6c-0'
---------------------------------------------------
content='hel' additional_kwargs={} response_metadata={'time_in_seconds': 3} id='run-27021883-ebfc-4419-b132-69a83f64bb17-0'
---------------------------------------------------
[AIMessage(content='hel', additional_kwargs={}, response_metadata={'time_in_seconds': 3}, id='run-e74822d3-b1b5-44f5-9bbf-190984467b03-0'), AIMessage(content='goo', additional_kwargs={}, response_metadata={'time_in_seconds': 3}, id='run-41dfe740-9519-49af-bb74-e3fab27539f0-0')]
---------------------------------------------------
c|a|t||
---------------------------------------------------
{'event': 'on_chat_model_start', 'data': {'input': 'cat'}, 'name': 'CustomChatModel', 'tags': [], 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'parent_ids': []}
{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='c', additional_kwargs={}, response_metadata={}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}
{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='a', additional_kwargs={}, response_metadata={}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}
{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='t', additional_kwargs={}, response_metadata={}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}
{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='', additional_kwargs={}, response_metadata={'time_in_sec': 3}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}
{'event': 'on_chat_model_end', 'data': {'output': AIMessageChunk(content='cat', additional_kwargs={}, response_metadata={'time_in_sec': 3}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'parent_ids': []}
---------------------------------------------------
"""

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

import asyncio

from langchain_core.language_models import BaseChatModel

from typing import List

from langchain_core.messages import BaseMessage

from typing import Optional

from langchain_core.callbacks import CallbackManagerForLLMRun

from typing import Any

from langchain_core.outputs import ChatResult

from langchain_core.messages import AIMessage

from langchain_core.outputs import ChatGeneration

from typing import Iterator

from langchain_core.outputs import ChatGenerationChunk

from langchain_core.messages import AIMessageChunk

from typing import Dict

from langchain_core.messages import HumanMessage

class CustomChatModel(BaseChatModel):

"""A custom chat model that echoes the first `n` characters of the input.

When contributing an implementation to LangChain, carefully document

the model including the initialization parameters, include

an example of how to initialize the model and include any relevant

links to the underlying models documentation or API.

Example:

.. code-block:: python

customChatModel = CustomChatModel(n = 2)

responseAIMessage = customChatModel.invoke([HumanMessage(content = "hello")])

responseAIMessageList = customChatModel.batch([[HumanMessage(content = "hello")], [HumanMessage(content = "world")]])

"""

model_name : str

"""The name of the model"""

n : int

"""The number of characters from the last message of the prompt to be echoed."""

def _generate(

self,

messages : List[BaseMessage],

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

run_manager : Optional[CallbackManagerForLLMRun] = None,

**kwargs : Any,

) -> ChatResult:

"""Override the _generate method to implement the chat model logic.

This can be a call to an API, a call to a local model, or any other

implementation that generates a response to the input prompt.

Args:

messages : the prompt composed of a list of messages.

stop : a list of strings on which the model should stop generating.

If generation stops due to a stop token, the stop token itself

SHOULD BE INCLUDED as part of the output. This is not enforced

across models right now, but it's a good practice to follow since

it makes it much easier to parse the output of the model

downstream and understand why generation stopped.

run_manager : A run manager with callbacks for the LLM.

"""

# 이것을 실제 논리로 바꿔서 메시지 목록에서 응답을 생성한다.

lastMessage = messages[-1]

tokenCount = lastMessage.content[: self.n]

aiMessage = AIMessage(

content = tokenCount,

additional_kwargs = {}, # 추가 페이로드(예 : 함수 호출 요청)를 추가하는 데 사용한다.

response_metadata = { # 응답 메타 데이터에 사용한다.

"time_in_seconds" : 3

}

)

chatGeneration = ChatGeneration(message = aiMessage)

return ChatResult(generations = [chatGeneration])

def _stream(

self,

messages : List[BaseMessage],

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

run_manager : Optional[CallbackManagerForLLMRun] = None,

**kwargs : Any,

) -> Iterator[ChatGenerationChunk]:

"""Stream the output of the model.

This method should be implemented if the model can generate output in a streaming fashion.

If the model does not support streaming, do not implement it.

In that case streaming requests will be automatically handled by the _generate method.

Args :

messages : the prompt composed of a list of messages.

stop : a list of strings on which the model should stop generating.

If generation stops due to a stop token, the stop token itself

SHOULD BE INCLUDED as part of the output. This is not enforced

across models right now, but it's a good practice to follow since

it makes it much easier to parse the output of the model

downstream and understand why generation stopped.

run_manager : A run manager with callbacks for the LLM.

"""

lastMessage = messages[-1]

tokenCount = lastMessage.content[: self.n]

for token in tokenCount:

chatGenerationChunk = ChatGenerationChunk(message = AIMessageChunk(content = token))

if run_manager:

# 이것은 LangChain의 최신 버전에서는 선택 사항이다.

# on_llm_new_token이 자동으로 호출된다.

run_manager.on_llm_new_token(token, chunk = chatGenerationChunk)

yield chatGenerationChunk

# 다른 정보(예 : 응답 메타 데이터)를 추가한다.

chatGenerationChunk = ChatGenerationChunk(message = AIMessageChunk(content = "", response_metadata = {"time_in_sec" : 3}))

if run_manager:

# 이것은 LangChain의 새로운 버전에서는 선택 사항이다.

# on_llm_new_token이 자동으로 호출된다.

run_manager.on_llm_new_token(token, chunk = chatGenerationChunk)

yield chatGenerationChunk

@property

def _llm_type(self) -> str:

"""Get the type of language model used by this chat model."""

return "echoing-chat-model-advanced"

@property

def _identifying_params(self) -> Dict[str, Any]:

"""Return a dictionary of identifying parameters.

This information is used by the LangChain callback system, which

is used for tracing purposes make it possible to monitor LLMs.

"""

return {

# 모델 이름을 사용하면 사용자가 사용자 정의 토큰 계산을 지정할 수 있다.

# LLM 모니터링 애플리케이션의 규칙(예 : LangSmith에서 사용자는 모델에 대한 토큰당 가격을 제공하고 지정된 LLM에 대한 비용을 모니터링할 수 있다)

"model_name": self.model_name,

}

customChatModel = CustomChatModel(n = 3, model_name = "custom_chat_model")

print("-" * 50)

responseAIMessage1 = customChatModel.invoke(

[

HumanMessage(content = "hello!"),

AIMessage(content = "Hi there human!"),

HumanMessage(content = "Meow!")

]

)

print(responseAIMessage1)

print("-" * 50)

responseAIMessage2 = customChatModel.invoke("hello")

print(responseAIMessage2)

print("-" * 50)

responseAIMessageList = customChatModel.batch(["hello", "goodbye"])

print(responseAIMessageList)

print("-" * 50)

for responseAIMessageChunk in customChatModel.stream("cat"):

print(responseAIMessageChunk.content, end = "|")

print()

print("-" * 50)

async def main():

async for eventDictionary in customChatModel.astream_events("cat", version = "v2"):

print(eventDictionary)

print("-" * 50)

asyncio.run(main())

"""

---------------------------------------------------

content='Meo' additional_kwargs={} response_metadata={'time_in_seconds': 3} id='run-dc5a9152-b42f-4767-b788-be2d845e5b6c-0'

---------------------------------------------------

content='hel' additional_kwargs={} response_metadata={'time_in_seconds': 3} id='run-27021883-ebfc-4419-b132-69a83f64bb17-0'

---------------------------------------------------

[AIMessage(content='hel', additional_kwargs={}, response_metadata={'time_in_seconds': 3}, id='run-e74822d3-b1b5-44f5-9bbf-190984467b03-0'), AIMessage(content='goo', additional_kwargs={}, response_metadata={'time_in_seconds': 3}, id='run-41dfe740-9519-49af-bb74-e3fab27539f0-0')]

---------------------------------------------------

c|a|t||

---------------------------------------------------

{'event': 'on_chat_model_start', 'data': {'input': 'cat'}, 'name': 'CustomChatModel', 'tags': [], 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'parent_ids': []}

{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='c', additional_kwargs={}, response_metadata={}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}

{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='a', additional_kwargs={}, response_metadata={}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}

{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='t', additional_kwargs={}, response_metadata={}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}

{'event': 'on_chat_model_stream', 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'data': {'chunk': AIMessageChunk(content='', additional_kwargs={}, response_metadata={'time_in_sec': 3}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'parent_ids': []}

{'event': 'on_chat_model_end', 'data': {'output': AIMessageChunk(content='cat', additional_kwargs={}, response_metadata={'time_in_sec': 3}, id='run-9c1a640c-8abc-4010-8aac-b950cc4cf3a8')}, 'run_id': '9c1a640c-8abc-4010-8aac-b950cc4cf3a8', 'name': 'CustomChatModel', 'tags': [], 'metadata': {'ls_provider': 'customchatmodel', 'ls_model_type': 'chat', 'ls_model_name': 'custom_chat_model'}, 'parent_ids': []}

---------------------------------------------------

"""

▶ requirements.txt


aiohappyeyeballs==2.4.3
aiohttp==3.11.7
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.144
multidict==6.1.0
numpy==1.26.4
orjson==3.10.11
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic_core==2.27.1
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.18.0

aiohappyeyeballs==2.4.3

aiohttp==3.11.7

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.144

multidict==6.1.0

numpy==1.26.4

orjson==3.10.11

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic_core==2.27.1

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.18.0

※ pip install langchain 명령을 실행했다.'

[PYTHON/LANGCHAIN] FunctionMessage 클래스 : 생성자에서 name/content 인자 사용하기

■ FunctionMessage 클래스의 생성자에서 name/content 인자를 사용하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from dotenv           import load_dotenv
from langchain.schema import HumanMessage
from langchain.schema import FunctionMessage
from langchain_openai import ChatOpenAI

load_dotenv()

messageList = [
    HumanMessage(content = "오늘 서울의 날씨는 어때?"),
    FunctionMessage(
        name    = "get_weather",
        content = "{'location' : '서울', 'temperature' : '20도', 'condition' : '맑음'}"
    )
]

chatOpenAI = ChatOpenAI()

responseAIMessage = chatOpenAI.invoke(messageList)

print(responseAIMessage)

"""
content='서울의 오늘 날씨는 맑고 기온은 20도입니다. 날씨가 매우 좋아 보입니다!' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 40, 'prompt_tokens': 53, 'total_tokens': 93, '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-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-1a28b731-fc7c-4dfd-abaf-1694c83bb0fa-0' usage_metadata={'input_tokens': 53, 'output_tokens': 40, 'total_tokens': 93, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
"""

from dotenv import load_dotenv

from langchain.schema import HumanMessage

from langchain.schema import FunctionMessage

from langchain_openai import ChatOpenAI

load_dotenv()

messageList = [

HumanMessage(content = "오늘 서울의 날씨는 어때?"),

FunctionMessage(

name = "get_weather",

content = "{'location' : '서울', 'temperature' : '20도', 'condition' : '맑음'}"

)

]

chatOpenAI = ChatOpenAI()

responseAIMessage = chatOpenAI.invoke(messageList)

print(responseAIMessage)

"""

content='서울의 오늘 날씨는 맑고 기온은 20도입니다. 날씨가 매우 좋아 보입니다!' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 40, 'prompt_tokens': 53, 'total_tokens': 93, '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-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-1a28b731-fc7c-4dfd-abaf-1694c83bb0fa-0' usage_metadata={'input_tokens': 53, 'output_tokens': 40, 'total_tokens': 93, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}

"""

▶ requirements.txt

[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] adispatch_custom_event 함수 : config 인자를 사용해 설정 정보 전달하기 (파이썬 버전 3.10 이하)

■ adispatch_custom_event 함수에서 config 인자를 사용해 설정 정보를 전달하는 방법을 보여준다. (파이썬 버전 3.10 이하) ※ 파이썬 버전 3.10 이하의 경우 수동으로

[PYTHON/LANGCHAIN] @RunnableLambda 데코레이터 사용하기

■ @RunnableLambda 데코레이터를 사용하는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.runnables         import RunnableLambda
from langchain_core.callbacks.manager import adispatch_custom_event

@RunnableLambda
async def testRunnableLambda(x : str) -> str:
    await adispatch_custom_event("event1", {"x" : x})
    await adispatch_custom_event("event2", 5)
    return x

async def main():
    asyncGenerator = testRunnableLambda.astream_events("hello world", version = "v2")
    async for eventDictionary in asyncGenerator:
        print(eventDictionary)

asyncio.run(main())

"""
{'event': 'on_chain_start', 'data': {'input': 'hello world'}, 'name': 'testRunnableLambda', 'tags': [], 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'metadata':
{}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event1', 'tags': [], 'metadata': {}, 'data': {'x': 'hello world'}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event2', 'tags': [], 'metadata': {}, 'data': 5, 'parent_ids': []}
{'event': 'on_chain_stream', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'data': {'chunk': 'hello world'}, 'parent_ids': []}
{'event': 'on_chain_end', 'data': {'output': 'hello world'}, 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'parent_ids': []}
"""

import asyncio

from langchain_core.runnables import RunnableLambda

from langchain_core.callbacks.manager import adispatch_custom_event

@RunnableLambda

async def testRunnableLambda(x : str) -> str:

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

await adispatch_custom_event("event2", 5)

return x

async def main():

asyncGenerator = testRunnableLambda.astream_events("hello world", version = "v2")

async for eventDictionary in asyncGenerator:

print(eventDictionary)

asyncio.run(main())

"""

{'event': 'on_chain_start', 'data': {'input': 'hello world'}, 'name': 'testRunnableLambda', 'tags': [], 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'metadata':

{}, 'parent_ids': []}

{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event1', 'tags': [], 'metadata': {}, 'data': {'x': 'hello world'}, 'parent_ids': []}

{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event2', 'tags': [], 'metadata': {}, 'data': 5, 'parent_ids': []}

{'event': 'on_chain_stream', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'data': {'chunk': 'hello world'}, 'parent_ids': []}

{'event': 'on_chain_end', 'data': {'output': 'hello world'}, 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'parent_ids': []}

"""

▶ 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] adispatch_custom_event 함수 : 비동기 이벤트 발생시키기

■ adispatch_custom_event 함수를 사용해 비동기 이벤트를 발생시키는 방법을 보여준다. ▶ main.py


import asyncio

from langchain_core.runnables         import RunnableLambda
from langchain_core.callbacks.manager import adispatch_custom_event

@RunnableLambda
async def testRunnableLambda(x : str) -> str:
    await adispatch_custom_event("event1", {"x" : x})
    await adispatch_custom_event("event2", 5)
    return x

async def main():
    asyncGenerator = testRunnableLambda.astream_events("hello world", version = "v2")
    async for eventDictionary in asyncGenerator:
        print(eventDictionary)

asyncio.run(main())

"""
{'event': 'on_chain_start', 'data': {'input': 'hello world'}, 'name': 'testRunnableLambda', 'tags': [], 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'metadata':
{}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event1', 'tags': [], 'metadata': {}, 'data': {'x': 'hello world'}, 'parent_ids': []}
{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event2', 'tags': [], 'metadata': {}, 'data': 5, 'parent_ids': []}
{'event': 'on_chain_stream', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'data': {'chunk': 'hello world'}, 'parent_ids': []}
{'event': 'on_chain_end', 'data': {'output': 'hello world'}, 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'parent_ids': []}
"""

import asyncio

from langchain_core.runnables import RunnableLambda

from langchain_core.callbacks.manager import adispatch_custom_event

@RunnableLambda

async def testRunnableLambda(x : str) -> str:

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

await adispatch_custom_event("event2", 5)

return x

async def main():

asyncGenerator = testRunnableLambda.astream_events("hello world", version = "v2")

async for eventDictionary in asyncGenerator:

print(eventDictionary)

asyncio.run(main())

"""

{'event': 'on_chain_start', 'data': {'input': 'hello world'}, 'name': 'testRunnableLambda', 'tags': [], 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'metadata':

{}, 'parent_ids': []}

{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event1', 'tags': [], 'metadata': {}, 'data': {'x': 'hello world'}, 'parent_ids': []}

{'event': 'on_custom_event', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'event2', 'tags': [], 'metadata': {}, 'data': 5, 'parent_ids': []}

{'event': 'on_chain_stream', 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'data': {'chunk': 'hello world'}, 'parent_ids': []}

{'event': 'on_chain_end', 'data': {'output': 'hello world'}, 'run_id': '0d42e221-af5c-49c8-9dde-7f2dc4c0f8f0', 'name': 'testRunnableLambda', 'tags': [], 'metadata': {}, 'parent_ids': []}

"""

▶ 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)

▶


from dotenv               import load_dotenv
from langchain_openai     import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt   import create_react_agent
from langgraph.errors     import GraphRecursionError

load_dotenv()

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

@tool
def magicFunction(input : int) -> int:
    """Applies a magic function to an input."""
    return input + 2

toolList = [magicFunction]

query = "what is the value of magicFunction(3)?"

RECURSION_LIMIT_COUNT = 2 * 3 + 1

compiledStateGraph = create_react_agent(chatOpenAI, tools = toolList)

try:
    for addableValuesDict in compiledStateGraph.stream(
        {"messages" : [("human", query)]},
        {"recursion_limit" : RECURSION_LIMIT_COUNT},
        stream_mode = "values"
    ):
        print(addableValuesDict["messages"][-1])
except GraphRecursionError:
    print({"input" : query, "output" : "Agent stopped due to max iterations."})

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from langchain_core.tools import tool

from langgraph.prebuilt import create_react_agent

from langgraph.errors import GraphRecursionError

load_dotenv()

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

@tool

def magicFunction(input : int) -> int:

"""Applies a magic function to an input."""

return input + 2

toolList = [magicFunction]

query = "what is the value of magicFunction(3)?"

RECURSION_LIMIT_COUNT = 2 * 3 + 1

compiledStateGraph = create_react_agent(chatOpenAI, tools = toolList)

try:

for addableValuesDict in compiledStateGraph.stream(

{"messages" : [("human", query)]},

{"recursion_limit" : RECURSION_LIMIT_COUNT},

stream_mode = "values"

print(addableValuesDict["messages"][-1])

except GraphRecursionError:

print({"input" : query, "output" : "Agent stopped due to max iterations."})

[카테고리:] LANGCHAIN

[PYTHON/LANGCHAIN] BaseBlobParser 클래스 : 커스텀 BLOB 파서 만들기

[PYTHON/LANGCHAIN] BaseLoader 클래스 : 커스텀 문서 로더 만들기

[PYTHON/LANGCHAIN] BaseRetriever 클래스 : 커스텀 검색기 만들기

[PYTHON/LANGCHAIN] LLM 클래스 : 커스텀 모델 만들기

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : model_kwargs 속성을 사용해 함수 설정하기

[PYTHON/LANGCHAIN] BaseChatModel 클래스 : 커스텀 채팅 모델 만들기

[PYTHON/LANGCHAIN] FunctionMessage 클래스 : 생성자에서 name/content 인자 사용하기

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

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

[PYTHON/LANGCHAIN] adispatch_custom_event 함수 : config 인자를 사용해 설정 정보 전달하기 (파이썬 버전 3.10 이하)

[PYTHON/LANGCHAIN] @RunnableLambda 데코레이터 사용하기

[PYTHON/LANGCHAIN] adispatch_custom_event 함수 : 비동기 이벤트 발생시키기

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

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

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

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

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

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

[PYTHON/LANGCHAIN] create_react_agent 함수 : state_modifier 인자를 사용해 장기 실행 에이전트의 중간 단계 제거하기

[PYTHON/LANGCHAIN] AgentExecutor 클래스 : 생성자에서 trim_intermediate_steps 인자를 사용해 장기 실행 에이전트의 중간 단계 제거하기

[PYTHON/LANGCHAIN] AgentExecutor 클래스 : 생성자에서 early_stopping_method/max_iterations 인자를 사용해 반복 제한/시간 제한 중단 문자열 반환하기

[PYTHON/LANGCHAIN] CompiledStateGraph 클래스 : asyncio 모듈의 create_task/wait_for 함수를 사용해 최대 실행 시간 설정하기

[PYTHON/LANGCHAIN] CompiledStateGraph 클래스 : step_timeout 속성을 사용해 최대 실행 시간 설정하기

[PYTHON/LANGCHAIN] AgentExecutor 클래스 : 생성자에서 max_execution_time 인자를 사용해 최대 실행 시간 설정하기

[PYTHON/LANGCHAIN] CompiledStateGraph 클래스 : stream 메소드에서 재귀 제한 수 설정하기