Tru Llama

Llama_index instrumentation and monitoring.

TruLlama

Bases: App

Instantiates the LLama Index Wrapper.

Usage:

LLama-Index code: LLama Index Quickstart

 # Code snippet taken from llama_index 0.8.15 (API subject to change with new versions)
from llama_index import VectorStoreIndex, SimpleWebPageReader

documents = SimpleWebPageReader(
    html_to_text=True
).load_data(["http://paulgraham.com/worked.html"])
index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine()

Trulens Eval Code:

from trulens_eval import TruLlama
# f_lang_match, f_qa_relevance, f_qs_relevance are feedback functions
tru_recorder = TruLlama(query_engine,
    app_id='LlamaIndex_App1',
    feedbacks=[f_lang_match, f_qa_relevance, f_qs_relevance])

with tru_recorder as recording:
    query_engine.query("What is llama index?")

tru_record = recording.records[0]

# To add record metadata 
with tru_recorder as recording:
    recording.record_metadata="this is metadata for all records in this context that follow this line"
    query_engine.query("What is llama index?")
    recording.record_metadata="this is different metadata for all records in this context that follow this line"
    query_engine.query("Where do I download llama index?")

See Feedback Functions for instantiating feedback functions.

Parameters:

Name	Type	Description	Default
app	BaseQueryEngine	A llama index application.	required

Source code in trulens_eval/trulens_eval/tru_llama.py

class TruLlama(App):
    """Instantiates the LLama Index Wrapper.

        **Usage:**

        LLama-Index code: [LLama Index Quickstart](https://gpt-index.readthedocs.io/en/stable/getting_started/starter_example.html)
        ```
         # Code snippet taken from llama_index 0.8.15 (API subject to change with new versions)
        from llama_index import VectorStoreIndex, SimpleWebPageReader

        documents = SimpleWebPageReader(
            html_to_text=True
        ).load_data(["http://paulgraham.com/worked.html"])
        index = VectorStoreIndex.from_documents(documents)

        query_engine = index.as_query_engine()

        ```

        Trulens Eval Code:
        ```
        from trulens_eval import TruLlama
        # f_lang_match, f_qa_relevance, f_qs_relevance are feedback functions
        tru_recorder = TruLlama(query_engine,
            app_id='LlamaIndex_App1',
            feedbacks=[f_lang_match, f_qa_relevance, f_qs_relevance])

        with tru_recorder as recording:
            query_engine.query("What is llama index?")

        tru_record = recording.records[0]

        # To add record metadata 
        with tru_recorder as recording:
            recording.record_metadata="this is metadata for all records in this context that follow this line"
            query_engine.query("What is llama index?")
            recording.record_metadata="this is different metadata for all records in this context that follow this line"
            query_engine.query("Where do I download llama index?")


        ```
        See [Feedback Functions](https://www.trulens.org/trulens_eval/api/feedback/) for instantiating feedback functions.

        Args:
            app (BaseQueryEngine): A llama index application.
    """

    class Config:
        arbitrary_types_allowed = True

    app: Union[BaseQueryEngine, BaseChatEngine]

    root_callable: ClassVar[FunctionOrMethod] = Field(
        default_factory=lambda: FunctionOrMethod.of_callable(TruLlama.query),
        const=True
    )

    def __init__(self, app: BaseQueryEngine, **kwargs):
        super().update_forward_refs()

        # TruLlama specific:
        kwargs['app'] = app
        kwargs['root_class'] = Class.of_object(app)  # TODO: make class property
        kwargs['instrument'] = LlamaInstrument(app=self)

        super().__init__(**kwargs)

        self.post_init()

    @classmethod
    def select_source_nodes(cls) -> JSONPath:
        """
        Get the path to the source nodes in the query output.
        """
        return cls.select_outputs().source_nodes[:]

    def main_input(
        self, func: Callable, sig: Signature, bindings: BoundArguments
    ) -> str:
        """
        Determine the main input string for the given function `func` with
        signature `sig` if it is to be called with the given bindings
        `bindings`.
        """

        if 'str_or_query_bundle' in bindings.arguments:
            # llama_index specific
            return bindings.arguments['str_or_query_bundle']

        elif 'message' in bindings.arguments:
            # llama_index specific
            return bindings.arguments['message']

        else:

            return App.main_input(self, func, sig, bindings)

    def main_output(
        self, func: Callable, sig: Signature, bindings: BoundArguments, ret: Any
    ) -> str:
        """
        Determine the main out string for the given function `func` with
        signature `sig` after it is called with the given `bindings` and has
        returned `ret`.
        """

        if isinstance(ret, Response):  # query, aquery
            return ret.response

        elif isinstance(ret, AgentChatResponse):  #  chat, achat
            return ret.response

        elif isinstance(ret, (StreamingResponse, StreamingAgentChatResponse)):
            logger.warning(
                "App produced a streaming response. "
                "Tracking content of streams in llama_index is not yet supported. "
                "App main_output will be None."
            )

            return None

        else:

            return App.main_output(self, func, sig, bindings, ret)

    # TODEP
    # llama_index.chat_engine.types.BaseChatEngine
    def chat(self, *args, **kwargs) -> AgentChatResponse:
        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(method="chat", is_async=False, with_record=False)

        res, _ = self.chat_with_record(*args, **kwargs)
        return res

    # TODEP
    # llama_index.chat_engine.types.BaseChatEngine
    async def achat(self, *args, **kwargs) -> AgentChatResponse:
        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(method="achat", is_async=True, with_record=False)

        res, _ = await self.achat_with_record(*args, **kwargs)
        return res

    # TODEP
    # llama_index.chat_engine.types.BaseChatEngine
    def stream_chat(self, *args, **kwargs) -> StreamingAgentChatResponse:
        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(
            method="stream_chat", is_async=False, with_record=False
        )

        res, _ = self.stream_chat_with_record(*args, **kwargs)
        return res

    # TODEP
    # llama_index.chat_engine.types.BaseChatEngine
    async def astream_chat(self, *args, **kwargs) -> StreamingAgentChatResponse:
        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(
            method="astream_chat", is_async=True, with_record=False
        )

        res, _ = await self.astream_chat_with_record(*args, **kwargs)
        return res

    # TODEP
    # llama_index.indices.query.base.BaseQueryEngine
    def query(self, *args, **kwargs) -> RESPONSE_TYPE:

        assert isinstance(
            self.app, llama_index.indices.query.base.BaseQueryEngine
        )

        self._with_dep_message(
            method="query", is_async=False, with_record=False
        )

        res, _ = self.query_with_record(*args, **kwargs)
        return res

    # TODEP
    # llama_index.indices.query.base.BaseQueryEngine
    async def aquery(self, *args, **kwargs) -> RESPONSE_TYPE:

        assert isinstance(
            self.app, llama_index.indices.query.base.BaseQueryEngine
        )

        self._with_dep_message(
            method="aquery", is_async=True, with_record=False
        )

        res, _ = await self.aquery_with_record(*args, **kwargs)
        return res

    # TODEP
    # Mirrors llama_index.indices.query.base.BaseQueryEngine.query .
    def query_with_record(self, *args,
                          **kwargs) -> Tuple[RESPONSE_TYPE, Record]:

        assert isinstance(
            self.app, llama_index.indices.query.base.BaseQueryEngine
        )

        self._with_dep_message(method="query", is_async=False, with_record=True)

        return self.with_record(self.app.query, *args, **kwargs)

    # TODEP
    # Mirrors llama_index.indices.query.base.BaseQueryEngine.aquery .
    async def aquery_with_record(self, *args,
                                 **kwargs) -> Tuple[RESPONSE_TYPE, Record]:
        assert isinstance(
            self.app, llama_index.indices.query.base.BaseQueryEngine
        )

        self._with_dep_message(method="aquery", is_async=True, with_record=True)

        return await self.awith_record(self.app.aquery, *args, **kwargs)

    # TODEP
    # Compatible with llama_index.chat_engine.types.BaseChatEngine.chat .
    def chat_with_record(self, *args,
                         **kwargs) -> Tuple[AgentChatResponse, Record]:

        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(method="chat", is_async=False, with_record=True)

        return self.with_record(self.app.chat, *args, **kwargs)

    # TODEP
    # Compatible with llama_index.chat_engine.types.BaseChatEngine.achat .
    async def achat_with_record(self, *args,
                                **kwargs) -> Tuple[AgentChatResponse, Record]:
        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(method="achat", is_async=True, with_record=True)

        return await self.awith_record(self.app.achat, *args, **kwargs)

    # TODEP
    # Compatible with llama_index.chat_engine.types.BaseChatEngine.stream_chat .
    def stream_chat_with_record(
        self, *args, **kwargs
    ) -> Tuple[StreamingAgentChatResponse, Record]:

        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(
            method="stream", is_async=False, with_record=True
        )

        return self.with_record(self.app.stream_chat, *args, **kwargs)

    # TODEP
    # Compatible with llama_index.chat_engine.types.BaseChatEngine.astream_chat .
    async def astream_chat_with_record(
        self, *args, **kwargs
    ) -> Tuple[StreamingAgentChatResponse, Record]:

        assert isinstance(
            self.app, llama_index.chat_engine.types.BaseChatEngine
        )

        self._with_dep_message(
            method="astream_chat", is_async=True, with_record=True
        )

        return await self.awith_record(self.app.astream_chat, *args, **kwargs)

