How to run a parallel chain with different config for using chat history

[JANGSOONMYUN]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import List

import openai
from openai import OpenAI
from transformers import GPT2Tokenizer
from get_config import get_api_key
from get_character import get_character
from handle_contents import load_contents, load_contents_csv

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.output_parsers import StructuredOutputParser, ResponseSchema
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.prompts import MessagesPlaceholder
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain import hub
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables import RunnableWithMessageHistory, ConfigurableFieldSpec
from langchain_community.chat_models import ChatAnthropic
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.prompts import PromptTemplate
from langchain_core.runnables import (
    RunnableLambda,
    ConfigurableFieldSpec,
    RunnablePassthrough,
)


# Define an in-memory chat message history
class InMemoryHistory(BaseChatMessageHistory):
    def __init__(self):
        self.messages = []

    def add_messages(self, messages: List[BaseMessage]) -> None:
        self.messages.extend(messages)

    def clear(self) -> None:
        self.messages = []

class GPTConfig:
    def __init__(self, character_id="default", stream=False, tokenizer=None, 
                 keep_dialog=None, model='gpt-4o-mini', temperature=0.3, max_tokens_output=4096, 
                 max_tokens_context=30000, api_key_path='./settings/config.json'):
        self.character_id = character_id
        self.stream = stream
        self.tokenizer = tokenizer
        self.keep_dialog = keep_dialog
        self.model = model
        self.temperature = temperature
        self.max_tokens_output = max_tokens_output
        self.max_tokens_context = max_tokens_context
        self.api_key_path = api_key_path
        

class LangChainModule():
    def __init__(self, config: GPTConfig) -> None:
        self.is_running = False
        
        self.config = config
        self.api_info = get_api_key(self.config.api_key_path)
        
        self.store = {}
        
        self.init_llm()
        
    def params(self, config: GPTConfig):
        self.config = config
        
    def init_llm(self):
        self.llm = ChatOpenAI(
            model=self.config.model,
            temperature=self.config.temperature,
            max_tokens=self.config.max_tokens_output,
            timeout=None,
            max_retries=2,
            api_key=self.api_info['api_key'], 
            organization=self.api_info['organization']
            # base_url="...",
            # other params...
        )
    
    # To store history (conversations)
    def get_session_history(self, user_id: str, conversation_id: str) -> BaseChatMessageHistory:
        if (user_id, conversation_id) not in self.store:
            self.store[(user_id, conversation_id)] = InMemoryHistory()
        return self.store[(user_id, conversation_id)]
    # with history
    def process_chain(self, instructions, prompts, input_dict, output_dict=None, with_history=None):
        llm = self.llm
        # keys: input_dict.keys()
        sys_msg = SystemMessagePromptTemplate.from_template(instructions)
        msg_plchldr = MessagesPlaceholder(variable_name="history")
        hum_msg =  HumanMessagePromptTemplate.from_template(
            prompts + \
            "\n{format_instructions}"
        )
        if with_history is None:
            messages = [
                        sys_msg,
                        hum_msg
                    ]
        else:
            messages = [
                        sys_msg,
                        msg_plchldr,
                        hum_msg
                    ]
            
        if output_dict is None:
            prompt_template = ChatPromptTemplate(
                messages=messages,
                input_variables=list(input_dict.keys())
            )
            chain = (
                prompt_template
                | llm
                | StrOutputParser()
            )
        else:
            output_parser = self.set_json_output(output_dict)
            prompt_template = ChatPromptTemplate(
                messages=messages,
                input_variables=list(input_dict.keys()),
                partial_variables={"format_instructions": output_parser.get_format_instructions()}
            )
            chain = (
                prompt_template
                | llm
                | output_parser
            )
        return chain
    def run_single_chain(self, main_chain, input_dict, config_dict=None):
        self.is_running = True
        
        # wrap config with configurable
        if 'configurable' not in config_dict:
            config_dict = {"configurable": config_dict.copy()}
        
        if config_dict is None:
            result = main_chain.invoke(input_dict)
        else:
            result = main_chain.invoke(input_dict, config=config_dict)
        return result
    
    def combine_parallel_chain(self, chains, chain_names=None):
        chain_dict = {}
        for i, chain in enumerate(chains):
            chain_name = 'chain_' + str(i)
            if chain_names is not None:
                chain_name = chain_names[i]
            
            chain_dict[chain_name] = chain
        map_chain = RunnableParallel(chain_dict)
        return map_chain

