9️⃣Expression Language(LCEL)

LCEL: LangChain Expression Language

LangChain Expression Language, LCEL은 체인을 쉽게 구성할 수 있는 선언적 방식입니다. LCEL은 가장 간단한 "프롬프트 + LLM" Chain부터 가장 복잡한 체인까지 코드 변경 없이 프로토타입을 프로덕션에 적용할 수 있도록 설계되었습니다.

LCEL을 사용하면 기본 구성 요소로 복잡한 체인을 쉽게 구축할 수 있으며 스트리밍, 병렬 처리, 로깅과 같은 기본 기능을 지원합니다.

LCEL을 사용하여 기본적인 Chain 구성을 실행해보고, ChatwithHistory를 실행해 보겠습니다.

Setup Environments

%pip install langchain-openai

import os
from dotenv import load_dotenv  

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

Basic ChatPromptTemplate

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template(
    "{topic}에 대한 흥미로운 사실을 알려줄래?"
)

prompt_val = prompt.invoke(
    {"topic": "dog"}
)
print(prompt_val)

messages=[HumanMessage(content='dog에 대한 흥미로운 사실을 알려줄래?')]

print(prompt_val.to_messages())

[HumanMessage(content='dog에 대한 흥미로운 사실을 알려줄래?')]

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-3.5-turbo"
)
result = model.invoke(prompt_val)
result

