LLMアプリの評価・分析ツール「Context.ai」を試す

!pip install --upgrade context-python

from google.colab import userdata
import os

os.environ["GETCONTEXT_TOKEN"] = userdata.get('GETCONTEXT_TOKEN')

import getcontext
from getcontext.generated.models import Conversation, Message, MessageRole, Rating
from getcontext.token import Credential
import os

token = os.environ.get("GETCONTEXT_TOKEN")

c = getcontext.ContextAPI(credential=Credential(token))

c.log.conversation(
    body={
        "conversation": Conversation(
            messages=[
                Message(
                    message="あなたは親切な日本語のアシスタントです。",
                    role=MessageRole.SYSTEM,
                ),
                Message(
                    message="こんにちは！",
                    role=MessageRole.USER,
                ),
                Message(
                    message="こんにちは！今日はどんなお手伝いをしましょうか？",
                    role=MessageRole.ASSISTANT,
                    rating=Rating.POSITIVE,
                ),
            ],
            metadata={
                "model": "gpt-3.5-turbo",
                "user_id": "1234",
                "environment": "test",
            }
        )
    }
)

c.log.conversation_upsert(
    body={
        "conversation": Conversation(
            messages=[
                Message(
                    message="あなたは親切な日本語のアシスタントです。",
                    role=MessageRole.SYSTEM,
                ),
                Message(
                    message="こんにちは！",
                    role=MessageRole.USER,
                ),
                Message(
                    message="こんにちは！今日はどんなお手伝いをしましょうか？",
                    role=MessageRole.ASSISTANT,
                    rating=Rating.POSITIVE,
                ),
                Message(
                    message="今日は最高の1日だったんだよね！聞いてくれる？",
                    role=MessageRole.USER,
                    rating=Rating.POSITIVE,
                ),
            ],
            metadata={
                "model": "gpt-3.5-turbo",
                "user_id": "1234",
                "environment": "test",
            }
        )
    }
)

c.log.conversation(
    body={
        "conversation": Conversation(
            messages=[
                Message(
                    message="あなたは親切な日本語のアシスタントです。",
                    role=MessageRole.SYSTEM,
                ),
                Message(
                    message="ちょっと今困ってることがあって、相談に乗ってくれる・・・？",
                    role=MessageRole.USER,
                ),
                Message(
                    message="それは大変ですね・・・お力になれるように努力しますので何でも聞いて下さい。どんなことでこまっているのですか？",
                    role=MessageRole.ASSISTANT,
                    rating=Rating.POSITIVE,
                ),
            ],
            metadata={
                "model": "gpt-4o",
                "user_id": "5678",
                "environment": "test",
            }
        )
    }
)

!pip install --upgrade --quiet langchain-core langchain-openai langchain_community context-python
!pip freeze | egrep -i "langchain|context-python"

context-python==0.18.0
langchain==0.2.6
langchain-community==0.2.6
langchain-core==0.2.11
langchain-openai==0.1.14
langchain-text-splitters==0.2.2

import os
from google.colab import userdata

os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY')
os.environ['GETCONTEXT_TOKEN'] = userdata.get('GETCONTEXT_TOKEN')

# ContextCallbackHandlerを継承したContextCallbackHandlerを定義
class CustomContextCallbackHandler(ContextCallbackHandler):
    # コンストラクタをオーバーライド
    def __init__(self, token: str = "", verbose: bool = False, **kwargs: Any) -> None:
        # kwargsでuser_idを含むmetadataを渡せるようにする
        metadata = kwargs.get('metadata')
        super().__init__(token=token, verbose=verbose, **kwargs)
        if metadata and isinstance(metadata, dict):
            self.metadata.update(metadata)

    # 会話履歴をContext.aiに送信する関数をオーバーライド
    def _log_conversation(self) -> None:
        """Log the conversation to the context API."""
        if len(self.messages) == 0:
            return

        self.client.log.conversation_upsert(
            body={
                "conversation": self.conversation_model(
                    messages=self.messages,
                    metadata=self.metadata,
                )
            }
        )

        # Upsertだと全部の会話履歴を上書きするはず、と思っていたけど、
        # どうも重複して追加されてしまうので、ここで初期化（元のコードのまま）
        self.messages = []
        # ここでmetadataを消すことでuser_idが消えてしまうのでコメントアウト
        #self.metadata = {}

# 会話履歴を取り出す関数
def get_session_history(user_id: str) -> BaseChatMessageHistory:
    if user_id not in store:
        store[user_id] = ChatMessageHistory()
    return store[user_id]

# 会話メモリを初期化
store = {}

# ユーザIDを定義
user_id = "abc1234"

# CustomContextCallbackHandlerを初期化、ここでContext.aiに送るmetadata（ユーザID）を渡す
token = os.environ.get("GETCONTEXT_TOKEN")
context_callback = CustomContextCallbackHandler(token, metadata={"user_id": user_id})

