OPENAI Archives - icodebroker

[PYTHON/LANGCHAIN] create_react_agent 함수 : CompiledStateGraph 객체를 만들고 CHROMA 벡터 저장소 검색하기

■ create_react_agent 함수를 사용해 CompiledStateGraph 객체를 만들고 CHROMA 벡터 저장소를 검색하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶

[PYTHON/LANGCHAIN] 채팅 히스토리를 갖고 CHROMA 벡터 저장소 검색하기

■ 채팅 히스토리를 갖고 CHROMA 벡터 저장소를 검색하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


import bs4

from dotenv                                     import load_dotenv
from langchain_openai                           import ChatOpenAI
from langchain_community.document_loaders       import WebBaseLoader
from langchain_text_splitters                   import RecursiveCharacterTextSplitter
from langchain_chroma                           import Chroma
from langchain_openai                           import OpenAIEmbeddings
from langchain_core.prompts                     import ChatPromptTemplate
from langchain_core.prompts                     import MessagesPlaceholder
from langchain.chains                           import create_history_aware_retriever
from langchain.chains.combine_documents         import create_stuff_documents_chain
from langchain.chains                           import create_retrieval_chain
from langchain_core.chat_history                import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history           import RunnableWithMessageHistory

load_dotenv()

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

webBaseLoader = WebBaseLoader(
    web_paths = ("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs = dict(parse_only = bs4.SoupStrainer(class_ = ("post-content", "post-title", "post-header")))
)

documentList = webBaseLoader.load()

recursiveCharacterTextSplitter = RecursiveCharacterTextSplitter(chunk_size = 1000, chunk_overlap = 200)

splitDocumentList = recursiveCharacterTextSplitter.split_documents(documentList)

openAIEmbeddings = OpenAIEmbeddings()

chroma = Chroma.from_documents(documents = splitDocumentList, embedding = openAIEmbeddings)

vectorStoreRetriever = chroma.as_retriever()

systemMessage1 = "Given a chat history and the latest user question which might reference context in the chat history, formulate a standalone question which can be understood without the chat history. Do NOT answer the question, just reformulate it if needed and otherwise return it as is."

chatPromptTemplate1 = ChatPromptTemplate.from_messages(
    [
        ("system", systemMessage1),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}")
    ]
)

runnableBinding1 = create_history_aware_retriever(chatOpenAI, vectorStoreRetriever, chatPromptTemplate1)

systemMessage2 = "You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, say that you don't know. Use three sentences maximum and keep the answer concise.\n\n{context}"