def _demo_parallel():
    config = GPTConfig(character_id="default", stream=False, tokenizer=None, 
                 keep_dialog=None, model='gpt-4o-mini', temperature=0.8, max_tokens_output=None, 
                 max_tokens_context=30000, api_key_path='./settings/config.json')
    lc_module = LangChainModule(config)
    
    # process_chain(self, llm, instructions, prompts, input_dict, output_dict=None, with_history=None):
    
    # chain 1
    input_dict = {"korean_word1": "양념", "korean_word2": "진득하다"}
    config={"configurable": {"user_id": "123", "conversation_id": "1"}}
    instructions = "output should be json format and the keys are english_word1, english_word2"
    prompts = "explain {korean_word1} and {korean_word2} using oxford dictionary to me in English."
    
    output_dict = {"english_word1": "explain the first word", "english_word2": "explain the second word"}
    
    main_chain = lc_module.process_chain(instructions=instructions, prompts=prompts,
                       input_dict=input_dict, output_dict=output_dict)
    
    # chain 2
    input_dict2 = {"korean_word3": "뭉실뭉실", "korean_word4": "개미허리"}
    config2={"configurable": {"user_id": "33", "conversation_id": "1"}}
    instructions2 = "output should be json format and the keys are english_word1, english_word2"
    prompts2 = "explain {korean_word3} and {korean_word4} using oxford dictionary to me in English."
    output_dict2 = {"english_word1": "explain the first word", "english_word2": "explain the second word"}
    
    chain2 = lc_module.process_chain(instructions=instructions2, prompts=prompts2,
                       input_dict=input_dict2, output_dict=output_dict2)
        
    map_chain = lc_module.combine_parallel_chain(chains=[main_chain, chain2])
    map_input_dict = {}
    map_input_dict.update(input_dict)
    map_input_dict.update(input_dict2)
    
    # run_single_chain(self, main_chain, input_dict, config_dict=None):
    result = lc_module.run_single_chain(main_chain=map_chain, input_dict=map_input_dict, config_dict=config)
    print(result)
    pass

# test
if __name__ == "__main__":
    _demo_parallel()

Description

in _demo_parallel function, I want to set config for two chains but individually.
So, [main_chain] should use [config] while [chain2] uses [config2]

How to do it? Please let me know.
Or please tell me better solution to run parallel chains with different parameters.

System Info

langchain==0.2.9
langchain-community==0.2.9
langchain-core==0.2.22
langchain-openai==0.1.10
langchain-text-splitters==0.2.2
langchainhub==0.1.21

linux(ubuntu)
python 3.9

[dosubot[bot]]

To run parallel chains with different configurations for each chain in the _demo_parallel function, you can use the RunnableParallel class to define each chain with its specific configuration. Here's how you can modify your _demo_parallel function to achieve this:

1. Define each chain with its specific configuration.
2. Combine the chains using RunnableParallel.
3. Invoke the parallel chains with the input and their respective configurations.

Here's the updated code:

