icodebroker - 686 중 5 번째 페이지 - 컴퓨터 프로그래밍 코드 데이터베이스

[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] RunnableSequence 클래스 : last 속성을 사용해 마지막 Runnable 객체 구하기

■ RunnableSequence 클래스의 last 속성을 사용해 마지막 Runnable 객체를 구하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py

[PYTHON/LANGCHAIN] dump 함수 : RunnableSequence 객체에서 JSON 파일 생성하기

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


import json

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

load_dotenv()

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

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

runnableSequence = chatPromptTemplate | chatOpenAI

jsonString = dumps(runnableSequence, pretty = True)

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

import json

from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.load import dumps

load_dotenv()

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

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

("user" , "{text}")

]

)

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

runnableSequence = chatPromptTemplate | chatOpenAI

jsonString = dumps(runnableSequence, pretty = True)

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

json.dump(jsonString, textIOWrapper)

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

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

[PYTHON/LANGCHAIN] dumps 함수 : RunnableSequence 객체에서 JSON 문자열 구하기

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


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

load_dotenv()

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

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

runnableSequence = chatPromptTemplate | chatOpenAI

json = dumps(runnableSequence, pretty = True)

print(json)

"""
{
  "lc": 1,
  "type": "constructor",
  "id": [
    "langchain",
    "schema",
    "runnable",
    "RunnableSequence"
  ],
  "kwargs": {
    "first": {
      "lc": 1,
      "type": "constructor",
      "id": [
        "langchain",
        "prompts",
        "chat",
        "ChatPromptTemplate"
      ],
      "kwargs": {
        "input_variables": [
          "language",
          "text"
        ],
        "messages": [
          {
            "lc": 1,
            "type": "constructor",
            "id": [
              "langchain",
              "prompts",
              "chat",
              "SystemMessagePromptTemplate"
            ],
            "kwargs": {
              "prompt": {
                "lc": 1,
                "type": "constructor",
                "id": [
                  "langchain",
                  "prompts",
                  "prompt",
                  "PromptTemplate"
                ],
                "kwargs": {
                  "input_variables": [
                    "language"
                  ],
                  "template": "Translate the following into {language}:",
                  "template_format": "f-string"
                },
                "name": "PromptTemplate"
              }
            }
          },
          {
            "lc": 1,
            "type": "constructor",
            "id": [
              "langchain",
              "prompts",
              "chat",
              "HumanMessagePromptTemplate"
            ],
            "kwargs": {
              "prompt": {
                "lc": 1,
                "type": "constructor",
                "id": [
                  "langchain",
                  "prompts",
                  "prompt",
                  "PromptTemplate"
                ],
                "kwargs": {
                  "input_variables": [
                    "text"
                  ],
                  "template": "{text}",
                  "template_format": "f-string"
                },
                "name": "PromptTemplate"
              }
            }
          }
        ]
      },
      "name": "ChatPromptTemplate"
    },
    "last": {
      "lc": 1,
      "type": "constructor",
      "id": [
        "langchain",
        "chat_models",
        "openai",
        "ChatOpenAI"
      ],
      "kwargs": {
        "model_name": "gpt-4o",
        "temperature": 0.7,
        "openai_api_key": {
          "lc": 1,
          "type": "secret",
          "id": [
            "OPENAI_API_KEY"
          ]
        },
        "max_retries": 2,
        "n": 1
      },
      "name": "ChatOpenAI"
    }
  },
  "name": "RunnableSequence"
}
"""

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

from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.load import dumps

load_dotenv()

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

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

("user" , "{text}")

]

)

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

runnableSequence = chatPromptTemplate | chatOpenAI

json = dumps(runnableSequence, pretty = True)

print(json)

"""

{

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"schema",

"runnable",

"RunnableSequence"

"kwargs": {

"first": {

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"prompts",

"chat",

"ChatPromptTemplate"

"kwargs": {

"input_variables": [

"language",

"text"

"messages": [

{

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"prompts",

"chat",

"SystemMessagePromptTemplate"

"kwargs": {

"prompt": {

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"prompts",

"prompt",

"PromptTemplate"

"kwargs": {

"input_variables": [

"language"

"template": "Translate the following into {language}:",

"template_format": "f-string"

"name": "PromptTemplate"

}

{

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"prompts",

"chat",

"HumanMessagePromptTemplate"

"kwargs": {

"prompt": {

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"prompts",

"prompt",

"PromptTemplate"

"kwargs": {

"input_variables": [

"text"

"template": "{text}",

"template_format": "f-string"

"name": "PromptTemplate"

}

]

"name": "ChatPromptTemplate"

"last": {

"lc": 1,

"type": "constructor",

"id": [

"langchain",

"chat_models",

"openai",

"ChatOpenAI"

"kwargs": {

"model_name": "gpt-4o",

"temperature": 0.7,

"openai_api_key": {

"lc": 1,

"type": "secret",

"id": [

"OPENAI_API_KEY"

]

"max_retries": 2,

"n": 1

"name": "ChatOpenAI"

}

"name": "RunnableSequence"

}

"""

