TruLens-Canopy Quickstart¶

Canopy is an open-source framework and context engine built on top of the Pinecone vector database so you can build and host your own production-ready chat assistant at any scale. By integrating TruLens into your Canopy assistant, you can quickly iterate on and gain confidence in the quality of your chat assistant.

In [ ]:

# !pip install trulens trulens-providers-openai canopy-sdk cohere ipywidgets tqdm

# !pip install trulens trulens-providers-openai canopy-sdk cohere ipywidgets tqdm

In [ ]:

import numpy

assert (
    numpy.__version__ >= "1.26"
), "Numpy version did not updated, if you are working on Colab please restart the session."

import numpy assert ( numpy.__version__ >= "1.26" ), "Numpy version did not updated, if you are working on Colab please restart the session."

Set Keys¶

In [ ]:

import os

os.environ["PINECONE_API_KEY"] = (
    "YOUR_PINECONE_API_KEY"  # take free trial key from https://app.pinecone.io/
)
os.environ["OPENAI_API_KEY"] = (
    "YOUR_OPENAI_API_KEY"  # take free trial key from https://platform.openai.com/api-keys
)
os.environ["CO_API_KEY"] = (
    "YOUR_COHERE_API_KEY"  # take free trial key from https://dashboard.cohere.com/api-keys
)

