mirror of
https://github.com/Mintplex-Labs/langchain-python.git
synced 2026-08-27 01:15:22 -04:00
Nuno Campos
11ab0be11a
<!-- Thank you for contributing to LangChain! Your PR will appear in our release under the title you set. Please make sure it highlights your valuable contribution. Replace this with a description of the change, the issue it fixes (if applicable), and relevant context. List any dependencies required for this change. After you're done, someone will review your PR. They may suggest improvements. If no one reviews your PR within a few days, feel free to @-mention the same people again, as notifications can get lost. Finally, we'd love to show appreciation for your contribution - if you'd like us to shout you out on Twitter, please also include your handle! --> <!-- Remove if not applicable --> Fixes # (issue) #### Before submitting <!-- If you're adding a new integration, please include: 1. a test for the integration - favor unit tests that does not rely on network access. 2. an example notebook showing its use See contribution guidelines for more information on how to write tests, lint etc: https://github.com/hwchase17/langchain/blob/master/.github/CONTRIBUTING.md --> #### Who can review? Tag maintainers/contributors who might be interested: <!-- For a quicker response, figure out the right person to tag with @ @hwchase17 - project lead Tracing / Callbacks - @agola11 Async - @agola11 DataLoaders - @eyurtsev Models - @hwchase17 - @agola11 Agents / Tools / Toolkits - @vowelparrot VectorStores / Retrievers / Memory - @dev2049 -->
290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""Chain that just formats a prompt and calls an LLM."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
|
|
|
|
from pydantic import Extra
|
|
|
|
from langchain.base_language import BaseLanguageModel
|
|
from langchain.callbacks.manager import (
|
|
AsyncCallbackManager,
|
|
AsyncCallbackManagerForChainRun,
|
|
CallbackManager,
|
|
CallbackManagerForChainRun,
|
|
Callbacks,
|
|
)
|
|
from langchain.chains.base import Chain
|
|
from langchain.input import get_colored_text
|
|
from langchain.load.dump import dumpd
|
|
from langchain.prompts.base import BasePromptTemplate
|
|
from langchain.prompts.prompt import PromptTemplate
|
|
from langchain.schema import LLMResult, PromptValue
|
|
|
|
|
|
class LLMChain(Chain):
|
|
"""Chain to run queries against LLMs.
|
|
|
|
Example:
|
|
.. code-block:: python
|
|
|
|
from langchain import LLMChain, OpenAI, PromptTemplate
|
|
prompt_template = "Tell me a {adjective} joke"
|
|
prompt = PromptTemplate(
|
|
input_variables=["adjective"], template=prompt_template
|
|
)
|
|
llm = LLMChain(llm=OpenAI(), prompt=prompt)
|
|
"""
|
|
|
|
@property
|
|
def lc_serializable(self) -> bool:
|
|
return True
|
|
|
|
prompt: BasePromptTemplate
|
|
"""Prompt object to use."""
|
|
llm: BaseLanguageModel
|
|
output_key: str = "text" #: :meta private:
|
|
|
|
class Config:
|
|
"""Configuration for this pydantic object."""
|
|
|
|
extra = Extra.forbid
|
|
arbitrary_types_allowed = True
|
|
|
|
@property
|
|
def input_keys(self) -> List[str]:
|
|
"""Will be whatever keys the prompt expects.
|
|
|
|
:meta private:
|
|
"""
|
|
return self.prompt.input_variables
|
|
|
|
@property
|
|
def output_keys(self) -> List[str]:
|
|
"""Will always return text key.
|
|
|
|
:meta private:
|
|
"""
|
|
return [self.output_key]
|
|
|
|
def _call(
|
|
self,
|
|
inputs: Dict[str, Any],
|
|
run_manager: Optional[CallbackManagerForChainRun] = None,
|
|
) -> Dict[str, str]:
|
|
response = self.generate([inputs], run_manager=run_manager)
|
|
return self.create_outputs(response)[0]
|
|
|
|
def generate(
|
|
self,
|
|
input_list: List[Dict[str, Any]],
|
|
run_manager: Optional[CallbackManagerForChainRun] = None,
|
|
) -> LLMResult:
|
|
"""Generate LLM result from inputs."""
|
|
prompts, stop = self.prep_prompts(input_list, run_manager=run_manager)
|
|
return self.llm.generate_prompt(
|
|
prompts, stop, callbacks=run_manager.get_child() if run_manager else None
|
|
)
|
|
|
|
async def agenerate(
|
|
self,
|
|
input_list: List[Dict[str, Any]],
|
|
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
|
|
) -> LLMResult:
|
|
"""Generate LLM result from inputs."""
|
|
prompts, stop = await self.aprep_prompts(input_list, run_manager=run_manager)
|
|
return await self.llm.agenerate_prompt(
|
|
prompts, stop, callbacks=run_manager.get_child() if run_manager else None
|
|
)
|
|
|
|
def prep_prompts(
|
|
self,
|
|
input_list: List[Dict[str, Any]],
|
|
run_manager: Optional[CallbackManagerForChainRun] = None,
|
|
) -> Tuple[List[PromptValue], Optional[List[str]]]:
|
|
"""Prepare prompts from inputs."""
|
|
stop = None
|
|
if "stop" in input_list[0]:
|
|
stop = input_list[0]["stop"]
|
|
prompts = []
|
|
for inputs in input_list:
|
|
selected_inputs = {k: inputs[k] for k in self.prompt.input_variables}
|
|
prompt = self.prompt.format_prompt(**selected_inputs)
|
|
_colored_text = get_colored_text(prompt.to_string(), "green")
|
|
_text = "Prompt after formatting:\n" + _colored_text
|
|
if run_manager:
|
|
run_manager.on_text(_text, end="\n", verbose=self.verbose)
|
|
if "stop" in inputs and inputs["stop"] != stop:
|
|
raise ValueError(
|
|
"If `stop` is present in any inputs, should be present in all."
|
|
)
|
|
prompts.append(prompt)
|
|
return prompts, stop
|
|
|
|
async def aprep_prompts(
|
|
self,
|
|
input_list: List[Dict[str, Any]],
|
|
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
|
|
) -> Tuple[List[PromptValue], Optional[List[str]]]:
|
|
"""Prepare prompts from inputs."""
|
|
stop = None
|
|
if "stop" in input_list[0]:
|
|
stop = input_list[0]["stop"]
|
|
prompts = []
|
|
for inputs in input_list:
|
|
selected_inputs = {k: inputs[k] for k in self.prompt.input_variables}
|
|
prompt = self.prompt.format_prompt(**selected_inputs)
|
|
_colored_text = get_colored_text(prompt.to_string(), "green")
|
|
_text = "Prompt after formatting:\n" + _colored_text
|
|
if run_manager:
|
|
await run_manager.on_text(_text, end="\n", verbose=self.verbose)
|
|
if "stop" in inputs and inputs["stop"] != stop:
|
|
raise ValueError(
|
|
"If `stop` is present in any inputs, should be present in all."
|
|
)
|
|
prompts.append(prompt)
|
|
return prompts, stop
|
|
|
|
def apply(
|
|
self, input_list: List[Dict[str, Any]], callbacks: Callbacks = None
|
|
) -> List[Dict[str, str]]:
|
|
"""Utilize the LLM generate method for speed gains."""
|
|
callback_manager = CallbackManager.configure(
|
|
callbacks, self.callbacks, self.verbose
|
|
)
|
|
run_manager = callback_manager.on_chain_start(
|
|
dumpd(self),
|
|
{"input_list": input_list},
|
|
)
|
|
try:
|
|
response = self.generate(input_list, run_manager=run_manager)
|
|
except (KeyboardInterrupt, Exception) as e:
|
|
run_manager.on_chain_error(e)
|
|
raise e
|
|
outputs = self.create_outputs(response)
|
|
run_manager.on_chain_end({"outputs": outputs})
|
|
return outputs
|
|
|
|
async def aapply(
|
|
self, input_list: List[Dict[str, Any]], callbacks: Callbacks = None
|
|
) -> List[Dict[str, str]]:
|
|
"""Utilize the LLM generate method for speed gains."""
|
|
callback_manager = AsyncCallbackManager.configure(
|
|
callbacks, self.callbacks, self.verbose
|
|
)
|
|
run_manager = await callback_manager.on_chain_start(
|
|
dumpd(self),
|
|
{"input_list": input_list},
|
|
)
|
|
try:
|
|
response = await self.agenerate(input_list, run_manager=run_manager)
|
|
except (KeyboardInterrupt, Exception) as e:
|
|
await run_manager.on_chain_error(e)
|
|
raise e
|
|
outputs = self.create_outputs(response)
|
|
await run_manager.on_chain_end({"outputs": outputs})
|
|
return outputs
|
|
|
|
def create_outputs(self, response: LLMResult) -> List[Dict[str, str]]:
|
|
"""Create outputs from response."""
|
|
return [
|
|
# Get the text of the top generated string.
|
|
{self.output_key: generation[0].text}
|
|
for generation in response.generations
|
|
]
|
|
|
|
async def _acall(
|
|
self,
|
|
inputs: Dict[str, Any],
|
|
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
|
|
) -> Dict[str, str]:
|
|
response = await self.agenerate([inputs], run_manager=run_manager)
|
|
return self.create_outputs(response)[0]
|
|
|
|
def predict(self, callbacks: Callbacks = None, **kwargs: Any) -> str:
|
|
"""Format prompt with kwargs and pass to LLM.
|
|
|
|
Args:
|
|
callbacks: Callbacks to pass to LLMChain
|
|
**kwargs: Keys to pass to prompt template.
|
|
|
|
Returns:
|
|
Completion from LLM.
|
|
|
|
Example:
|
|
.. code-block:: python
|
|
|
|
completion = llm.predict(adjective="funny")
|
|
"""
|
|
return self(kwargs, callbacks=callbacks)[self.output_key]
|
|
|
|
async def apredict(self, callbacks: Callbacks = None, **kwargs: Any) -> str:
|
|
"""Format prompt with kwargs and pass to LLM.
|
|
|
|
Args:
|
|
callbacks: Callbacks to pass to LLMChain
|
|
**kwargs: Keys to pass to prompt template.
|
|
|
|
Returns:
|
|
Completion from LLM.
|
|
|
|
Example:
|
|
.. code-block:: python
|
|
|
|
completion = llm.predict(adjective="funny")
|
|
"""
|
|
return (await self.acall(kwargs, callbacks=callbacks))[self.output_key]
|
|
|
|
def predict_and_parse(
|
|
self, callbacks: Callbacks = None, **kwargs: Any
|
|
) -> Union[str, List[str], Dict[str, Any]]:
|
|
"""Call predict and then parse the results."""
|
|
result = self.predict(callbacks=callbacks, **kwargs)
|
|
if self.prompt.output_parser is not None:
|
|
return self.prompt.output_parser.parse(result)
|
|
else:
|
|
return result
|
|
|
|
async def apredict_and_parse(
|
|
self, callbacks: Callbacks = None, **kwargs: Any
|
|
) -> Union[str, List[str], Dict[str, str]]:
|
|
"""Call apredict and then parse the results."""
|
|
result = await self.apredict(callbacks=callbacks, **kwargs)
|
|
if self.prompt.output_parser is not None:
|
|
return self.prompt.output_parser.parse(result)
|
|
else:
|
|
return result
|
|
|
|
def apply_and_parse(
|
|
self, input_list: List[Dict[str, Any]], callbacks: Callbacks = None
|
|
) -> Sequence[Union[str, List[str], Dict[str, str]]]:
|
|
"""Call apply and then parse the results."""
|
|
result = self.apply(input_list, callbacks=callbacks)
|
|
return self._parse_result(result)
|
|
|
|
def _parse_result(
|
|
self, result: List[Dict[str, str]]
|
|
) -> Sequence[Union[str, List[str], Dict[str, str]]]:
|
|
if self.prompt.output_parser is not None:
|
|
return [
|
|
self.prompt.output_parser.parse(res[self.output_key]) for res in result
|
|
]
|
|
else:
|
|
return result
|
|
|
|
async def aapply_and_parse(
|
|
self, input_list: List[Dict[str, Any]], callbacks: Callbacks = None
|
|
) -> Sequence[Union[str, List[str], Dict[str, str]]]:
|
|
"""Call apply and then parse the results."""
|
|
result = await self.aapply(input_list, callbacks=callbacks)
|
|
return self._parse_result(result)
|
|
|
|
@property
|
|
def _chain_type(self) -> str:
|
|
return "llm_chain"
|
|
|
|
@classmethod
|
|
def from_string(cls, llm: BaseLanguageModel, template: str) -> LLMChain:
|
|
"""Create LLMChain from LLM and template."""
|
|
prompt_template = PromptTemplate.from_template(template)
|
|
return cls(llm=llm, prompt=prompt_template)
|