chatPromptTemplate2 = ChatPromptTemplate.from_messages(
    [
        ("system", systemMessage2),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)

runnableBinding2 = create_stuff_documents_chain(chatOpenAI, chatPromptTemplate2)

runnableBinding3 = create_retrieval_chain(runnableBinding1, runnableBinding2)

chatMessageHistoryDictionary = {}

def GetChatMessageHistoryDictionary(session_id : str) -> BaseChatMessageHistory:
    if session_id not in chatMessageHistoryDictionary:
        chatMessageHistoryDictionary[session_id] = ChatMessageHistory()
    return chatMessageHistoryDictionary[session_id]

runnableWithMessageHistory = RunnableWithMessageHistory(
    runnableBinding3,
    GetChatMessageHistoryDictionary,
    input_messages_key   = "input",
    history_messages_key = "chat_history",
    output_messages_key  = "answer",
)

responseDictionary1 = runnableWithMessageHistory.invoke(
    {"input" : "What is Task Decomposition?"},
    config = {"configurable" : {"session_id" : "abc123"}}
)

answer1 = responseDictionary1["answer"]

print(answer1)
print("-" * 50)

"""
Task Decomposition is a technique used to break down complex tasks into smaller, manageable steps. It is often implemented using methods like Chain of Thought (CoT) or Tree of Thoughts, which help in systematically exploring and reasoning through various possibilities. This approach enhances model performance by allowing a step-by-step analysis and execution of tasks.
"""

responseDictionary2 = runnableWithMessageHistory.invoke(
    {"input" : "What are common ways of doing it?"},
    config = {"configurable" : {"session_id" : "abc123"}}
)

answer2 = responseDictionary2["answer"]

print(answer2)
print("-" * 50)

from langchain_core.messages import AIMessage

for message in chatMessageHistoryDictionary["abc123"].messages:
    if isinstance(message, AIMessage):
        prefix = "AI"
    else:
        prefix = "User"

    print(f"{prefix} : {message.content}")
print("-" * 50)

"""
Task decomposition is the process of breaking down a complex task into smaller, more manageable steps or subgoals. This approach, often used in conjunction with techniques like Chain of Thought (CoT), helps enhance model performance by enabling step-by-step reasoning. It can be achieved through prompting, task-specific instructions, or human inputs.
--------------------------------------------------
Common ways of performing task decomposition include using straightforward prompts like "Steps for XYZ.\n1." or "What are the subgoals for achieving XYZ?", employing
task-specific instructions such as "Write a story outline" for writing a novel, and incorporating human inputs.
--------------------------------------------------
User : What is Task Decomposition?
AI : Task decomposition is the process of breaking down a complex task into smaller, more manageable steps or subgoals. This approach, often used in conjunction with
techniques like Chain of Thought (CoT), helps enhance model performance by enabling step-by-step reasoning. It can be achieved through prompting, task-specific instructions, or human inputs.
User : What are common ways of doing it?
AI : Common ways of performing task decomposition include using straightforward prompts like "Steps for XYZ.\n1." or "What are the subgoals for achieving XYZ?", employing task-specific instructions such as "Write a story outline" for writing a novel, and incorporating human inputs.
--------------------------------------------------
"""

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

import bs4

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from langchain_community.document_loaders import WebBaseLoader

from langchain_text_splitters import RecursiveCharacterTextSplitter

from langchain_chroma import Chroma

from langchain_openai import OpenAIEmbeddings

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.prompts import MessagesPlaceholder

from langchain.chains import create_history_aware_retriever

from langchain.chains.combine_documents import create_stuff_documents_chain

from langchain.chains import create_retrieval_chain

from langchain_core.chat_history import BaseChatMessageHistory

from langchain_community.chat_message_histories import ChatMessageHistory

from langchain_core.runnables.history import RunnableWithMessageHistory

load_dotenv()

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

webBaseLoader = WebBaseLoader(

web_paths = ("https://lilianweng.github.io/posts/2023-06-23-agent/",),

bs_kwargs = dict(parse_only = bs4.SoupStrainer(class_ = ("post-content", "post-title", "post-header")))

)

documentList = webBaseLoader.load()

recursiveCharacterTextSplitter = RecursiveCharacterTextSplitter(chunk_size = 1000, chunk_overlap = 200)

splitDocumentList = recursiveCharacterTextSplitter.split_documents(documentList)

openAIEmbeddings = OpenAIEmbeddings()

chroma = Chroma.from_documents(documents = splitDocumentList, embedding = openAIEmbeddings)

vectorStoreRetriever = chroma.as_retriever()

systemMessage1 = "Given a chat history and the latest user question which might reference context in the chat history, formulate a standalone question which can be understood without the chat history. Do NOT answer the question, just reformulate it if needed and otherwise return it as is."

chatPromptTemplate1 = ChatPromptTemplate.from_messages(

[

("system", systemMessage1),

MessagesPlaceholder("chat_history"),

("human", "{input}")

]

)

runnableBinding1 = create_history_aware_retriever(chatOpenAI, vectorStoreRetriever, chatPromptTemplate1)

systemMessage2 = "You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, say that you don't know. Use three sentences maximum and keep the answer concise.\n\n{context}"

chatPromptTemplate2 = ChatPromptTemplate.from_messages(

[

("system", systemMessage2),

MessagesPlaceholder("chat_history"),

("human", "{input}"),

]

)

runnableBinding2 = create_stuff_documents_chain(chatOpenAI, chatPromptTemplate2)

runnableBinding3 = create_retrieval_chain(runnableBinding1, runnableBinding2)

chatMessageHistoryDictionary = {}

def GetChatMessageHistoryDictionary(session_id : str) -> BaseChatMessageHistory:

if session_id not in chatMessageHistoryDictionary:

chatMessageHistoryDictionary[session_id] = ChatMessageHistory()

return chatMessageHistoryDictionary[session_id]

runnableWithMessageHistory = RunnableWithMessageHistory(

runnableBinding3,

GetChatMessageHistoryDictionary,

input_messages_key = "input",

history_messages_key = "chat_history",

output_messages_key = "answer",

)

responseDictionary1 = runnableWithMessageHistory.invoke(

{"input" : "What is Task Decomposition?"},

config = {"configurable" : {"session_id" : "abc123"}}

)

answer1 = responseDictionary1["answer"]

print(answer1)

print("-" * 50)

"""

Task Decomposition is a technique used to break down complex tasks into smaller, manageable steps. It is often implemented using methods like Chain of Thought (CoT) or Tree of Thoughts, which help in systematically exploring and reasoning through various possibilities. This approach enhances model performance by allowing a step-by-step analysis and execution of tasks.

"""

responseDictionary2 = runnableWithMessageHistory.invoke(

{"input" : "What are common ways of doing it?"},

config = {"configurable" : {"session_id" : "abc123"}}

)

answer2 = responseDictionary2["answer"]

print(answer2)

print("-" * 50)

from langchain_core.messages import AIMessage

for message in chatMessageHistoryDictionary["abc123"].messages:

if isinstance(message, AIMessage):

prefix = "AI"

else:

prefix = "User"

print(f"{prefix} : {message.content}")

print("-" * 50)

"""

Task decomposition is the process of breaking down a complex task into smaller, more manageable steps or subgoals. This approach, often used in conjunction with techniques like Chain of Thought (CoT), helps enhance model performance by enabling step-by-step reasoning. It can be achieved through prompting, task-specific instructions, or human inputs.

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

Common ways of performing task decomposition include using straightforward prompts like "Steps for XYZ.\n1." or "What are the subgoals for achieving XYZ?", employing

task-specific instructions such as "Write a story outline" for writing a novel, and incorporating human inputs.

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

User : What is Task Decomposition?

AI : Task decomposition is the process of breaking down a complex task into smaller, more manageable steps or subgoals. This approach, often used in conjunction with

techniques like Chain of Thought (CoT), helps enhance model performance by enabling step-by-step reasoning. It can be achieved through prompting, task-specific instructions, or human inputs.

User : What are common ways of doing it?

AI : Common ways of performing task decomposition include using straightforward prompts like "Steps for XYZ.\n1." or "What are the subgoals for achieving XYZ?", employing task-specific instructions such as "Write a story outline" for writing a novel, and incorporating human inputs.

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

"""

▶

[PYTHON/LANGCHAIN] loads 함수 : JSON 파일에서 RunnableSequence 객체 만들기

■ loads 함수를 사용해 JSON 파일에서 RunnableSequence 객체를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


import json
import os

from dotenv                 import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai       import ChatOpenAI
from langchain_core.load    import dumps
from langchain_core.load    import loads

load_dotenv()

chatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        ("system", "Translate the following text into {language} :"),
        ("user"  , "{text}")
    ]
)