[PYTHON/LANGCHAIN] @tool 데코레이터 : args_schema 인자에 BaseModel 파생 객체 설정하기

■ @tool 데코레이터의 args_schema 인자에 BaseModel 파생 객체를 설정하는 방법을 보여준다. ▶ main.py


from pydantic             import BaseModel
from pydantic             import Field
from langchain_core.tools import tool

class CalculatorInput(BaseModel):
    a : int = Field(description = "first number" )
    b : int = Field(description = "second number")

@tool("multiplication-tool", args_schema = CalculatorInput, return_direct = True)
def multiply(a : int, b : int) -> int:
    """Multiply two numbers."""
    return a * b

print(multiply.name)
print(multiply.description)
print(multiply.args_schema.model_json_schema())
print(multiply.return_direct)

"""
multiplication-tool
Multiply two numbers.
{'properties': {'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'CalculatorInput', 'type': 'object'}
True
"""

from pydantic import BaseModel

from pydantic import Field

from langchain_core.tools import tool

class CalculatorInput(BaseModel):

a : int = Field(description = "first number" )

b : int = Field(description = "second number")

@tool("multiplication-tool", args_schema = CalculatorInput, return_direct = True)

def multiply(a : int, b : int) -> int:

"""Multiply two numbers."""

return a * b

print(multiply.name)

print(multiply.description)

print(multiply.args_schema.model_json_schema())

print(multiply.return_direct)

"""

multiplication-tool

Multiply two numbers.

{'properties': {'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'CalculatorInput', 'type': 'object'}

True

"""

▶ 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] @tool 데코레이터 : 대상 함수의 인자에 Annotated 객체를 사용해 인자 구문 분석 지원하기

■ @tool 데코레이터를 사용해 대상 함수의 인자에 Annotated 객체를 사용해 인자 구문 분석을 지원하는 방법을 보여준다. ▶ main.py


from langchain_core.tools import tool
from typing               import Annotated
from typing               import List

@tool
def multiplyByMaximumValue(
    value     : Annotated[str      , "scale factor"],
    valueList : Annotated[List[int], "list of ints over which to take maximum"]
) -> int:
    """Multiply a by the maximum of b."""
    return value * max(valueList)

modelMetaclass = multiplyByMaximumValue.args_schema

modelJSONSchemaDictionary = modelMetaclass.model_json_schema()

print(modelJSONSchemaDictionary)

"""
{
    'description' : 'Multiply a by the maximum of b.',
    'properties'  : {
        'value'     : {
            'description' : 'scale factor',
            'title'       : 'Value',
            'type'        : 'string'
        },
        'valueList' : {
            'description' : 'list of ints over which to take maximum',
            'items'       : {'type' : 'integer'},
            'title'       : 'Valuelist',
            'type'        : 'array'
        }
    },
    'required'    : ['value', 'valueList'],
    'title'       :  'multiplyByMaximumValue',
    'type'        : 'object'
}
"""

from langchain_core.tools import tool

from typing import Annotated

from typing import List

@tool

def multiplyByMaximumValue(

value : Annotated[str , "scale factor"],

valueList : Annotated[List[int], "list of ints over which to take maximum"]

) -> int:

"""Multiply a by the maximum of b."""

return value * max(valueList)

modelMetaclass = multiplyByMaximumValue.args_schema

modelJSONSchemaDictionary = modelMetaclass.model_json_schema()

print(modelJSONSchemaDictionary)

"""

{

'description' : 'Multiply a by the maximum of b.',

'properties' : {

'value' : {

'description' : 'scale factor',

'title' : 'Value',

'type' : 'string'

'valueList' : {

'description' : 'list of ints over which to take maximum',

'items' : {'type' : 'integer'},

'title' : 'Valuelist',

'type' : 'array'

}

'required' : ['value', 'valueList'],

'title' : 'multiplyByMaximumValue',

'type' : 'object'

}

"""

▶ 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

