AI & Generative AI

Unpacking LLMs with LangChain

Data Mastery Series — Episode 32: LangChain Website (Part 7)

26 Oct 202410 min readLangChainDashboard

Unpacking LLMs with LangChain

Data Mastery Series — Episode 32: LangChain Website (Part 7)

Connect with me and follow our journey: Linkedin, Facebook

Welcome to Episode 32 of the Data Mastery Series, If you haven’t caught up on our previous episodes, please check them out for a stronger foundation:

Part 1: LangChain Model I/O Basics
Part 2: Unpacking Prompt Templates with LangChain
Part 3: Exploring Few-Shot Prompts with LangChain
Part 4: Unpacking Chat Models with LangChain
Part 5: Exploring Chat Models with LangChain
Part 6: Continue Exploring Chat Models with LangChain

Note: As we dive into LangChain, I’ll be sharing insights and key notes from my own study of the LangChain documentation. Let’s jump in and explore some fascinating features in today’s episode!

Source: https://python.langchain.com/v0.1/docs/modules/model%5Fio/llms/

In this episode, we’ll dive into creating custom language models (LLMs) with LangChain. We’ll also explore key techniques like caching and streaming, which can help streamline and optimize your AI workflows. Let’s get started!

Quick Start in LLMs

LangChain’s Language Models (LLMs) are built on a flexible interface called “Runnable,” allowing for smooth integration and versatile functionality. This interface supports various tasks like:

invoke: Runs a single request.
stream: Delivers responses piece by piece
batch: Handles multiple requests at once.

For a deeper dive into these functions, check out Episode 29. Here’s a quick example of using invoke to run a query:

Example - invoke

llm.invoke("What are some theories about the relationship between unemployment and inflation?")

Output:

'''
\n\n1. Phillips Curve Theory: This theory, proposed by economist A.W. Phillips, suggests an inverse relationship between unemployment and inflation. It states that when unemployment is low, inflation tends to be high and vice versa.\n\n2. Demand-Pull Theory: According to this theory, inflation is caused by excess demand in the economy, which leads to higher prices. This excess demand can be a result of low unemployment, as more people have jobs and are able to spend more.\n\n3. Cost-Push Theory: This theory suggests that inflation is caused by an increase in production costs, such as wages and raw materials. When unemployment is low, there is a higher demand for labor, which leads to higher wages. As a result, businesses may increase prices to cover their higher costs.\n\n4. Rational Expectations Theory: This theory states that people's expectations about future inflation can influence current inflation. For example, if people expect prices to rise in the future, they may demand higher ...
'''

Custom LLM

Creating a custom LLM in LangChain is straightforward. By wrapping your model with LangChain’s interface, you can add custom features with minimal setup. To create a custom LLM, you only need to define:

_call: The main function that processes input text and returns a response.
_llm_type: Labels the model (e.g., “custom”) for tracking and identification.

There are also some optional functions you can implement for added functionality, such as

_identifying_params: Provides additional model details (like name and version) for easy identification.
_acall: Allows handling multiple requests simultaneously.ns.
_stream: Sends responses gradually, ideal for real-time interactions.
_astream: Async version of _stream for handling multiple users.

Here’s a simple example of a custom LLM that returns the first n characters of any input:

Example - Custom LLM

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

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

Example:  

    .. code-block:: python  

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

n: int # จำนวนอักขระจากข้อความ prompt ที่จะคืนกลับ  
"""The number of characters from the last message of the prompt to be echoed."""  

# ฟังก์ชัน _call เป็นการทำงานหลักของโมเดล ซึ่งใช้ในการประมวลผลข้อมูลที่ส่งเข้ามา (prompt) และคืนค่าผลลัพธ์ตามที่กำหนด  
def _call(   
    self,  
    prompt: str,  
    stop: Optional[List[str]] = None,  
    run_manager: Optional[CallbackManagerForLLMRun] = None,  
    **kwargs: Any,  
) -> str:  
    """Run the LLM on the given input.  

    Override this method to implement the LLM logic.  

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

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

# ฟังก์ชัน _stream ฟังก์ชันที่ทำงานแบบสตรีม ซึ่งหมายถึงการคืนค่าข้อความทีละตัวอักษร (streaming output)  
def _stream(  
    self,  
    prompt: str,  
    stop: Optional[List[str]] = None,  
    run_manager: Optional[CallbackManagerForLLMRun] = None,  
    **kwargs: Any,  
) -> Iterator[GenerationChunk]:  
    """Stream the LLM on the given prompt.  

    This method should be overridden by subclasses that support streaming.  

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

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

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

        yield chunk  

# ฟังก์ชัน _identifying_params คืนค่าข้อมูลเกี่ยวกับโมเดลในรูปแบบของ dictionary ซึ่งในกรณีนี้เป็น model_name: "CustomChatModel"  
@property  
def _identifying_params(self) -> Dict[str, Any]:  
    """Return a dictionary of identifying parameters."""  
    return {  
        # The model name allows users to specify custom token counting  
        # rules in LLM monitoring applications (e.g., in LangSmith users  
        # can provide per token pricing for their model and monitor  
        # costs for the given LLM.)  
        "model_name": "CustomChatModel",  
    }  