AIMessage(content='개는 인간이 길들여진 가장 오래된 동물 중 하나입니다. 약 1만 3천년 전부터 사람들과 함께 살아왔으며, 진화의 과정에서 인간과 함께 살아가며 조화롭게 발전해왔습니다. 이는 개가 인간의 가장 가까운 친구 중 하나로 여겨지는 이유 중 하나입니다.', response_metadata={'token_usage': {'completion_tokens': 125, 'prompt_tokens': 29, 'total_tokens': 154}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-4aebcc59-1495-41b9-b460-13f9d15f56bc-0')

from langchain_core.output_parsers import StrOutputParser

output_parser = StrOutputParser()
output_parser.invoke(result)

'개는 인간이 길들여진 가장 오래된 동물 중 하나입니다. 약 1만 3천년 전부터 사람들과 함께 살아왔으며, 진화의 과정에서 인간과 함께 살아가며 조화롭게 발전해왔습니다. 이는 개가 인간의 가장 가까운 친구 중 하나로 여겨지는 이유 중 하나입니다.'

LCEL(LangChain Expression Language) Chain

prompt = ChatPromptTemplate.from_template(
    "{topic}에 대한 흥미로운 사실을 알려줄래?"
)
model = ChatOpenAI()
output_parser = StrOutputParser()

# lcel로 multi-chain을 구성
basicchain = model | output_parser

basicchain.invoke("안녕!")

'안녕하세요! 무엇을 도와드릴까요?'

# lcel로 prompt-model-output_parser 구성
chain = prompt | model | output_parser

chain.invoke({"topic": "dog"})

'개는 인간의 가장 오래된 친구 중 하나로 알려져 있습니다. 실제로, 최근 연구에 따르면 15,000년에서 30,000년 전에 이미 인간과 함께 살았던 것으로 밝혀졌습니다. 이는 개가 인류의 가장 오래된 반렬족 중 하나로 여겨지는 이유 중 하나입니다.'

RAG with LCEL

from langchain.schema import Document
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import RunnablePassthrough

# openAI Embedding 사용
embedding_function = OpenAIEmbeddings()

# prompt 구성
docs = [
    Document(
        page_content="개는 피자 먹는 것을 좋아한다.", metadata={"source": "animal.txt"}
    ),
    Document(
        page_content="고양이는 생선 먹는 것을 좋아한다.", metadata={"source": "animal.txt"}
    ),
]

# Chroma VectorStore 설정
db = Chroma.from_documents(
    docs, embedding_function
)
retriever = db.as_retriever()

retriever.get_relevant_documents(
    "개는 어떤 음식을 먹기를 원해?"
)

[Document(page_content='개는 피자 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}),
 Document(page_content='고양이는 생선 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}),
 Document(page_content='the dog loves to eat pizza', metadata={'source': 'animal.txt'}),
 Document(page_content='the cat loves to eat lasagna', metadata={'source': 'animal.txt'})]

retriever.invoke(
    "개는 어떤 음식을 먹기를 원해?"
)

[Document(page_content='개는 피자 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}),
 Document(page_content='고양이는 생선 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}),
 Document(page_content='the dog loves to eat pizza', metadata={'source': 'animal.txt'}),
 Document(page_content='the cat loves to eat lasagna', metadata={'source': 'animal.txt'})]

template = """Answer the question based only on the following context:
{context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()

from operator import itemgetter

# Retrieval Chain으로 'context', 'question'을 구성한다.
retrieval_chain = (
    {
        "context": (lambda x: x["question"]) | retriever,
        # "question": lambda x: x["question"],
        "question": itemgetter("question"),
    }
    | prompt
    | model
    | StrOutputParser()
)

retrieval_chain.invoke(
    {"question": "개는 어떤 음식을 먹기를 원해?"}
)

'피자'

RunnablePassthrough()로 사용자 정의 함수를 대신 표현하자.

template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

retrieval_chain.invoke(
    "개는 어떤 음식을 먹기를 원해?"
)

'피자(replace pizza)'

Runnable Function

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

runnable = RunnableParallel(
    passed=RunnablePassthrough(),
    extra=RunnablePassthrough.assign(upper=lambda x: x["input"].upper()),
    modified=lambda x: x["input"] * 3,
)

runnable.invoke({"input": "hello!"})

{'passed': {'input': 'hello!'},
 'extra': {'input': 'hello!', 'upper': 'HELLO!'},
 'modified': 'hello!hello!hello!'}

---

LCEL: ChatWithHistory

import os
from dotenv import load_dotenv  

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

from langchain.schema import Document
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
import os
from dotenv import load_dotenv

# openAI Embedding 사용
embedding_function = OpenAIEmbeddings()

# prompt 구성
docs = [
    Document(
        page_content="개는 피자 먹는 것을 좋아한다.", metadata={"source": "animal.txt"}
    ),
    Document(
        page_content="고양이는 생선 먹는 것을 좋아한다.", metadata={"source": "animal.txt"}
    ),
]

# Chroma VectorStore 설정
db = Chroma.from_documents(
    docs, embedding_function
)
retriever = db.as_retriever()

retriever.invoke("정확히 무슨 뜻이야?")

Number of requested results 4 is greater than number of elements in index 2, updating n_results = 2
[Document(page_content='개는 피자 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}),
 Document(page_content='고양이는 생선 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'})]

from langchain.prompts.prompt import PromptTemplate

rephrase_template = """다음 대화와 후속 질문이 주어졌을 때 후속 질문을 원래 언어로 독립된 질문으로 바꾸어 줄래.

Chat History: {chat_history}
Follow Up Input: {question}
Standalone question:"""

REPHRASE_TEMPLATE = PromptTemplate.from_template(rephrase_template)

from langchain_core.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

rephrase_chain = REPHRASE_TEMPLATE | ChatOpenAI(temperature=0) | StrOutputParser()

rephrase_chain.invoke(
    {
        "question": "아니, 정말로?",
        "chat_history": [
            HumanMessage(content="개는 어떤 음식을 먹기를 원해?"),
            AIMessage(content="참치!"),
        ],
    }
)

'개는 어떤 음식을 먹기를 원해?'

from langchain_core.prompts import ChatPromptTemplate

template = """Answer the question based only on the following context:
{context}

Question: {question}
"""

ANSWER_PROMPT = ChatPromptTemplate.from_template(template)

from langchain_core.runnables import RunnablePassthrough

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | ANSWER_PROMPT
    | ChatOpenAI(temperature=0)
    | StrOutputParser()
)

final_chain = rephrase_chain | retrieval_chain

final_chain.invoke(
    {
        "question": "아니, 정말로?",
        "chat_history": [
            HumanMessage(content="개는 어떤 음식을 먹기를 원해?"),
            AIMessage(content="참치"),
        ],
    }
)

Number of requested results 4 is greater than number of elements in index 2, updating n_results = 2





'피자'

Chat with returning documents

retrieved_documents = {"docs": retriever, "question": RunnablePassthrough()}

final_inputs = {
    "context": lambda x: "\n".join(doc.page_content for doc in x["docs"]),
    "question": RunnablePassthrough(),
}
answer = {
    "answer": final_inputs | ANSWER_PROMPT | ChatOpenAI() | StrOutputParser(),
    "docs": RunnablePassthrough(),
}

final_chain = rephrase_chain | retrieved_documents | answer

result = final_chain.invoke(
    {
        "question": "아니, 정말로?",
        "chat_history": [
            HumanMessage(content="개는 어떤 음식을 먹기를 원해?"),
            AIMessage(content="참치!"),
        ],
    }
)
print(result)

Number of requested results 4 is greater than number of elements in index 2, updating n_results = 2


{'answer': '피자', 'docs': {'docs': [Document(page_content='개는 피자 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}), Document(page_content='고양이는 생선 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'})], 'question': '개는 어떤 음식을 먹기를 원해?'}}

result["answer"]

'피자'

result["docs"]["docs"]

[Document(page_content='개는 피자 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'}),
 Document(page_content='고양이는 생선 먹는 것을 좋아한다.', metadata={'source': 'animal.txt'})]