| from __future__ import annotations |
|
|
| import json |
| from typing import Any, Dict, List, Optional |
|
|
| from langchain_core._api import deprecated |
| from langchain_core.callbacks import CallbackManagerForChainRun |
| from langchain_core.language_models import BaseLanguageModel |
| from langchain_core.prompts import BasePromptTemplate |
| from langchain_text_splitters import RecursiveCharacterTextSplitter, TextSplitter |
| from pydantic import Field |
|
|
| from langchain.chains.base import Chain |
| from langchain.chains.llm import LLMChain |
| from langchain.chains.qa_generation.prompt import PROMPT_SELECTOR |
|
|
|
|
| @deprecated( |
| since="0.2.7", |
| alternative=( |
| "example in API reference with more detail: " |
| "https://api.python.langchain.com/en/latest/chains/langchain.chains.qa_generation.base.QAGenerationChain.html" |
| ), |
| removal="1.0", |
| ) |
| class QAGenerationChain(Chain): |
| """Base class for question-answer generation chains. |
| |
| This class is deprecated. See below for an alternative implementation. |
| |
| Advantages of this implementation include: |
| |
| - Supports async and streaming; |
| - Surfaces prompt and text splitter for easier customization; |
| - Use of JsonOutputParser supports JSONPatch operations in streaming mode, |
| as well as robustness to markdown. |
| |
| .. code-block:: python |
| |
| from langchain.chains.qa_generation.prompt import CHAT_PROMPT as prompt |
| # Note: import PROMPT if using a legacy non-chat model. |
| from langchain_core.output_parsers import JsonOutputParser |
| from langchain_core.runnables import ( |
| RunnableLambda, |
| RunnableParallel, |
| RunnablePassthrough, |
| ) |
| from langchain_core.runnables.base import RunnableEach |
| from langchain_openai import ChatOpenAI |
| from langchain_text_splitters import RecursiveCharacterTextSplitter |
| |
| llm = ChatOpenAI() |
| text_splitter = RecursiveCharacterTextSplitter(chunk_overlap=500) |
| split_text = RunnableLambda( |
| lambda x: text_splitter.create_documents([x]) |
| ) |
| |
| chain = RunnableParallel( |
| text=RunnablePassthrough(), |
| questions=( |
| split_text | RunnableEach(bound=prompt | llm | JsonOutputParser()) |
| ) |
| ) |
| """ |
|
|
| llm_chain: LLMChain |
| """LLM Chain that generates responses from user input and context.""" |
| text_splitter: TextSplitter = Field( |
| default=RecursiveCharacterTextSplitter(chunk_overlap=500) |
| ) |
| """Text splitter that splits the input into chunks.""" |
| input_key: str = "text" |
| """Key of the input to the chain.""" |
| output_key: str = "questions" |
| """Key of the output of the chain.""" |
| k: Optional[int] = None |
| """Number of questions to generate.""" |
|
|
| @classmethod |
| def from_llm( |
| cls, |
| llm: BaseLanguageModel, |
| prompt: Optional[BasePromptTemplate] = None, |
| **kwargs: Any, |
| ) -> QAGenerationChain: |
| """ |
| Create a QAGenerationChain from a language model. |
| |
| Args: |
| llm: a language model |
| prompt: a prompt template |
| **kwargs: additional arguments |
| |
| Returns: |
| a QAGenerationChain class |
| """ |
| _prompt = prompt or PROMPT_SELECTOR.get_prompt(llm) |
| chain = LLMChain(llm=llm, prompt=_prompt) |
| return cls(llm_chain=chain, **kwargs) |
|
|
| @property |
| def _chain_type(self) -> str: |
| raise NotImplementedError |
|
|
| @property |
| def input_keys(self) -> List[str]: |
| return [self.input_key] |
|
|
| @property |
| def output_keys(self) -> List[str]: |
| return [self.output_key] |
|
|
| def _call( |
| self, |
| inputs: Dict[str, Any], |
| run_manager: Optional[CallbackManagerForChainRun] = None, |
| ) -> Dict[str, List]: |
| docs = self.text_splitter.create_documents([inputs[self.input_key]]) |
| results = self.llm_chain.generate( |
| [{"text": d.page_content} for d in docs], run_manager=run_manager |
| ) |
| qa = [json.loads(res[0].text) for res in results.generations] |
| return {self.output_key: qa} |
|
|
