📓 🦜️🔗 LangChain Integration¶
TruLens provides TruChain, a deep integration with LangChain to allow you to inspect and evaluate the internals of your application built using LangChain. This is done through the instrumentation of key LangChain classes. To see a list of classes instrumented, see Appendix: Instrumented _LangChain_ Classes and Methods.
In addition to the default instrumentation, TruChain exposes the select_context method for evaluations that require access to retrieved context. Exposing select_context bypasses the need to know the json structure of your app ahead of time, and makes your evaluations re-usable across different apps.
Example Usage¶
Below is a quick example of usage. First, we'll create a standard LLMChain.
# required imports
from langchain_openai import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.prompts.chat import HumanMessagePromptTemplate, ChatPromptTemplate
from trulens_eval import TruChain
# typical LangChain rag setup
full_prompt = HumanMessagePromptTemplate(
prompt=PromptTemplate(
template=
"Provide a helpful response with relevant background information for the following: {prompt}",
input_variables=["prompt"],
)
)
chat_prompt_template = ChatPromptTemplate.from_messages([full_prompt])
llm = OpenAI(temperature=0.9, max_tokens=128)
chain = LLMChain(llm=llm, prompt=chat_prompt_template, verbose=True)
To instrument an LLM chain, all that's required is to wrap it using TruChain.
# instrument with TruChain
tru_recorder = TruChain(chain)
🦑 Tru initialized with db url sqlite:///default.sqlite . 🛑 Secret keys may be written to the database. See the `database_redact_keys` option of Tru` to prevent this.
Similarly, LangChain apps defined with LangChain Expression Language (LCEL) are also supported.
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
tru_recorder = TruChain(
chain,
app_id='Chain1_ChatApplication'
)
To properly evaluate LLM apps we often need to point our evaluation at an internal step of our application, such as the retreived context. Doing so allows us to evaluate for metrics including context relevance and groundedness.
For LangChain applications where the BaseRetriever is used, select_context
can
be used to access the retrieved text for evaluation.
Example usage:
context = TruChain.select_context(rag_chain)
f_context_relevance = (
Feedback(provider.qs_relevance)
.on_input()
.on(context)
.aggregate(np.mean)
)
For added flexibility, the select_context method is also made available through
trulens_eval.app.App
. This allows you to switch between frameworks without
changing your context selector:
from trulens_eval.app import App
context = App.select_context(rag_chain)
You can find the full quickstart available here: LangChain Quickstart
Async Support¶
TruChain also provides async support for LangChain through the acall
method. This allows you to track and evaluate async and streaming LangChain applications.
As an example, below is an LLM chain set up with an async callback.
from langchain import LLMChain
from langchain.callbacks import AsyncIteratorCallbackHandler
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from trulens_eval import TruChain
# Set up an async callback.
callback = AsyncIteratorCallbackHandler()
# Setup a simple question/answer chain with streaming ChatOpenAI.
prompt = PromptTemplate.from_template("Honestly answer this question: {question}.")
llm = ChatOpenAI(
temperature=0.0,
streaming=True, # important
callbacks=[callback]
)
async_chain = LLMChain(llm=llm, prompt=prompt)
Once you have created the async LLM chain you can instrument it just as before.
async_tc_recorder = TruChain(async_chain)
with async_tc_recorder as recording:
await async_chain.ainvoke(input=dict(question="What is 1+2? Explain your answer."))
For more usage examples, check out the LangChain examples directory.
Appendix: Instrumented LangChain Classes and Methods¶
The modules, classes, and methods that trulens instruments can be retrieved from the appropriate Instrument subclass.
from trulens_eval.tru_chain import LangChainInstrument
LangChainInstrument().print_instrumentation()
Module langchain* Class langchain.agents.agent.BaseMultiActionAgent Method plan: (self, intermediate_steps: 'List[Tuple[AgentAction, str]]', callbacks: 'Callbacks' = None, **kwargs: 'Any') -> 'Union[List[AgentAction], AgentFinish]' Method aplan: (self, intermediate_steps: 'List[Tuple[AgentAction, str]]', callbacks: 'Callbacks' = None, **kwargs: 'Any') -> 'Union[List[AgentAction], AgentFinish]' Class langchain.agents.agent.BaseSingleActionAgent Method plan: (self, intermediate_steps: 'List[Tuple[AgentAction, str]]', callbacks: 'Callbacks' = None, **kwargs: 'Any') -> 'Union[AgentAction, AgentFinish]' Method aplan: (self, intermediate_steps: 'List[Tuple[AgentAction, str]]', callbacks: 'Callbacks' = None, **kwargs: 'Any') -> 'Union[AgentAction, AgentFinish]' Class langchain.chains.base.Chain Method __call__: (self, inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False, callbacks: Union[List[langchain_core.callbacks.base.BaseCallbackHandler], langchain_core.callbacks.base.BaseCallbackManager, NoneType] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, include_run_info: bool = False) -> Dict[str, Any] Method invoke: (self, input: Dict[str, Any], config: Optional[langchain_core.runnables.config.RunnableConfig] = None, **kwargs: Any) -> Dict[str, Any] Method ainvoke: (self, input: Dict[str, Any], config: Optional[langchain_core.runnables.config.RunnableConfig] = None, **kwargs: Any) -> Dict[str, Any] Method run: (self, *args: Any, callbacks: Union[List[langchain_core.callbacks.base.BaseCallbackHandler], langchain_core.callbacks.base.BaseCallbackManager, NoneType] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Any Method arun: (self, *args: Any, callbacks: Union[List[langchain_core.callbacks.base.BaseCallbackHandler], langchain_core.callbacks.base.BaseCallbackManager, NoneType] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Any Method _call: (self, inputs: Dict[str, Any], run_manager: Optional[langchain_core.callbacks.manager.CallbackManagerForChainRun] = None) -> Dict[str, Any] Method _acall: (self, inputs: Dict[str, Any], run_manager: Optional[langchain_core.callbacks.manager.AsyncCallbackManagerForChainRun] = None) -> Dict[str, Any] Method acall: (self, inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False, callbacks: Union[List[langchain_core.callbacks.base.BaseCallbackHandler], langchain_core.callbacks.base.BaseCallbackManager, NoneType] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, include_run_info: bool = False) -> Dict[str, Any] Class langchain.memory.chat_memory.BaseChatMemory Method save_context: (self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None Method clear: (self) -> None Class langchain_core.chat_history.BaseChatMessageHistory Class langchain_core.documents.base.Document Class langchain_core.language_models.base.BaseLanguageModel Class langchain_core.language_models.llms.BaseLLM Class langchain_core.load.serializable.Serializable Class langchain_core.memory.BaseMemory Method save_context: (self, inputs: 'Dict[str, Any]', outputs: 'Dict[str, str]') -> 'None' Method clear: (self) -> 'None' Class langchain_core.prompts.base.BasePromptTemplate Class langchain_core.retrievers.BaseRetriever Method _get_relevant_documents: (self, query: 'str', *, run_manager: 'CallbackManagerForRetrieverRun') -> 'List[Document]' Method get_relevant_documents: (self, query: 'str', *, callbacks: 'Callbacks' = None, tags: 'Optional[List[str]]' = None, metadata: 'Optional[Dict[str, Any]]' = None, run_name: 'Optional[str]' = None, **kwargs: 'Any') -> 'List[Document]' Method aget_relevant_documents: (self, query: 'str', *, callbacks: 'Callbacks' = None, tags: 'Optional[List[str]]' = None, metadata: 'Optional[Dict[str, Any]]' = None, run_name: 'Optional[str]' = None, **kwargs: 'Any') -> 'List[Document]' Method _aget_relevant_documents: (self, query: 'str', *, run_manager: 'AsyncCallbackManagerForRetrieverRun') -> 'List[Document]' Class langchain_core.runnables.base.RunnableSerializable Class langchain_core.tools.BaseTool Method _arun: (self, *args: 'Any', **kwargs: 'Any') -> 'Any' Method _run: (self, *args: 'Any', **kwargs: 'Any') -> 'Any' Module trulens_eval.* Class trulens_eval.feedback.feedback.Feedback Method __call__: (self, *args, **kwargs) -> 'Any' Class trulens_eval.utils.langchain.WithFeedbackFilterDocuments Method _get_relevant_documents: (self, query: str, *, run_manager) -> List[langchain_core.documents.base.Document] Method get_relevant_documents: (self, query: 'str', *, callbacks: 'Callbacks' = None, tags: 'Optional[List[str]]' = None, metadata: 'Optional[Dict[str, Any]]' = None, run_name: 'Optional[str]' = None, **kwargs: 'Any') -> 'List[Document]' Method aget_relevant_documents: (self, query: 'str', *, callbacks: 'Callbacks' = None, tags: 'Optional[List[str]]' = None, metadata: 'Optional[Dict[str, Any]]' = None, run_name: 'Optional[str]' = None, **kwargs: 'Any') -> 'List[Document]' Method _aget_relevant_documents: (self, query: 'str', *, run_manager: 'AsyncCallbackManagerForRetrieverRun') -> 'List[Document]'
Instrumenting other classes/methods.¶
Additional classes and methods can be instrumented by use of the
trulens_eval.instruments.Instrument
methods and decorators. Examples of
such usage can be found in the custom app used in the custom_example.ipynb
notebook which can be found in
trulens_eval/examples/expositional/end2end_apps/custom_app/custom_app.py
. More
information about these decorators can be found in the
docs/trulens_eval/tracking/instrumentation/index.ipynb
notebook.
Inspecting instrumentation¶
The specific objects (of the above classes) and methods instrumented for a
particular app can be inspected using the App.print_instrumented
as
exemplified in the next cell. Unlike Instrument.print_instrumentation
, this
function only shows what in an app was actually instrumented.
async_tc_recorder.print_instrumented()
Components: TruChain (Other) at 0x2b60a3660 with path __app__ LLMChain (Other) at 0x2b5cdb3e0 with path __app__.app PromptTemplate (Custom) at 0x2b605e580 with path __app__.app.prompt ChatOpenAI (Custom) at 0x2b5cdb4d0 with path __app__.app.llm StrOutputParser (Custom) at 0x2b60a3750 with path __app__.app.output_parser Methods: Object at 0x2b5cdb3e0: <function Chain.__call__ at 0x2a6c17560> with path __app__.app <function Chain.invoke at 0x2a6c16de0> with path __app__.app <function Chain.ainvoke at 0x2a6c16e80> with path __app__.app <function Chain.run at 0x2a6c17b00> with path __app__.app <function Chain.arun at 0x2a6c17d80> with path __app__.app <function LLMChain._call at 0x2a6c6c2c0> with path __app__.app <function LLMChain._acall at 0x2a6c6c860> with path __app__.app <function Chain.acall at 0x2a6c177e0> with path __app__.app <function Chain._call at 0x2a6c17380> with path __app__.app <function Chain._acall at 0x2a6c17420> with path __app__.app <function Runnable.invoke at 0x2a669ba60> with path __app__.app <function Runnable.ainvoke at 0x2a669bb00> with path __app__.app