def _demo_parallel():
    config = GPTConfig(character_id="default", stream=False, tokenizer=None, 
                 keep_dialog=None, model='gpt-4o-mini', temperature=0.8, max_tokens_output=None, 
                 max_tokens_context=30000, api_key_path='./settings/config.json')
    lc_module = LangChainModule(config)
    
    # process_chain(self, llm, instructions, prompts, input_dict, output_dict=None, with_history=None):
    
    # chain 1
    input_dict = {"korean_word1": "양념", "korean_word2": "진득하다"}
    config1 = {"configurable": {"user_id": "123", "conversation_id": "1"}}
    instructions = "output should be json format and the keys are english_word1, english_word2"
    prompts = "explain {korean_word1} and {korean_word2} using oxford dictionary to me in English."
    
    output_dict = {"english_word1": "explain the first word", "english_word2": "explain the second word"}
    
    main_chain = lc_module.process_chain(instructions=instructions, prompts=prompts,
                       input_dict=input_dict, output_dict=output_dict)
    
    # chain 2
    input_dict2 = {"korean_word3": "뭉실뭉실", "korean_word4": "개미허리"}
    config2 = {"configurable": {"user_id": "33", "conversation_id": "1"}}
    instructions2 = "output should be json format and the keys are english_word1, english_word2"
    prompts2 = "explain {korean_word3} and {korean_word4} using oxford dictionary to me in English."
    output_dict2 = {"english_word1": "explain the first word", "english_word2": "explain the second word"}
    
    chain2 = lc_module.process_chain(instructions=instructions2, prompts=prompts2,
                       input_dict=input_dict2, output_dict=output_dict2)
        
    # Combine both chains to run in parallel
    map_chain = RunnableParallel(chain1=main_chain, chain2=chain2)

    # Invoke the parallel chains with the input and their respective configurations
    result = map_chain.invoke({
        "chain1": {"input": input_dict, "config": config1},
        "chain2": {"input": input_dict2, "config": config2}
    })

    print(result)

# test
if __name__ == "__main__":
    _demo_parallel()

In this example:

• main_chain and chain2 are defined with their specific configurations (config1 and config2).
• RunnableParallel is used to combine both chains.
• The invoke method is used to execute both chains with their respective inputs and configurations.

This approach ensures that each chain runs with its own configuration in parallel ^[1].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

To resolve the ValueError: Missing keys ['conversation_id', 'user_id'] in config['configurable'], you need to ensure that the config dictionary passed to the invoke method includes both conversation_id and user_id keys within the configurable section for each chain.

Here's the updated code to ensure that each chain receives its specific configuration:

def _demo_parallel():
    config = GPTConfig(character_id="default", stream=False, tokenizer=None, 
                 keep_dialog=None, model='gpt-4o-mini', temperature=0.8, max_tokens_output=None, 
                 max_tokens_context=30000, api_key_path='./settings/config.json')
    lc_module = LangChainModule(config)
    
    # chain 1
    input_dict1 = {"korean_word1": "양념", "korean_word2": "진득하다"}
    config1 = {"configurable": {"user_id": "123", "conversation_id": "1"}}
    instructions1 = "output should be json format and the keys are english_word1, english_word2"
    prompts1 = "explain {korean_word1} and {korean_word2} using oxford dictionary to me in English."
    output_dict1 = {"english_word1": "explain the first word", "english_word2": "explain the second word"}
    
    main_chain = lc_module.process_chain(instructions=instructions1, prompts=prompts1,
                       input_dict=input_dict1, output_dict=output_dict1)
    
    # chain 2
    input_dict2 = {"korean_word3": "뭉실뭉실", "korean_word4": "개미허리"}
    config2 = {"configurable": {"user_id": "33", "conversation_id": "1"}}
    instructions2 = "output should be json format and the keys are english_word1, english_word2"
    prompts2 = "explain {korean_word3} and {korean_word4} using oxford dictionary to me in English."
    output_dict2 = {"english_word1": "explain the first word", "english_word2": "explain the second word"}
    
    chain2 = lc_module.process_chain(instructions=instructions2, prompts=prompts2,
                       input_dict=input_dict2, output_dict=output_dict2)
        
    # Combine both chains to run in parallel
    map_chain = RunnableParallel({
        "chain1": main_chain,
        "chain2": chain2
    })

    # Provide the required input variables and configurations for each chain
    input_data = {
        "chain1": {"input": input_dict1, "config": config1},
        "chain2": {"input": input_dict2, "config": config2}
    }

    # Run the parallel chains with the input data and configurations
    result = map_chain.invoke(input_data)
    print(result)

# test
if __name__ == "__main__":
    _demo_parallel()

In this example:

• main_chain and chain2 are defined with their specific configurations (config1 and config2), ensuring that both conversation_id and user_id are included in the configurable section.
• RunnableParallel is used to combine both chains.
• The invoke method is used to execute both chains with their respective inputs and configurations.