import os os.environ["PINECONE_API_KEY"] = ( "YOUR_PINECONE_API_KEY" # take free trial key from https://app.pinecone.io/ ) os.environ["OPENAI_API_KEY"] = ( "YOUR_OPENAI_API_KEY" # take free trial key from https://platform.openai.com/api-keys ) os.environ["CO_API_KEY"] = ( "YOUR_COHERE_API_KEY" # take free trial key from https://dashboard.cohere.com/api-keys )

In [ ]:

assert (
    os.environ["PINECONE_API_KEY"] != "YOUR_PINECONE_API_KEY"
), "please provide PINECONE API key"
assert (
    os.environ["OPENAI_API_KEY"] != "YOUR_OPENAI_API_KEY"
), "please provide OpenAI API key"
assert (
    os.environ["CO_API_KEY"] != "YOUR_COHERE_API_KEY"
), "please provide Cohere API key"

assert ( os.environ["PINECONE_API_KEY"] != "YOUR_PINECONE_API_KEY" ), "please provide PINECONE API key" assert ( os.environ["OPENAI_API_KEY"] != "YOUR_OPENAI_API_KEY" ), "please provide OpenAI API key" assert ( os.environ["CO_API_KEY"] != "YOUR_COHERE_API_KEY" ), "please provide Cohere API key"

In [ ]:

from pinecone import PodSpec

# Defines the cloud and region where the index should be deployed
# Read more about it here - https://docs.pinecone.io/docs/create-an-index
spec = PodSpec(environment="gcp-starter")

from pinecone import PodSpec # Defines the cloud and region where the index should be deployed # Read more about it here - https://docs.pinecone.io/docs/create-an-index spec = PodSpec(environment="gcp-starter")

Load data¶

Downloading Pinecone's documentation as data to ingest to our Canopy chatbot:

In [ ]:

import warnings

import pandas as pd

warnings.filterwarnings("ignore")

data = pd.read_parquet(
    "https://storage.googleapis.com/pinecone-datasets-dev/pinecone_docs_ada-002/raw/file1.parquet"
)
data.head()

import warnings import pandas as pd warnings.filterwarnings("ignore") data = pd.read_parquet( "https://storage.googleapis.com/pinecone-datasets-dev/pinecone_docs_ada-002/raw/file1.parquet" ) data.head()

In [ ]:

print(
    data["text"][50][:847]
    .replace("\n\n", "\n")
    .replace("[Suggest Edits](/edit/limits)", "")
    + "\n......"
)
print("source: ", data["source"][50])

print( data["text"][50][:847] .replace("\n\n", "\n") .replace("[Suggest Edits](/edit/limits)", "") + "\n......" ) print("source: ", data["source"][50])

Setup Tokenizer¶

In [ ]:

from canopy.tokenizer import Tokenizer

Tokenizer.initialize()

tokenizer = Tokenizer()

tokenizer.tokenize("Hello world!")

from canopy.tokenizer import Tokenizer Tokenizer.initialize() tokenizer = Tokenizer() tokenizer.tokenize("Hello world!")

Create and Load Index¶

In [ ]:

from canopy.knowledge_base import KnowledgeBase
from canopy.knowledge_base import list_canopy_indexes
from canopy.models.data_models import Document
from tqdm.auto import tqdm

index_name = "pinecone-docs"

kb = KnowledgeBase(index_name)

if not any(name.endswith(index_name) for name in list_canopy_indexes()):
    kb.create_canopy_index(spec=spec)

kb.connect()

documents = [Document(**row) for _, row in data.iterrows()]

batch_size = 100

for i in tqdm(range(0, len(documents), batch_size)):
    kb.upsert(documents[i : i + batch_size])

from canopy.knowledge_base import KnowledgeBase from canopy.knowledge_base import list_canopy_indexes from canopy.models.data_models import Document from tqdm.auto import tqdm index_name = "pinecone-docs" kb = KnowledgeBase(index_name) if not any(name.endswith(index_name) for name in list_canopy_indexes()): kb.create_canopy_index(spec=spec) kb.connect() documents = [Document(**row) for _, row in data.iterrows()] batch_size = 100 for i in tqdm(range(0, len(documents), batch_size)): kb.upsert(documents[i : i + batch_size])

Create context and chat engine¶

In [ ]:

from canopy.chat_engine import ChatEngine
from canopy.context_engine import ContextEngine

context_engine = ContextEngine(kb)


chat_engine = ChatEngine(context_engine)

from canopy.chat_engine import ChatEngine from canopy.context_engine import ContextEngine context_engine = ContextEngine(kb) chat_engine = ChatEngine(context_engine)

API for chat is exactly the same as for OpenAI:

In [ ]:

from canopy.models.data_models import UserMessage

chat_history = [
    UserMessage(
        content="What is the the maximum top-k for a query to Pinecone?"
    )
]

chat_engine.chat(chat_history).choices[0].message.content

from canopy.models.data_models import UserMessage chat_history = [ UserMessage( content="What is the the maximum top-k for a query to Pinecone?" ) ] chat_engine.chat(chat_history).choices[0].message.content

Instrument static methods used by engine with TruLens¶

In [ ]:

warnings.filterwarnings("ignore")

warnings.filterwarnings("ignore")

In [ ]:

from canopy.chat_engine import ChatEngine
from canopy.context_engine import ContextEngine
from trulens.apps.custom import instrument

instrument.method(ContextEngine, "query")

instrument.method(ChatEngine, "chat")

from canopy.chat_engine import ChatEngine from canopy.context_engine import ContextEngine from trulens.apps.custom import instrument instrument.method(ContextEngine, "query") instrument.method(ChatEngine, "chat")

Create feedback functions using instrumented methods¶

In [ ]:

from trulens.core import TruSession

session = TruSession(database_redact_keys=True)

from trulens.core import TruSession session = TruSession(database_redact_keys=True)

In [ ]:

import numpy as np
from trulens.core import Feedback
from trulens.core import Select
from trulens.providers.openai import OpenAI as fOpenAI

# Initialize provider class
provider = fOpenAI()

grounded = Groundedness(groundedness_provider=provider)

prompt = Select.RecordCalls.chat.args.messages[0].content
context = (
    Select.RecordCalls.context_engine.query.rets.content.root[:]
    .snippets[:]
    .text
)
output = Select.RecordCalls.chat.rets.choices[0].message.content

# Define a groundedness feedback function
f_groundedness = (
    Feedback(
        provider.groundedness_measure_with_cot_reasons,
        name="Groundedness",
        higher_is_better=True,
    )
    .on(context.collect())
    .on(output)
)

# Question/answer relevance between overall question and answer.
f_qa_relevance = (
    Feedback(
        provider.relevance_with_cot_reasons,
        name="Answer Relevance",
        higher_is_better=True,
    )
    .on(prompt)
    .on(output)
)

# Question/statement relevance between question and each context chunk.
f_context_relevance = (
    Feedback(
        provider.context_relevance_with_cot_reasons,
        name="Context Relevance",
        higher_is_better=True,
    )
    .on(prompt)
    .on(context)
    .aggregate(np.mean)
)

import numpy as np from trulens.core import Feedback from trulens.core import Select from trulens.providers.openai import OpenAI as fOpenAI # Initialize provider class provider = fOpenAI() grounded = Groundedness(groundedness_provider=provider) prompt = Select.RecordCalls.chat.args.messages[0].content context = ( Select.RecordCalls.context_engine.query.rets.content.root[:] .snippets[:] .text ) output = Select.RecordCalls.chat.rets.choices[0].message.content # Define a groundedness feedback function f_groundedness = ( Feedback( provider.groundedness_measure_with_cot_reasons, name="Groundedness", higher_is_better=True, ) .on(context.collect()) .on(output) ) # Question/answer relevance between overall question and answer. f_qa_relevance = ( Feedback( provider.relevance_with_cot_reasons, name="Answer Relevance", higher_is_better=True, ) .on(prompt) .on(output) ) # Question/statement relevance between question and each context chunk. f_context_relevance = ( Feedback( provider.context_relevance_with_cot_reasons, name="Context Relevance", higher_is_better=True, ) .on(prompt) .on(context) .aggregate(np.mean) )

Create recorded app and run it¶

In [ ]:

from trulens.apps.custom import TruCustomApp

app_name = "canopy default"
tru_recorder = TruCustomApp(
    chat_engine,
    app_name=app_name,
    feedbacks=[f_groundedness, f_qa_relevance, f_context_relevance],
)

from trulens.apps.custom import TruCustomApp app_name = "canopy default" tru_recorder = TruCustomApp( chat_engine, app_name=app_name, feedbacks=[f_groundedness, f_qa_relevance, f_context_relevance], )

In [ ]:

from canopy.models.data_models import UserMessage

queries = [
    [
        UserMessage(
            content="What is the maximum dimension for a dense vector in Pinecone?"
        )
    ],
    [UserMessage(content="How can you get started with Pinecone and TruLens?")],
    [
        UserMessage(
            content="What is the the maximum top-k for a query to Pinecone?"
        )
    ],
]

answers = []

for query in queries:
    with tru_recorder as recording:
        response = chat_engine.chat(query)
        answers.append(response.choices[0].message.content)

from canopy.models.data_models import UserMessage queries = [ [ UserMessage( content="What is the maximum dimension for a dense vector in Pinecone?" ) ], [UserMessage(content="How can you get started with Pinecone and TruLens?")], [ UserMessage( content="What is the the maximum top-k for a query to Pinecone?" ) ], ] answers = [] for query in queries: with tru_recorder as recording: response = chat_engine.chat(query) answers.append(response.choices[0].message.content)

As you can see, we got the wrong answer, the limits for sparse vectors instead of dense vectors:

In [ ]:

print(queries[0][0].content + "\n")
print(answers[0])

print(queries[0][0].content + "\n") print(answers[0])

In [ ]:

session.get_leaderboard(app_ids=[tru_recorder.app_id])

session.get_leaderboard(app_ids=[tru_recorder.app_id])

Run Canopy with Cohere reranker¶

In [ ]:

from canopy.knowledge_base.reranker.cohere import CohereReranker

kb = KnowledgeBase(
    index_name=index_name, reranker=CohereReranker(top_n=3), default_top_k=30
)
kb.connect()

reranker_chat_engine = ChatEngine(ContextEngine(kb))

from canopy.knowledge_base.reranker.cohere import CohereReranker kb = KnowledgeBase( index_name=index_name, reranker=CohereReranker(top_n=3), default_top_k=30 ) kb.connect() reranker_chat_engine = ChatEngine(ContextEngine(kb))

In [ ]:

reranking_app_name = "canopy_reranking"
reranking_tru_recorder = TruCustomApp(
    reranker_chat_engine,
    app_name=reranking_app_name,
    feedbacks=[f_groundedness, f_qa_relevance, f_context_relevance],
)

answers = []

for query in queries:
    with reranking_tru_recorder as recording:
        answers.append(
            reranker_chat_engine.chat(query).choices[0].message.content
        )

reranking_app_name = "canopy_reranking" reranking_tru_recorder = TruCustomApp( reranker_chat_engine, app_name=reranking_app_name, feedbacks=[f_groundedness, f_qa_relevance, f_context_relevance], ) answers = [] for query in queries: with reranking_tru_recorder as recording: answers.append( reranker_chat_engine.chat(query).choices[0].message.content )

With reranking we get the right answer!

In [ ]:

print(queries[0][0].content + "\n")
print(answers[0])

print(queries[0][0].content + "\n") print(answers[0])

Evaluate the effect of reranking¶

In [ ]:

session.get_leaderboard(app_ids=[tru_recorder.app_id, reranking_tru_recorder.app_id])

session.get_leaderboard(app_ids=[tru_recorder.app_id, reranking_tru_recorder.app_id])

Explore more in the TruLens dashboard¶

In [ ]:

from trulens.dashboard import run_dashboard

run_dashboard(session)

# stop_dashboard(session) # stop if needed

from trulens.dashboard import run_dashboard run_dashboard(session) # stop_dashboard(session) # stop if needed