chatOpenAI = ChatOpenAI(
    model       = "gpt-4",
    temperature = 0.7
)

runnableSequence1 = chatPromptTemplate | chatOpenAI

jsonString = dumps(runnableSequence1, pretty = True)

with open("chain.json", "w") as textIOWrapper:
    json.dump(jsonString, textIOWrapper)

with open("chain.json", "r") as textIOWrapper:
    runnableSequence2 = loads(json.load(textIOWrapper), secrets_map = {"OPENAI_API_KEY" : os.environ["OPENAI_API_KEY"]})

    try:
        response = runnableSequence2.invoke(
            {
                "language": "Korean",
                "text"    : "I am a student"
            }
        )
        print("번역 결과 :", response.content)
    except Exception as exception:
        print(f"오류 발생 : {str(exception)}")

import json

import os

from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.load import dumps

from langchain_core.load import loads

load_dotenv()

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

("system", "Translate the following text into {language} :"),

("user" , "{text}")

]

)

chatOpenAI = ChatOpenAI(

model = "gpt-4",

temperature = 0.7

)

runnableSequence1 = chatPromptTemplate | chatOpenAI

jsonString = dumps(runnableSequence1, pretty = True)

with open("chain.json", "w") as textIOWrapper:

json.dump(jsonString, textIOWrapper)