main_input(func, sig, bindings)

Determine the main input string for the given function func with signature sig if it is to be called with the given bindings bindings.

Source code in trulens_eval/trulens_eval/tru_llama.py

def main_input(
    self, func: Callable, sig: Signature, bindings: BoundArguments
) -> str:
    """
    Determine the main input string for the given function `func` with
    signature `sig` if it is to be called with the given bindings
    `bindings`.
    """

    if 'str_or_query_bundle' in bindings.arguments:
        # llama_index specific
        return bindings.arguments['str_or_query_bundle']

    elif 'message' in bindings.arguments:
        # llama_index specific
        return bindings.arguments['message']

    else:

        return App.main_input(self, func, sig, bindings)

main_output(func, sig, bindings, ret)

Determine the main out string for the given function func with signature sig after it is called with the given bindings and has returned ret.

Source code in trulens_eval/trulens_eval/tru_llama.py

def main_output(
    self, func: Callable, sig: Signature, bindings: BoundArguments, ret: Any
) -> str:
    """
    Determine the main out string for the given function `func` with
    signature `sig` after it is called with the given `bindings` and has
    returned `ret`.
    """

    if isinstance(ret, Response):  # query, aquery
        return ret.response

    elif isinstance(ret, AgentChatResponse):  #  chat, achat
        return ret.response

    elif isinstance(ret, (StreamingResponse, StreamingAgentChatResponse)):
        logger.warning(
            "App produced a streaming response. "
            "Tracking content of streams in llama_index is not yet supported. "
            "App main_output will be None."
        )

        return None

    else:

        return App.main_output(self, func, sig, bindings, ret)

select_source_nodes() classmethod

Get the path to the source nodes in the query output.

Source code in trulens_eval/trulens_eval/tru_llama.py

@classmethod
def select_source_nodes(cls) -> JSONPath:
    """
    Get the path to the source nodes in the query output.
    """
    return cls.select_outputs().source_nodes[:]