# ฟังก์ชัน _llm_type คืนค่าประเภทของโมเดลเป็น custom ใช้สำหรับระบุประเภทของโมเดลเพื่อการบันทึกข้อมูลการใช้งาน  
@property  
def _llm_type(self) -> str:  
    """Get the type of language model used by this chat model. Used for logging purposes only."""  
    return "custom"

Example - Let's test it

llm = CustomLLM(n=7)
print(llm)

Output:

'''
CustomLLM
Params: {'model_name': 'CustomChatModel'}
'''

##############################################

print(llm.invoke("This is a foobar thing"))

Output: จะคืนค่า 7 ตัวอักษรแรก

'''
'This is'
'''

##############################################

llm.batch(["woof woof woof", "meow meow meow"])

Output: จะคืนค่า 7 ตัวอักษรแรก

'''
['woof wo', 'meow me']
'''

Caching

<same as Caching topic in Episode 31>

LangChain’s caching feature helps reduce API costs and speeds up response times by storing answers to frequent requests. With caching, the first call is processed normally, but repeated calls become almost instant.

Example - Caching (First call takes longer as it's not cached)

llm = OpenAI(model_name="gpt-3.5-turbo-instruct", n=2, best_of=2, api_key=OPENAI_API_KEY)
set_llm_cache(InMemoryCache())

Start time

start_time = time.time()
llm.predict("Tell me a joke")

End time

end_time = time.time()
print("Processing Time:", end_time - start_time, "seconds")

Output: # 0.681973934173584 seconds

'''
Why couldn't the bicycle stand up by itself?
Because it was two-tired.
'''

##############################################

(Repeated call)

Output: # 0.001056671142578125 seconds

'''
Why couldn't the bicycle stand up by itself?
Because it was two-tired.
'''

Streaming

<same as Streaming topic in Episode 29>

Streaming lets you receive responses gradually, token by token, rather than waiting for the entire output. This is particularly useful for long outputs, enhancing responsiveness and interactivity.

Example - Streaming

llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512, api_key=OPENAI_API_KEY)

for chunk in llm.stream("Write me a song about sparkling water."):
print(chunk, end="", flush=True)

Output:

'''
Verse 1:
Bubbles dancing in my glass
Clear and crisp, it's such a blast
Refreshing taste, it's like a dream
Sparkling water, you make me beam

Chorus:
Oh sparkling water, you're my delight
With every sip, you make me feel so right
You're like a party in my mouth
I can't get enough, I'm hooked no doubt

Verse 2:
No sugar, no calories, just pure bliss
You're the perfect drink, I must confess
From lemon to lime, so many flavors to choose
Sparkling water, you never fail to amuse

Chorus:
Oh sparkling water, you're my delight
With every sip, you make me feel so right
You're like a party in my mouth
I can't get enough, I'm hooked no doubt

Bridge:
Some may say you're just plain water
But to me, you're so much more
You bring a sparkle to my day
In every single way

Chorus:
Oh sparkling water, you're my delight
With every sip, you make me feel so right
You're like a party in my mouth
I can't get enough, I'm hooked no doubt

Outro:
So here's to you, my dear sparkling water
You'll always be my go-to drink forever
With every sip, I feel so alive
Sparkling water, you're my love, my vibe.
'''

Tracking token usage

Monitoring token usage helps manage costs and optimize performance. LangChain offers token tracking with the OpenAI API, giving insights into usage and costs for each request.

Example - Tracking token usage

llm = OpenAI(model_name="gpt-3.5-turbo-instruct", n=2, best_of=2, api_key=OPENAI_API_KEY)

with get_openai_callback() as cb:
result = llm.invoke("Tell me a joke about food")
print(result)
print(cb)

Output:

'''
Why did the tomato turn red?
Because it saw the salad dressing!
Tokens Used: 0
Prompt Tokens: 0
Completion Tokens: 0
Successful Requests: 0
Total Cost (USD): $0.0
'''

In this episode, we explored building custom LLMs, caching for efficiency, streaming for interactive responses, and tracking token usage. These techniques help create more flexible, efficient, and cost-effective AI solutions with LangChain. Stay tuned for our next episode as we continue exploring powerful tools and techniques in the Data Mastery Series!

Data Science Explore the world of data science with Donato_Story

Dashboard Discover the power of data visualization with Donato_Story

Donato_Journey Join me on my journey (Thai version)

Course_Review Discover the training courses with Donato_Story (Thai version)

I really value your thoughts and feedback. Please share any comments or questions you have below, or reach out to me on:

Medium: medium.com/donato-story
Facebook: web.facebook.com/DonatoStory
Linkedin: linkedin.com/in/nattapong-thanngam

Originally published on Medium

AI & Generative AI