I am using LangChain and have a class with many methods. I intend to use parallel chains to process lots of data.
Here is one of my steps, which happens to use other methods in the class in which it is contained:
def management_plan(self, input) -> str: # uses management_plan_model
print("PROCESSING write_notes coroutine: starting management_plan()")
for notes_template_item in self.notes_template_item_list:
if notes_template_item in self.similar_prompt_items['MANAGEMENT PLAN']:
prompt = self.personality[notes_template_item].replace("{input_text}", input.messages[0].content)
context = self.modify_context(self.personality['CONTEXT'], RAG_data=self.rag_bot.output+self.rag_bot.issues_list+self.rag_bot.differentials)
prompt_template = ChatPromptTemplate.from_messages([("system","{context}"), ("human","{prompt}")])
return prompt_template.format_prompt(context=context, prompt=prompt)
This is one of my branches:
demographics_review_chain = (RunnableLambda(lambda x: demographics_review(self, input=x)) | self.general_model | StrOutputParser())
This is a part of the chain:
chain = (
input_template
| RunnableParallel(branches={"demographics_review":demographics_review_chain, "management_plan":management_plan_chain...
| RunnableLambda(lambda x: combine_all(self, x["branches"]["demographics_review"], x["branches"]["management_plan"]...
)
But this does not work. For one thing, even the model (self.general_model) cannot be reached, since it belongs to the class. So that branch provides no output.
How do I pass self into the RunnableLambda to use methods and variables that are given in the class?
I am using LangChain and have a class with many methods. I intend to use parallel chains to process lots of data.
Here is one of my steps, which happens to use other methods in the class in which it is contained:
def management_plan(self, input) -> str: # uses management_plan_model
print("PROCESSING write_notes coroutine: starting management_plan()")
for notes_template_item in self.notes_template_item_list:
if notes_template_item in self.similar_prompt_items['MANAGEMENT PLAN']:
prompt = self.personality[notes_template_item].replace("{input_text}", input.messages[0].content)
context = self.modify_context(self.personality['CONTEXT'], RAG_data=self.rag_bot.output+self.rag_bot.issues_list+self.rag_bot.differentials)
prompt_template = ChatPromptTemplate.from_messages([("system","{context}"), ("human","{prompt}")])
return prompt_template.format_prompt(context=context, prompt=prompt)
This is one of my branches:
demographics_review_chain = (RunnableLambda(lambda x: demographics_review(self, input=x)) | self.general_model | StrOutputParser())
This is a part of the chain:
chain = (
input_template
| RunnableParallel(branches={"demographics_review":demographics_review_chain, "management_plan":management_plan_chain...
| RunnableLambda(lambda x: combine_all(self, x["branches"]["demographics_review"], x["branches"]["management_plan"]...
)
But this does not work. For one thing, even the model (self.general_model) cannot be reached, since it belongs to the class. So that branch provides no output.
How do I pass self into the RunnableLambda to use methods and variables that are given in the class?
1 Answer
Reset to default 0In most cases,easiest way to "pass self" in your Runnable steps is simply define a lambda that captures self inside your class method.
class MyPipeline:
def __init__(self):
self.general_model = SomeLangChainModel()
...
def demographics_review(self,input):
# do smth with `self`
return "some text"
def management_plan(self, input):
# do something else
return "some other text"
def build_chain(self):
demographics_review_chain = (
RunnableLambda(lambda x:self.demographics_review(x))
| self.general_model
| StrOutputParser()
)
management_plan_chain= (
RunnableLambda(lambda x: self.management_plan(x))
| self.general_model
| StrOutputParser()
)
# now use RunnableParallel,capturing self again to combine:
chain = (
input_template
| RunnableParallel(
{
"demographics_review": demographics_review_chain,
"management_plan":management_plan_chain
}
)
| RunnableLambda(
lambda x: selfbine_all(
x["branches"]["demographics_review"],
x["branches"]["management_plan"]
)
)
)
return chain