with open("chain.json", "r") as textIOWrapper:

runnableSequence2 = loads(json.load(textIOWrapper), secrets_map = {"OPENAI_API_KEY" : os.environ["OPENAI_API_KEY"]})

try:

response = runnableSequence2.invoke(

{

"language": "Korean",

"text" : "I am a student"

}

)

print("번역 결과 :", response.content)

except Exception as exception:

print(f"오류 발생 : {str(exception)}")

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : openai_api_key 속성을 사용해 OPENAI_API_KEY 값 구하기

■ ChatOpenAI 클래스의 openai_api_key 속성을 사용해 OPENAI_API_KEY 값을 구하는 방법을 보여준다. ▶ main.py


from dotenv           import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

chatOpenAI = ChatOpenAI(
    model       = "gpt-4",
    temperature = 0.7
)

secretStr = chatOpenAI.openai_api_key

secretValue = secretStr.get_secret_value()

print(secretValue)

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

load_dotenv()

chatOpenAI = ChatOpenAI(

model = "gpt-4",

temperature = 0.7

)

secretStr = chatOpenAI.openai_api_key

secretValue = secretStr.get_secret_value()

print(secretValue)

▶ requirements.txt


annotated-types==0.7.0
anyio==4.6.2.post1
certifi==2024.8.30
charset-normalizer==3.4.0
colorama==0.4.6
distro==1.9.0
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-core==0.3.21
langchain-openai==0.2.9
langsmith==0.1.146
openai==1.55.1
orjson==3.10.12
packaging==24.2
pydantic==2.10.1
pydantic_core==2.27.1
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
tenacity==9.0.0
tiktoken==0.8.0
tqdm==4.67.1
typing_extensions==4.12.2
urllib3==2.2.3

annotated-types==0.7.0

anyio==4.6.2.post1

certifi==2024.8.30

charset-normalizer==3.4.0

colorama==0.4.6

distro==1.9.0

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-core==0.3.21

langchain-openai==0.2.9

langsmith==0.1.146

openai==1.55.1

orjson==3.10.12

packaging==24.2

pydantic==2.10.1

pydantic_core==2.27.1

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

tenacity==9.0.0

tiktoken==0.8.0

tqdm==4.67.1

typing_extensions==4.12.2

urllib3==2.2.3

※ install python-dotenv langchain-openai 명령을

[PYTHON/LANGCHAIN] load 함수 : JSON 직렬화 가능 딕셔너리에서 RunnableSequence 객체 만들기

■ load 함수를 사용해 SON 직렬화 가능 딕셔너리에서 RunnableSequence 객체를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶

[PYTHON/LANGCHAIN] loads 함수 : JSON 문자열에서 RunnableSequence 객체 만들기

■ loads 함수를 사용해 JSON 문자열에서 RunnableSequence 객체를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


import os

from dotenv                 import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai       import ChatOpenAI
from langchain_core.load    import dumps
from langchain_core.load    import loads

load_dotenv()

chatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        ("system", "Translate the following text into {language} :"),
        ("user"  , "{text}")
    ]
)

chatOpenAI = ChatOpenAI(
    model       = "gpt-4",
    temperature = 0.7
)

runnableSequence1 = chatPromptTemplate | chatOpenAI

