Source code for langchain_experimental.fallacy_removal.base

"""Chain for applying removals of logical fallacies."""
from __future__ import annotations

from typing import Any, Dict, List, Optional

from langchain.chains.base import Chain
from langchain.chains.llm import LLMChain
from langchain.schema import BasePromptTemplate
from langchain_core.callbacks.manager import CallbackManagerForChainRun
from langchain_core.language_models import BaseLanguageModel

from langchain_experimental.fallacy_removal.fallacies import FALLACIES
from langchain_experimental.fallacy_removal.models import LogicalFallacy
from langchain_experimental.fallacy_removal.prompts import (
    FALLACY_CRITIQUE_PROMPT,
    FALLACY_REVISION_PROMPT,
)


[docs]class FallacyChain(Chain): """Chain for applying logical fallacy evaluations. It is modeled after Constitutional AI and in same format, but applying logical fallacies as generalized rules to remove in output. Example: .. code-block:: python from langchain_community.llms import OpenAI from langchain.chains import LLMChain from langchain_experimental.fallacy import FallacyChain from langchain_experimental.fallacy_removal.models import LogicalFallacy llm = OpenAI() qa_prompt = PromptTemplate( template="Q: {question} A:", input_variables=["question"], ) qa_chain = LLMChain(llm=llm, prompt=qa_prompt) fallacy_chain = FallacyChain.from_llm( llm=llm, chain=qa_chain, logical_fallacies=[ LogicalFallacy( fallacy_critique_request="Tell if this answer meets criteria.", fallacy_revision_request=\ "Give an answer that meets better criteria.", ) ], ) fallacy_chain.run(question="How do I know if the earth is round?") """ chain: LLMChain logical_fallacies: List[LogicalFallacy] fallacy_critique_chain: LLMChain fallacy_revision_chain: LLMChain return_intermediate_steps: bool = False
[docs] @classmethod def get_fallacies(cls, names: Optional[List[str]] = None) -> List[LogicalFallacy]: if names is None: return list(FALLACIES.values()) else: return [FALLACIES[name] for name in names]
[docs] @classmethod def from_llm( cls, llm: BaseLanguageModel, chain: LLMChain, fallacy_critique_prompt: BasePromptTemplate = FALLACY_CRITIQUE_PROMPT, fallacy_revision_prompt: BasePromptTemplate = FALLACY_REVISION_PROMPT, **kwargs: Any, ) -> "FallacyChain": """Create a chain from an LLM.""" fallacy_critique_chain = LLMChain(llm=llm, prompt=fallacy_critique_prompt) fallacy_revision_chain = LLMChain(llm=llm, prompt=fallacy_revision_prompt) return cls( chain=chain, fallacy_critique_chain=fallacy_critique_chain, fallacy_revision_chain=fallacy_revision_chain, **kwargs, )
@property def input_keys(self) -> List[str]: """Input keys.""" return self.chain.input_keys @property def output_keys(self) -> List[str]: """Output keys.""" if self.return_intermediate_steps: return ["output", "fallacy_critiques_and_revisions", "initial_output"] return ["output"] def _call( self, inputs: Dict[str, Any], run_manager: Optional[CallbackManagerForChainRun] = None, ) -> Dict[str, Any]: _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager() response = self.chain.run( **inputs, callbacks=_run_manager.get_child("original"), ) initial_response = response input_prompt = self.chain.prompt.format(**inputs) _run_manager.on_text( text="Initial response: " + response + "\n\n", verbose=self.verbose, color="yellow", ) fallacy_critiques_and_revisions = [] for logical_fallacy in self.logical_fallacies: # Fallacy critique below fallacy_raw_critique = self.fallacy_critique_chain.run( input_prompt=input_prompt, output_from_model=response, fallacy_critique_request=logical_fallacy.fallacy_critique_request, callbacks=_run_manager.get_child("fallacy_critique"), ) fallacy_critique = self._parse_critique( output_string=fallacy_raw_critique, ).strip() # if fallacy critique contains "No fallacy critique needed" then done if "no fallacy critique needed" in fallacy_critique.lower(): fallacy_critiques_and_revisions.append((fallacy_critique, "")) continue fallacy_revision = self.fallacy_revision_chain.run( input_prompt=input_prompt, output_from_model=response, fallacy_critique_request=logical_fallacy.fallacy_critique_request, fallacy_critique=fallacy_critique, revision_request=logical_fallacy.fallacy_revision_request, callbacks=_run_manager.get_child("fallacy_revision"), ).strip() response = fallacy_revision fallacy_critiques_and_revisions.append((fallacy_critique, fallacy_revision)) _run_manager.on_text( text=f"Applying {logical_fallacy.name}..." + "\n\n", verbose=self.verbose, color="green", ) _run_manager.on_text( text="Logical Fallacy: " + fallacy_critique + "\n\n", verbose=self.verbose, color="blue", ) _run_manager.on_text( text="Updated response: " + fallacy_revision + "\n\n", verbose=self.verbose, color="yellow", ) final_output: Dict[str, Any] = {"output": response} if self.return_intermediate_steps: final_output["initial_output"] = initial_response final_output[ "fallacy_critiques_and_revisions" ] = fallacy_critiques_and_revisions return final_output @staticmethod def _parse_critique(output_string: str) -> str: if "Fallacy Revision request:" not in output_string: return output_string output_string = output_string.split("Fallacy Revision request:")[0] if "\n\n" in output_string: output_string = output_string.split("\n\n")[0] return output_string