[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] @tool 데코레이터 : 비동기 함수를 StructuredTool 객체 만들기

■ @tool 데코레이터를 사용해 비동기 함수를 StructuredTool 객체로 만드는 방법을 보여준다. ▶ main.py


from langchain_core.tools import tool

@tool
async def multiplyAsync(a : int, b : int) -> int:
    """Multiply two numbers."""
    return a * b

print(type(multiplyAsync))
print(multiplyAsync.name)
print(multiplyAsync.description)
print(multiplyAsync.args)

"""
<class 'langchain_core.tools.structured.StructuredTool'>
multiplyAsync
Multiply two numbers.
{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}
"""

from langchain_core.tools import tool

@tool

async def multiplyAsync(a : int, b : int) -> int:

"""Multiply two numbers."""

return a * b

print(type(multiplyAsync))

print(multiplyAsync.name)

print(multiplyAsync.description)

print(multiplyAsync.args)

"""

multiplyAsync

Multiply two numbers.

{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}

"""

▶ 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] @tool 데코레이터 : 함수를 StructuredTool 객체 만들기

■ @tool 데코레이터를 사용해 함수를 StructuredTool 객체로 만드는 방법을 보여준다. ▶ main.py


from langchain_core.tools import tool

@tool
def multiply(a : int, b : int) -> int:
    """Multiply two numbers."""
    return a * b

print(type(multiply))
print(multiply.name)
print(multiply.description)
print(multiply.args)

"""
<class 'langchain_core.tools.structured.StructuredTool'>
multiply
Multiply two numbers.
{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}
"""

from langchain_core.tools import tool

@tool

def multiply(a : int, b : int) -> int:

"""Multiply two numbers."""

return a * b

print(type(multiply))

print(multiply.name)

print(multiply.description)

print(multiply.args)

"""

multiply

Multiply two numbers.

{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}

"""

▶ 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] BaseOutputParser 클래스 : 커스텀 출력 파서 만들기

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


from langchain_core.output_parsers import BaseOutputParser
from langchain_core.exceptions     import OutputParserException

# [bool]은 제네릭의 매개변수화를 설명한다.
# 기본적으로 parse의 반환 유형이 무엇인지 나타낸다. 이 경우 반환 유형은 True 또는 False이다.
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"

customOutputParser = CustomOutputParser()

responseValue1 = customOutputParser.invoke("YES")

print(responseValue1)
print("-" * 50)

try:
    customOutputParser.invoke("MEOW")
except Exception as exception:
    print(f"Triggered an exception of type: {type(exception)}")
print("-" * 50)

customOutputParser2 = CustomOutputParser(trueString = "OKAY")

responseValue2 = customOutputParser2.invoke("OKAY")

print(responseValue2)
print("-" * 50)

responseValueList = customOutputParser2.batch(["OKAY", "NO"])

print(responseValueList)
print("-" * 50)

"""
True
--------------------------------------------------
Triggered an exception of type: <class 'langchain_core.exceptions.OutputParserException'>
--------------------------------------------------
True
--------------------------------------------------
[True, False]
--------------------------------------------------
"""

from langchain_core.output_parsers import BaseOutputParser

from langchain_core.exceptions import OutputParserException

# [bool]은 제네릭의 매개변수화를 설명한다.

# 기본적으로 parse의 반환 유형이 무엇인지 나타낸다. 이 경우 반환 유형은 True 또는 False이다.

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"

customOutputParser = CustomOutputParser()

responseValue1 = customOutputParser.invoke("YES")

print(responseValue1)

print("-" * 50)

try:

customOutputParser.invoke("MEOW")

except Exception as exception:

print(f"Triggered an exception of type: {type(exception)}")

print("-" * 50)

customOutputParser2 = CustomOutputParser(trueString = "OKAY")

responseValue2 = customOutputParser2.invoke("OKAY")

print(responseValue2)

print("-" * 50)

responseValueList = customOutputParser2.batch(["OKAY", "NO"])

print(responseValueList)

print("-" * 50)

"""

True

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

Triggered an exception of type: <class 'langchain_core.exceptions.OutputParserException'>

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

True

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

[True, False]

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

"""

▶ 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] 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] GenericLoader 클래스 : 커스텀 로더 만들기

■ GenericLoader 클래스를 사용해 커스텀 로더를 만드는 방법을 보여준다. ▶ 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
from langchain_community.document_loaders.generic import GenericLoader
from typing                                       import Any

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

class CustomLoader(GenericLoader):
    @staticmethod
    def get_parser(**keywordArgumentArray : Any) -> BaseBlobParser:
        """Override this method to associate a default parser with the class."""
        return CustomBlobParser()