json = dumps(runnableSequence1, pretty = True)

runnableSequence2 = loads(json, secrets_map = {"OPENAI_API_KEY" : os.environ["OPENAI_API_KEY"]})

try:
    responseAIMessage = runnableSequence2.invoke(
        {
            "language": "Korean",
            "text": "I am a student"
        }
    )
    print("번역 결과 :", responseAIMessage.content)
except Exception as exception:
    print(f"오류 발생 : {str(exception)}")

"""
번역 결과 : 저는 학생입니다.
"""

import os

from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.load import dumps

from langchain_core.load import loads

load_dotenv()

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

("system", "Translate the following text into {language} :"),

("user" , "{text}")

]

)

chatOpenAI = ChatOpenAI(

model = "gpt-4",

temperature = 0.7

)

runnableSequence1 = chatPromptTemplate | chatOpenAI

json = dumps(runnableSequence1, pretty = True)

runnableSequence2 = loads(json, secrets_map = {"OPENAI_API_KEY" : os.environ["OPENAI_API_KEY"]})

try:

responseAIMessage = runnableSequence2.invoke(

{

"language": "Korean",

"text": "I am a student"

}

)

print("번역 결과 :", responseAIMessage.content)

except Exception as exception:

print(f"오류 발생 : {str(exception)}")

"""

번역 결과 : 저는 학생입니다.

"""

[PYTHON/LANGCHAIN] BaseOutputParser 클래스 : 커스텀 출력 파서를 체인에서 사용하기

■ BaseOutputParser 클래스를 사용해 커스텀 출력 파서를 체인에서 사용하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from dotenv                        import load_dotenv
from langchain_core.output_parsers import BaseOutputParser
from langchain_core.exceptions     import OutputParserException
from langchain_openai              import ChatOpenAI

load_dotenv()

class CustomOutputParser(BaseOutputParser[bool]):
    """Custom boolean parser."""

    trueString  : str = "YES"
    falseString : str = "NO"

    def parse(self, text : str) -> bool:
        textCleaned = text.strip().upper()
        if textCleaned not in (self.trueString.upper(), self.falseString.upper()):
            raise OutputParserException(
                f"BooleanOutputParser expected output value to either be "
                f"{self.trueString} or {self.falseString} (case-insensitive). "
                f"Received {textCleaned}."
            )
        return textCleaned == self.trueString.upper()

    @property
    def _type(self) -> str:
        return "custom_output_parser"

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

customOutputParser = CustomOutputParser()

runnableSequence = chatOpenAI | customOutputParser

responseString = runnableSequence.invoke("say YES or NO")

print(responseString)

"""
True
"""

from dotenv import load_dotenv

from langchain_core.output_parsers import BaseOutputParser

from langchain_core.exceptions import OutputParserException

from langchain_openai import ChatOpenAI

load_dotenv()

class CustomOutputParser(BaseOutputParser[bool]):

"""Custom boolean parser."""

trueString : str = "YES"

falseString : str = "NO"

def parse(self, text : str) -> bool:

textCleaned = text.strip().upper()

if textCleaned not in (self.trueString.upper(), self.falseString.upper()):

raise OutputParserException(

f"BooleanOutputParser expected output value to either be "

f"{self.trueString} or {self.falseString} (case-insensitive). "

f"Received {textCleaned}."

)

return textCleaned == self.trueString.upper()

@property

def _type(self) -> str:

return "custom_output_parser"

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

customOutputParser = CustomOutputParser()

runnableSequence = chatOpenAI | customOutputParser

responseString = runnableSequence.invoke("say YES or NO")

print(responseString)

"""

True

"""

[PYTHON/LANGCHAIN] RunnableGenerator 클래스 : 커스텀 출력 파서 만들기

■ RunnableGenerator 클래스를 사용해 커스텀 출력 파서를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from dotenv                   import load_dotenv
from langchain_openai         import ChatOpenAI
from langchain_core.messages  import AIMessageChunk
from typing                   import Iterable
from langchain_core.runnables import RunnableGenerator