# モデルの定義、ここでcallbackも定義する
model = ChatOpenAI(model="gpt-3.5-turbo-0125", callbacks=[context_callback])

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "あなたは大阪のおばちゃんです。20字以内で簡潔、かつ大阪弁で元気に明るく回答します。",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

runnable = prompt | model | StrOutputParser()

with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
    history_factory_config=[
        ConfigurableFieldSpec(
            id="user_id",
            annotation=str,
            name="User ID",
            description="Unique identifier for the user.",
            default="",
            is_shared=True,
        ),
    ],
)

with_message_history.invoke(
    {"input": "おばちゃん、元気？"},
    config={"configurable": {"user_id": user_id}},
)

with_message_history.invoke(
    {"input": "そうなんや、こっちは競馬に負けて最低やわ・・・"},
    config={"configurable": {"user_id": user_id}},
)

with_message_history.invoke(
    {"input": "そやなー、競馬は明日もあるし、頑張って取り戻すわ！おばちゃん、いつもありがとうな！"},
    config={"configurable": {"user_id": user_id}},
)

import getcontext
from getcontext.generated.models import Conversation, Message, MessageRole, Rating, Thread
from getcontext.token import Credential
import os
from google.colab import userdata

os.environ["GETCONTEXT_TOKEN"] = userdata.get('GETCONTEXT_TOKEN')

token = os.environ.get("GETCONTEXT_TOKEN")

c = getcontext.ContextAPI(credential=Credential(token))

res = c.log.conversation_thread(
    body={
        "conversation": Conversation(
            messages=[
                Message(
                    message="あなたは親切な日本語のアシスタントです。",
                    role=MessageRole.SYSTEM,
                ),
                Message(
                    message="こんにちは！",
                    role=MessageRole.USER,
                ),
                Message(
                    message="こんにちは！今日はどんなお手伝いをしましょうか？",
                    role=MessageRole.ASSISTANT,
                ),
            ],
            metadata={
                "model": "gpt-3.5-turbo",
                "user_id": "abcd1234",
                "environment": "test",
            }
        )
    }
)

thread_id = res.data.id
thread_id

c_6lenhmg68l1y19p

c.log.conversation_thread(
    body={
        "conversation": Thread(
            id=res.data.id,
            messages=[
                Message(
                    message="明日の天気を教えて。",
                    role=MessageRole.USER,
                ),
                Message(
                    message="明日の東京都の天気は晴れ、1日中雨の心配はいりません。お出かけ日和ですね！",
                    role=MessageRole.ASSISTANT,
                    rating=Rating.POSITIVE,
                ),
            ],
            metadata={
                "model": "gpt-3.5-turbo",
                "user_id": "abcd1234",
                "environment": "test",
            }
        )
    }
)

from google.colab import userdata
import getcontext
from getcontext.generated.models import Conversation, Thread
from getcontext.token import Credential
from openai import OpenAI

openai_client = OpenAI(api_key=userdata.get('OPENAI_API_KEY'))
context_client = getcontext.ContextAPI(credential=Credential(userdata.get('GETCONTEXT_TOKEN')))

metadata={
    "model": "gpt-3.5-turbo",
    "user_id": "hogehoge",
    "environment": "test",
}


def get_response(messages):
    try:
        response = openai_client.chat.completions.create(
            model=metadata["model"],
            messages=messages
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"エラーが発生しました: {str(e)}"


def convert_messages_for_context(messages: list[dict]) -> list[dict]:
    """OpenAIのメッセージフォーマットをContextのメッセージフォーマットに変換する"""
    return [
        {"message" if k == "content" else k: v for k, v in message.items()}
        for message in messages
    ]


def chat():
    print("チャットアプリへようこそ！")
    print("終了するには 'quit' と入力してください。")

    messages = [
        {"role": "system", "content": "あなたは親切なアシスタントです。"}
    ]

    # スレッドIDを初期化 
    thread_id = None

    while True:
        user_input = input("User: ")
        if user_input.lower() == 'quit':
            print("Assistant: チャットを終了します。さようなら！")
            break
        
        messages.append({"role": "user", "content": user_input})
        
        gpt_response = get_response(messages)
        print(f"Assistant: {gpt_response}")
        
        messages.append({"role": "assistant", "content": gpt_response})

        # スレッドIDがなければスレッドIDを取得、あればそのスレッドIDを使う
        if thread_id == None:
            res = context_client.log.conversation_thread(
               body={
                    "conversation": Conversation(
                        messages=convert_messages_for_context(messages),
                        metadata=metadata,
                    )
                }
            )
            thread_id = res.data.id
        else:
            res = context_client.log.conversation_thread(
                body={
                    "conversation": Thread(
                        id=thread_id,
                        messages=convert_messages_for_context(messages[-2:]),
                        metadata=metadata,
                    )
                }
            )

if __name__ == "__main__":
    chat()