customLoader = CustomLoader.from_filesystem(path = ".", glob = "*.txt", show_progress = True)

generator = customLoader.lazy_load()

for index, document in enumerate(generator):
    print(document)

from langchain_core.document_loaders import BaseBlobParser

from langchain_core.document_loaders import Blob

from typing import Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.generic import GenericLoader

from typing import Any

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}

)

class CustomLoader(GenericLoader):

@staticmethod

def get_parser(**keywordArgumentArray : Any) -> BaseBlobParser:

"""Override this method to associate a default parser with the class."""

return CustomBlobParser()

customLoader = CustomLoader.from_filesystem(path = ".", glob = "*.txt", show_progress = True)

generator = customLoader.lazy_load()

for index, document in enumerate(generator):

print(document)

▶ 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
colorama==0.4.6
dataclasses-json==0.6.7
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.27.2
httpx-sse==0.4.0
idna==3.10
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.8
langchain-community==0.3.8
langchain-core==0.3.21
langchain-text-splitters==0.3.2
langsmith==0.1.145
marshmallow==3.23.1
multidict==6.1.0
mypy-extensions==1.0.0
numpy==1.26.4
orjson==3.10.12
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic-settings==2.6.1
pydantic_core==2.27.1
python-dotenv==1.0.1
PyYAML==6.0.2
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.35
tenacity==9.0.0
tqdm==4.67.0
typing-inspect==0.9.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

colorama==0.4.6

dataclasses-json==0.6.7

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.27.2

httpx-sse==0.4.0

idna==3.10

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.8

langchain-community==0.3.8

langchain-core==0.3.21

langchain-text-splitters==0.3.2

langsmith==0.1.145

marshmallow==3.23.1

multidict==6.1.0

mypy-extensions==1.0.0

numpy==1.26.4

orjson==3.10.12

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic-settings==2.6.1

pydantic_core==2.27.1

python-dotenv==1.0.1

PyYAML==6.0.2

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.35

tenacity==9.0.0

tqdm==4.67.0

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.18.0

※ pip install langchain langchain-community tqdm 명령을

[PYTHON/LANGCHAIN] GenericLoader 클래스 : from_filesystem 메소드에서 path/glob/show_progress/parser 인자를 사용해 GenericLoader 객체 만들기

■ GenericLoader 클래스의 from_filesystem 메소드에서 path/glob/show_progress/parser 인자를 사용해 GenericLoader 객체를 만드는 방법을 보여준다. ▶ 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
from langchain_community.document_loaders.generic import GenericLoader

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

genericLoader = GenericLoader.from_filesystem(path = ".", glob = "*.txt", show_progress = True, parser = CustomBlobParser())

generator = genericLoader.lazy_load()

for index, document in enumerate(generator):
    print(document)

from langchain_core.document_loaders import BaseBlobParser

from langchain_core.document_loaders import Blob

from typing import Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.generic import GenericLoader

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}

)

genericLoader = GenericLoader.from_filesystem(path = ".", glob = "*.txt", show_progress = True, parser = CustomBlobParser())

generator = genericLoader.lazy_load()

for index, document in enumerate(generator):

print(document)

▶ 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
colorama==0.4.6
dataclasses-json==0.6.7
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.27.2
httpx-sse==0.4.0
idna==3.10
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.8
langchain-community==0.3.8
langchain-core==0.3.21
langchain-text-splitters==0.3.2
langsmith==0.1.145
marshmallow==3.23.1
multidict==6.1.0
mypy-extensions==1.0.0
numpy==1.26.4
orjson==3.10.12
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic-settings==2.6.1
pydantic_core==2.27.1
python-dotenv==1.0.1
PyYAML==6.0.2
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.35
tenacity==9.0.0
tqdm==4.67.0
typing-inspect==0.9.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

colorama==0.4.6

dataclasses-json==0.6.7

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.27.2

httpx-sse==0.4.0

idna==3.10

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.8

langchain-community==0.3.8

langchain-core==0.3.21

langchain-text-splitters==0.3.2

langsmith==0.1.145

marshmallow==3.23.1

multidict==6.1.0

mypy-extensions==1.0.0

numpy==1.26.4

orjson==3.10.12

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic-settings==2.6.1

pydantic_core==2.27.1

python-dotenv==1.0.1

PyYAML==6.0.2

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.35

tenacity==9.0.0

tqdm==4.67.0

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.18.0

※ pip install