load_dotenv()

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

def streamingParse(tee : Iterable[AIMessageChunk]) -> Iterable[str]: # tee의 데이터 타입은 itertools._tee이다.
    for chunk in tee: # chunk 데이터 타입은 AIMessage 또는 AIMessageChunk가 된다.
        yield chunk.content.swapcase()

runnableGenerator = RunnableGenerator(streamingParse)

runnableSequence = chatOpenAI | runnableGenerator

responseString = runnableSequence.invoke("hello")

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

for chunkString in runnableSequence.stream("tell me about yourself in one sentence"):
    print(chunkString, end = "|", flush = True)
print()
print("-" * 50)

"""
hELLO! hOW CAN i ASSIST YOU TODAY?
--------------------------------------------------
|i'M| A| LANGUAGE| MODEL| DESIGNED| BY| oPEN|ai| TO| ASSIST| WITH| A| WIDE| RANGE| OF| QUESTIONS| AND| TASKS| BY| PROVIDING| INFORMATION| AND| GENERATING| TEXT|-BASED| RESPONSES|.||
--------------------------------------------------
"""

from dotenv import load_dotenv

from langchain_openai import ChatOpenAI

from langchain_core.messages import AIMessageChunk

from typing import Iterable

from langchain_core.runnables import RunnableGenerator

load_dotenv()

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

def streamingParse(tee : Iterable[AIMessageChunk]) -> Iterable[str]: # tee의 데이터 타입은 itertools._tee이다.

for chunk in tee: # chunk 데이터 타입은 AIMessage 또는 AIMessageChunk가 된다.

yield chunk.content.swapcase()

runnableGenerator = RunnableGenerator(streamingParse)

runnableSequence = chatOpenAI | runnableGenerator

responseString = runnableSequence.invoke("hello")

print(responseString)

print("-" * 50)

for chunkString in runnableSequence.stream("tell me about yourself in one sentence"):

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

print()

print("-" * 50)

"""

hELLO! hOW CAN i ASSIST YOU TODAY?

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

|i'M| A| LANGUAGE| MODEL| DESIGNED| BY| oPEN|ai| TO| ASSIST| WITH| A| WIDE| RANGE| OF| QUESTIONS| AND| TASKS| BY| PROVIDING| INFORMATION| AND| GENERATING| TEXT|-BASED| RESPONSES|.||

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

"""

▶

[PYTHON/LANGCHAIN] RunnableLambda 클래스 : 커스텀 출력 파서 만들기

■ RunnableLambda 클래스를 사용해 커스텀 출력 파서를 만드는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ※ LCEL은 | 구문을

[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] 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] 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)

▶

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

■ create_react_agent 함수의 state_modifier 인자를 사용해 장기 실행 에이전트의 중간 단계를 제거하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

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

■ AgentExecutor 클래스의 생성자에서 trim_intermediate_steps 인자를 사용해 장기 실행 에이전트의 중간 단계를 제거하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에

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

■ AgentExecutor 클래스의 생성자에서 early_stopping_method/max_iterations 인자를 사용해 반복 제한/시간 제한 중단 문자열을 반환하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에

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

■ CompiledStateGraph 클래스에서 asyncio 패키지의 create_task/wait_for 함수를 사용해 최대 실행 시간을 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

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

■ CompiledStateGraph 클래스의 step_timeout 속성을 사용해 최대 실행 시간을 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py

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

■ AgentExecutor 클래스의 생성자에서 max_execution_time 인자를 사용해 최대 실행 시간을 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶

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

■ CompiledStateGraph 클래스의 stream 메소드에서 재귀 제한 수를 설정하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


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."})

[PYTHON/LANGCHAIN] AgentExecutor 클래스 : 생성자에서 max_iterations 인자를 사용해 사용자 지정 반복 횟수 초과 실행시 중단하기

■ AgentExecutor 클래스의 생성자에서 max_iterations 인자를 사용해 사용자 지정 반복 횟수 초과 실행시 중단하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env