This setup ensures that each chain runs with its own configuration in parallel, avoiding the ValueError ^[1][2].

To continue talking to Dosu, mention @dosu.

[JANGSOONMYUN]

@dosu
All of your answers are incorrect. Please never mind.
And I have a new question. Do you have any other idea to run invoke with RunnableParallel with diferent chat_history?

[dosubot[bot]]

To run invoke with RunnableParallel with different chat_history, you can modify the input dictionary to include different chat_history values for each parallel process. Here is an example:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI()

# Define the chains with different prompts
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model

# Create a RunnableParallel with different chat_history
map_chain = RunnableParallel(
    joke=joke_chain,
    poem=poem_chain
)

# Invoke with different chat_history
result = map_chain.invoke({
    "joke": {"topic": "bear", "chat_history": "previous joke history"},
    "poem": {"topic": "bear", "chat_history": "previous poem history"}
})

print(result)

In this example, the invoke method is called with a dictionary that includes different chat_history values for the joke and poem chains. Additionally, you can use configurable fields and alternatives to further customize the behavior of your chains, as demonstrated in the LangChain unit tests. For instance, you can configure different prompt templates and responses for each chain, ensuring that each parallel process operates with its own unique context and configuration.

Moreover, you can leverage RunnableWithMessageHistory to manage session-specific histories. Here is an example of how you can use it:

from langchain_core.runnables import RunnableLambda, RunnableWithMessageHistory
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.history import InMemoryChatMessageHistory
from langchain_core.config import RunnableConfig

# Define a simple runnable that echoes the messages
runnable = RunnableLambda(
    lambda messages: "you said: " + "\n".join(str(m.content) for m in messages if isinstance(m, HumanMessage))
)

# Function to get session history
def _get_get_session_history(store=None):
    store = store or {}
    def get_session_history(session_id):
        return store.setdefault(session_id, InMemoryChatMessageHistory())
    return get_session_history

store = {}
get_session_history = _get_get_session_history(store=store)
with_history = RunnableWithMessageHistory(runnable, get_session_history)

# Configuring session-specific histories
config = {"configurable": {"session_id": "1"}}
output = with_history.invoke([HumanMessage(content="hello")], config)
print(output)  # Output: "you said: hello"

output = with_history.invoke([HumanMessage(content="good bye")], config)
print(output)  # Output: "you said: hello\ngood bye"

This approach ensures that each session maintains its own history, which can be particularly useful for managing different chat_history contexts in parallel processes ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[JANGSOONMYUN]

@dosu
So how to combine RunnableWithMessageHistory and RunnableParallel and run invoke?

[dosubot[bot]]

To combine RunnableWithMessageHistory and RunnableParallel and run invoke, you can wrap the RunnableWithMessageHistory instance with RunnableParallel. Here is an example:

from langchain_core.runnables import RunnableLambda, RunnableWithMessageHistory, RunnableParallel
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.history import InMemoryChatMessageHistory

# Define a simple runnable that echoes the messages
runnable = RunnableLambda(
    lambda messages: "you said: " + "\n".join(str(m.content) for m in messages if isinstance(m, HumanMessage))
)

# Function to get session history
def _get_get_session_history(store=None):
    store = store or {}
    def get_session_history(session_id):
        return store.setdefault(session_id, InMemoryChatMessageHistory())
    return get_session_history

store = {}
get_session_history = _get_get_session_history(store=store)

# Wrap the runnable with message history
with_history = RunnableWithMessageHistory(runnable, get_session_history)

# Now wrap the with_history runnable with RunnableParallel
parallel_runnable = RunnableParallel(with_history)

# Define the configuration
config = {"configurable": {"session_id": "1"}}

# Run the invoke method
output = parallel_runnable.invoke([HumanMessage(content="hello")], config)
print(output)  # Expected output: "you said: hello"

This code demonstrates how to combine RunnableWithMessageHistory and RunnableParallel and run the invoke method ^[1].

To continue talking to Dosu, mention @dosu.