[PYTHON/LANGCHAIN] FileSystemBlobLoader 클래스 : 생성자에서 path/glob/show_progress 인자를 사용해 FileSystemBlobLoader 객체 만들기

■ FileSystemBlobLoader 클래스의 생성자에서 path/glob/show_progress 인자를 사용해 FileSystemBlobLoader 객체를 만드는 방법을 보여준다. ▶ 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
from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader

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

fileSystemBlobLoader = FileSystemBlobLoader(path = ".", glob = "*.txt", show_progress = True)
customBlobParser     = CustomBlobParser()

for blob in fileSystemBlobLoader.yield_blobs():
    for document in customBlobParser.lazy_parse(blob):
        print(document)
        break

from langchain_core.document_loaders import BaseBlobParser

from langchain_core.document_loaders import Blob

from typing import Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader

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}

)

fileSystemBlobLoader = FileSystemBlobLoader(path = ".", glob = "*.txt", show_progress = True)

customBlobParser = CustomBlobParser()

for blob in fileSystemBlobLoader.yield_blobs():

for document in customBlobParser.lazy_parse(blob):

print(document)

break

▶ 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
colorama==0.4.6
dataclasses-json==0.6.7
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.27.2
httpx-sse==0.4.0
idna==3.10
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.8
langchain-community==0.3.8
langchain-core==0.3.21
langchain-text-splitters==0.3.2
langsmith==0.1.145
marshmallow==3.23.1
multidict==6.1.0
mypy-extensions==1.0.0
numpy==1.26.4
orjson==3.10.12
packaging==24.2
propcache==0.2.0
pydantic==2.10.1
pydantic-settings==2.6.1
pydantic_core==2.27.1
python-dotenv==1.0.1
PyYAML==6.0.2
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
SQLAlchemy==2.0.35
tenacity==9.0.0
tqdm==4.67.0
typing-inspect==0.9.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

colorama==0.4.6

dataclasses-json==0.6.7

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.27.2

httpx-sse==0.4.0

idna==3.10

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.8

langchain-community==0.3.8

langchain-core==0.3.21

langchain-text-splitters==0.3.2

langsmith==0.1.145

marshmallow==3.23.1

multidict==6.1.0

mypy-extensions==1.0.0

numpy==1.26.4

orjson==3.10.12

packaging==24.2

propcache==0.2.0

pydantic==2.10.1

pydantic-settings==2.6.1

pydantic_core==2.27.1

python-dotenv==1.0.1

PyYAML==6.0.2

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

SQLAlchemy==2.0.35

tenacity==9.0.0

tqdm==4.67.0

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.2.3

yarl==1.18.0

※ pip install langchain

[PYTHON/LANGCHAIN] Blob 클래스 : as_string 메소드를 사용해 문자열 구하기

■ Blob 클래스의 as_string 메소드를 사용해 문자열을 구하는 방법을 보여준다. ▶ main.py


from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

string = blob.as_string()

print(string)

"""
meow meow1
meow meow2
meow meow3
"""

from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

string = blob.as_string()

print(string)

"""

meow meow1

meow meow2

meow meow3

"""

▶ 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] Blob 클래스 : as_bytes 메소드를 사용해 bytes 객체 구하기

■ Blob 클래스의 as_bytes 메소드를 사용해 bytes 객체를 구하는 방법을 보여준다. ▶ main.py


from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

bytes1 = blob.as_bytes()

print(bytes1)

"""
b'meow meow1\r\nmeow meow2\r\nmeow meow3'
"""

from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

bytes1 = blob.as_bytes()

print(bytes1)

"""

b'meow meow1\r\nmeow meow2\r\nmeow meow3'

"""

▶ 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] Blob 클래스 : encoding/source/metadata 속성 사용하기

■ Blob 클래스의 encoding/source/metadata 속성을 사용하는 방법을 보여준다. ▶ main.py


from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

print(blob.encoding)
print(blob.source)
print(blob.metadata)

from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

print(blob.encoding)

print(blob.source)

print(blob.metadata)

▶ 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 명령을 실행했다. meow.txt

[PYTHON/LANGCHAIN] Blob 클래스 : from_path 정적 메소드를 사용해 파일 로드하기

■ Blob 클래스의 from_path 정적 메소드를 사용해 파일을 로드하는 방법을 보여준다. ▶ main.py


from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

from langchain_core.document_loaders import Blob

blob = Blob.from_path("meow.txt", metadata = {"foo" : "bar"})

▶ 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] 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 명령을 실행했다.