from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
load_dotenv()True6 LCEL Deepdive
prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
chain.invoke({"topic": "ice cream"})'Why did the ice cream truck break down? Because it had too many "scoops" on board!'print(prompt.invoke({"topic": "ice cream"}))messages=[HumanMessage(content='tell me a short joke about ice cream')]from langchain_core.messages.human import HumanMessage
messages = [HumanMessage(content='tell me a short joke about ice cream')]
model.invoke(messages)AIMessage(content='Why did the ice cream go to therapy? Because it had too many sprinkles of anxiety!', response_metadata={'token_usage': {'completion_tokens': 19, 'prompt_tokens': 15, 'total_tokens': 34}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-7f790959-6dd6-4ffc-bd43-d4f0608dafe0-0')6.0.1 What is this “|” in Python?
# Bitwise OR
x = 5 # Binary: 0101
y = 3 # Binary: 0011
result = x | y # Bitwise OR
print(result) # Output: 7 (Binary: 0111)7from abc import ABC, abstractmethod
class CRunnable(ABC):
def __init__(self):
self.next = None
@abstractmethod
def process(self, data):
"""
This method must be implemented by subclasses to define
data processing behavior.
"""
pass
def invoke(self, data):
if self.next is not None:
return self.next.invoke(processed_data)
return processed_data
def __or__(self, other):
return CRunnableSequence(self, other)
class CRunnableSequence(CRunnable):
def __init__(self, first, second):
super().__init__()
self.first = first
self.second = second
def process(self, data):
return data
def invoke(self, data):
first_result = self.first.invoke(data)
return self.second.invoke(first_result)
class AddTen(CRunnable):
def process(self, data):
print("AddTen: ", data)
return data + 10
class MultiplyByTwo(CRunnable):
def process(self, data):
print("Multiply by 2: ", data)
return data * 2
class ConvertToString(CRunnable):
def process(self, data):
print("Convert to string: ", data)
return f"Result: {data}"a = AddTen()
b = MultiplyByTwo()
c = ConvertToString()
chain = a | b | cresult = chain.invoke(10)
print(result)--------------------------------------------------------------------------- NameError Traceback (most recent call last) Cell In[20], line 1 ----> 1 result = chain.invoke(10) 2 print(result) Cell In[17], line 34, in CRunnableSequence.invoke(self, data) 33 def invoke(self, data): ---> 34 first_result = self.first.invoke(data) 35 return self.second.invoke(first_result) Cell In[17], line 34, in CRunnableSequence.invoke(self, data) 33 def invoke(self, data): ---> 34 first_result = self.first.invoke(data) 35 return self.second.invoke(first_result) Cell In[17], line 19, in CRunnable.invoke(self, data) 17 if self.next is not None: 18 return self.next.invoke(processed_data) ---> 19 return processed_data NameError: name 'processed_data' is not defined
6.0.2 Runnables from LangChain
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallelchain = RunnablePassthrough() | RunnablePassthrough () | RunnablePassthrough ()
chain.invoke("hello")'hello'def input_to_upper(input: str):
output = input.upper()
return outputchain = RunnablePassthrough() | RunnableLambda(input_to_upper) | RunnablePassthrough()
chain.invoke("hello") 'HELLO'chain = RunnableParallel({"x": RunnablePassthrough(), "y": RunnablePassthrough()})chain.invoke("hello"){'x': 'hello', 'y': 'hello'}chain.invoke({"input": "hello", "input2": "goodbye"}){'x': {'input': 'hello', 'input2': 'goodbye'},
'y': {'input': 'hello', 'input2': 'goodbye'}}chain = RunnableParallel({"x": RunnablePassthrough(), "y": lambda z: z["input2"]})chain.invoke({"input": "hello", "input2": "goodbye"}){'x': {'input': 'hello', 'input2': 'goodbye'}, 'y': 'goodbye'}6.0.3 Nested chains - now it gets more complicated!
def find_keys_to_uppercase(input: dict):
output = input.get("input", "not found").upper()
return outputchain = RunnableParallel({"x": RunnablePassthrough() | RunnableLambda(find_keys_to_uppercase), "y": lambda z: z["input2"]})chain.invoke({"input": "hello", "input2": "goodbye"}){'x': 'HELLO', 'y': 'goodbye'}chain = RunnableParallel({"x": RunnablePassthrough()})
def assign_func(input):
return 100
def multiply(input):
return input * 10chain.invoke({"input": "hello", "input2": "goodbye"}){'x': {'input': 'hello', 'input2': 'goodbye'}}chain = RunnableParallel({"x": RunnablePassthrough()}).assign(extra=RunnableLambda(assign_func))result = chain.invoke({"input": "hello", "input2": "goodbye"})
print(result){'x': {'input': 'hello', 'input2': 'goodbye'}, 'extra': 100}6.0.4 Combine multiple chains (incl. coercion)
def extractor(input: dict):
return input.get("extra", "Key not found")
def cupper(upper: str):
return str(upper).upper()
new_chain = RunnableLambda(extractor) | RunnableLambda(cupper)new_chain.invoke({"extra": "test"})'TEST'final_chain = chain | new_chain
final_chain.invoke({"input": "hello", "input2": "goodbye"})'100'6.0.5 Real Work example
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
vectorstore = FAISS.from_texts(
["Cats love thuna"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template=template)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
RunnableParallel({"context": retriever | format_docs, "question": RunnablePassthrough()})
| prompt
| ChatOpenAI()
| StrOutputParser()
)retriever.invoke("Eat")[Document(page_content='Cats love thuna')]rag_chain.invoke("What do cats like to eat?")'Tuna'