ValueError: Missing some input keys: {'input_documents'}

[godiegege]

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 langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.chains.retrieval import create_retrieval_chain
from langchain.chains.combine_documents.refine import RefineDocumentsChain
from langchain_community.llms.tongyi import Tongyi

llm = Tongyi()

document_prompt = PromptTemplate(
    input_variables=["page_content"],
        template="{page_content}"
)
document_variable_name = "context"

prompt = PromptTemplate.from_template(
    "Summarize this content: {context}"
)
initial_llm_chain = LLMChain(llm=llm, prompt=prompt)
initial_response_name = "prev_response"


prompt_refine = PromptTemplate.from_template(
    "Here's your first summary: {prev_response}. "
    "Now add to it based on the following context: {context}"
)
refine_llm_chain = LLMChain(llm=llm, prompt=prompt_refine)
chain = RefineDocumentsChain(
    initial_llm_chain=initial_llm_chain,
    refine_llm_chain=refine_llm_chain,
    document_prompt=document_prompt,
    document_variable_name=document_variable_name,
    initial_response_name=initial_response_name,
)
qa_chain = create_retrieval_chain(retriever=data_db1.as_retriever(), combine_docs_chain=chain)
qa_chain.invoke({'input': "stm32"})

Description

---

ValueError Traceback (most recent call last)
Cell In[9], line 1
----> 1 qa_chain1.invoke({'input': 'stm32'})

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\base.py:4580, in RunnableBindingBase.invoke(self, input, config, **kwargs)
4574 def invoke(
4575 self,
4576 input: Input,
4577 config: Optional[RunnableConfig] = None,
4578 **kwargs: Optional[Any],
4579 ) -> Output:
-> 4580 return self.bound.invoke(
4581 input,
4582 self._merge_configs(config),
4583 **{**self.kwargs, **kwargs},
4584 )

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\base.py:2499, in RunnableSequence.invoke(self, input, config, **kwargs)
2497 input = step.invoke(input, config, **kwargs)
2498 else:
-> 2499 input = step.invoke(input, config)
2500 # finish the root run
2501 except BaseException as e:

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\passthrough.py:469, in RunnableAssign.invoke(self, input, config, **kwargs)
463 def invoke(
464 self,
465 input: Dict[str, Any],
466 config: Optional[RunnableConfig] = None,
467 **kwargs: Any,
468 ) -> Dict[str, Any]:
--> 469 return self._call_with_config(self._invoke, input, config, **kwargs)

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\base.py:1593, in Runnable._call_with_config(self, func, input, config, run_type, **kwargs)
1589 context = copy_context()
1590 context.run(_set_config_context, child_config)
1591 output = cast(
1592 Output,
-> 1593 context.run(
1594 call_func_with_variable_args, # type: ignore[arg-type]
1595 func, # type: ignore[arg-type]
1596 input, # type: ignore[arg-type]
1597 config,
1598 run_manager,
1599 **kwargs,
1600 ),
1601 )
1602 except BaseException as e:
1603 run_manager.on_chain_error(e)

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\config.py:380, in call_func_with_variable_args(func, input, config, run_manager, **kwargs)
378 if run_manager is not None and accepts_run_manager(func):
379 kwargs["run_manager"] = run_manager
--> 380 return func(input, **kwargs)

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\passthrough.py:456, in RunnableAssign._invoke(self, input, run_manager, config, **kwargs)
443 def _invoke(
444 self,
445 input: Dict[str, Any],
(...)
448 **kwargs: Any,
449 ) -> Dict[str, Any]:
450 assert isinstance(
451 input, dict
452 ), "The input to RunnablePassthrough.assign() must be a dict."
454 return {
455 **input,
--> 456 **self.mapper.invoke(
457 input,
458 patch_config(config, callbacks=run_manager.get_child()),
459 **kwargs,
460 ),
461 }

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\base.py:3144, in RunnableParallel.invoke(self, input, config)
3131 with get_executor_for_config(config) as executor:
3132 futures = [
3133 executor.submit(
3134 step.invoke,
(...)
3142 for key, step in steps.items()
3143 ]
-> 3144 output = {key: future.result() for key, future in zip(steps, futures)}
3145 # finish the root run
3146 except BaseException as e:

File d:\Program Files\Python39\lib\site-packages\langchain_core\runnables\base.py:3144, in (.0)
3131 with get_executor_for_config(config) as executor:
3132 futures = [
3133 executor.submit(
3134 step.invoke,
(...)
3142 for key, step in steps.items()
3143 ]
-> 3144 output = {key: future.result() for key, future in zip(steps, futures)}
3145 # finish the root run
3146 except BaseException as e:

File d:\Program Files\Python39\lib\concurrent\futures_base.py:439, in Future.result(self, timeout)
437 raise CancelledError()
438 elif self._state == FINISHED:
--> 439 return self.__get_result()
441 self._condition.wait(timeout)
443 if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:

File d:\Program Files\Python39\lib\concurrent\futures_base.py:391, in Future.__get_result(self)
389 if self._exception:
390 try:
--> 391 raise self._exception
392 finally:
393 # Break a reference cycle with the exception in self._exception
394 self = None

File d:\Program Files\Python39\lib\concurrent\futures\thread.py:58, in _WorkItem.run(self)
55 return
57 try:
---> 58 result = self.fn(*self.args, **self.kwargs)
59 except BaseException as exc:
60 self.future.set_exception(exc)

File d:\Program Files\Python39\lib\site-packages\langchain\chains\base.py:166, in Chain.invoke(self, input, config, **kwargs)
164 except BaseException as e:
165 run_manager.on_chain_error(e)
--> 166 raise e
167 run_manager.on_chain_end(outputs)
169 if include_run_info:

File d:\Program Files\Python39\lib\site-packages\langchain\chains\base.py:154, in Chain.invoke(self, input, config, **kwargs)
147 run_manager = callback_manager.on_chain_start(
148 dumpd(self),
149 inputs,
150 run_id,
151 name=run_name,
152 )
153 try:
--> 154 self._validate_inputs(inputs)
155 outputs = (
156 self._call(inputs, run_manager=run_manager)
157 if new_arg_supported
158 else self._call(inputs)
159 )
161 final_outputs: Dict[str, Any] = self.prep_outputs(
162 inputs, outputs, return_only_outputs
163 )

File d:\Program Files\Python39\lib\site-packages\langchain\chains\base.py:284, in Chain._validate_inputs(self, inputs)
282 missing_keys = set(self.input_keys).difference(inputs)
283 if missing_keys:
--> 284 raise ValueError(f"Missing some input keys: {missing_keys}")

ValueError: Missing some input keys: {'input_documents'}

System Info

langchang v0.2.7
system windows11
python 3.9.13

[dosubot[bot]]

Hey there, @godiegege! I'm here to help you with any questions or bugs you might have. Whether it's solving code issues, answering technical questions, or becoming a contributor, I've got you covered. Let me know how I can assist you!

To resolve the ValueError: Missing some input keys: {'input_documents'} error, you need to ensure that the input dictionary you pass to the chain includes the key input_documents. This key is expected by the RefineDocumentsChain class.

Here's how you can adjust your code:

1. Ensure you have a list of Document objects.
2. Include the input_documents key in the input dictionary.

Here's an example:

from langchain_core.documents import Document

# Assuming you have a list of Document objects
documents = [Document(page_content="Document 1 content"), Document(page_content="Document 2 content")]

# Create the input dictionary with the required key
inputs = {
    "input_documents": documents,
    "input": "stm32"
}

# Pass the inputs to your chain
result = qa_chain.invoke(inputs)

Make sure that the key input_documents is present in the dictionary and that it maps to a list of Document objects. This should resolve the ValueError ^[1][2].

To continue talking to Dosu, mention @dosu.

[godiegege]

I have a question about why I need to pass {"input_documents": documents} during invocation even though I already provided a retriever in qa_chain = create_retrieval_chain(retriever=data_db1.as_retriever(), combine_docs_chain=chain),thanks.@dosu

[dosubot[bot]]

You need to pass {"input_documents": documents} during invocation because the combine_docs_chain expects the retrieved documents to be provided in the input dictionary under the key input_documents. This is necessary for the combine_docs_chain to process the documents and produce the final output.

The create_retrieval_chain function sets up the chain such that the retriever retrieves documents and assigns them to the context key. However, when invoking the chain, you need to explicitly pass the documents to ensure they are available for the combine_docs_chain to use.

Here is the relevant part of the code that explains this:

retrieval_chain = (
    RunnablePassthrough.assign(
        context=parse_retriever_input | retriever,
    ).assign(answer=document_chain)
).with_config(run_name="retrieval_chain")

This setup ensures that the retrieved documents are assigned to the context key, but you still need to pass the documents in the input dictionary when invoking the chain. This is illustrated in the LangChain documentation where the document_chain is invoked with the retrieved documents as context:

document_chain.invoke(
    {
        "context": docs,
        "messages": [
            HumanMessage(content="Can LangSmith help test my LLM applications?")
        ],
    }
)

This ensures that the documents are correctly passed to the combine_docs_chain for processing ^[1][2].

To continue talking to Dosu, mention @dosu.

[godiegege]

When i pass the docoments to the dicts the inference time is very long.how to solve it?thanks.@dosu

[dosubot[bot]]

To reduce the inference time when passing documents to the input dictionary, consider the following strategies:

1. Batching Requests: Process multiple documents in batches to better utilize computational resources.

   def batch_requests(requests, batch_size):
       for i in range(0, len(requests), batch_size):
           yield requests[i:i + batch_size]
   
   # Usage
   documents = [...]  # List of Document objects
   batch_size = 10
   for batch in batch_requests(documents, batch_size):
       inputs = {"input_documents": batch, "input": "stm32"}
       result = qa_chain.invoke(inputs)

2. Self-hosting Models: Use smaller, self-hosted models to reduce latency.

   from langchain_community.embeddings import SelfHostedEmbeddings
   from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
   import runhouse as rh
   
   gpu = rh.cluster(name="rh-a10x", instance_type="A100:1")
   def get_pipeline():
       model_id = "facebook/bart-large"
       tokenizer = AutoTokenizer.from_pretrained(model_id)
       model = AutoModelForCausalLM.from_pretrained(model_id)
       return pipeline("feature-extraction", model=model, tokenizer=tokenizer)
   embeddings = SelfHostedEmbeddings(
       model_load_fn=get_pipeline,
       hardware=gpu,
       model_reqs=["./", "torch", "transformers"],
   )

3. Quantized Models: Use quantized models to increase speed.

   from langchain_community.embeddings import QuantizedBiEncoderEmbeddings
   
   model_name = "Intel/bge-small-en-v1.5-rag-int8-static"
   encode_kwargs = {'normalize_embeddings': True}
   hf = QuantizedBiEncoderEmbeddings(
       model_name,
       encode_kwargs=encode_kwargs,
       query_instruction="Represent this sentence for searching relevant passages: "
   )

4. Post-Processing and Compression: Compress and filter documents to reduce the load.

   from langchain.retrievers.document_compressors import EmbeddingsFilter
   from langchain_openai import OpenAIEmbeddings
   from langchain_text_splitters import RecursiveCharacterTextSplitter
   
   splitter = RecursiveCharacterTextSplitter(
       chunk_size=400,
       chunk_overlap=0,
       separators=["\n\n", "\n", ".", " "],
       keep_separator=False,
   )
   compressor = EmbeddingsFilter(embeddings=OpenAIEmbeddings(), k=10)
   
   def split_and_filter(input) -> List[Document]:
       docs = input["docs"]
       question = input["question"]
       split_docs = splitter.split_documents(docs)
       stateful_docs = compressor.compress_documents(split_docs, question)
       return [stateful_doc for stateful_doc in stateful_docs]
   
   retrieve = (
       RunnableParallel(question=RunnablePassthrough(), docs=wiki) | split_and_filter
   )
   docs = retrieve.invoke("How fast are cheetahs?")
   for doc in docs:
       print(doc.page_content)
       print("\n\n")

By implementing these strategies, you can optimize the inference time and improve the overall performance of your application ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.