LangChain

Weave を使用して、LangChain Python ライブラリを介して行われるすべての呼び出しをトラッキングしてログします

Weave は、LangChain Python ライブラリを介して行われるすべての呼び出しを、簡単にトラッキングしてログできるように設計されています。

LLM を扱う際、デバッグは避けられません。モデルの呼び出しが失敗したり、出力の形式が崩れたり、ネストされたモデル呼び出しが混乱を招いたりすると、問題の特定が難しくなることがあります。LangChain アプリケーションは多くの場合、複数の step と LLM 呼び出しで構成されるため、チェーンやエージェントの内部動作を理解することが重要です。

Weave は、LangChain アプリケーションのトレースを自動的に取得することで、このプロセスを簡素化します。これにより、アプリケーションのパフォーマンスを監視および分析でき、LLM ワークフローのデバッグと最適化が容易になります。

​

はじめに

まず、スクリプトの先頭で weave.init() を呼び出します。weave.init() の引数にはプロジェクト名を指定します。これは、トレースを整理するのに役立ちます。

import weave
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# project 名を指定してWeaveを初期化する
weave.init("langchain_demo")

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

output = llm_chain.invoke({"number": 2})

print(output)

​

Call メタデータのトラッキング

LangChain の Call のメタデータをトラッキングするには、weave.attributes コンテキストマネージャーを使用できます。このコンテキストマネージャーを使用すると、チェーンや単一のリクエストなど、特定のコードブロックにカスタムメタデータを設定できます。

import weave
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# project 名で Weave を初期化する
weave.init("langchain_demo")

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

with weave.attributes({"my_awesome_attribute": "value"}):
    output = llm_chain.invoke()

print(output)

Weave は、LangChain Call のトレースに関連付けてメタデータを自動的にトラッキングします。メタデータは、以下に示すように Weave の Web インターフェースで確認できます。

​

トレース

LLM アプリケーションのトレースを一元的なデータベースに保存することは、開発時にも本番環境でも非常に重要です。これらのトレースは、有用なデータセットとして機能し、アプリケーションのデバッグや改善に欠かせません。

Weave は、LangChain アプリケーションのトレースを自動的に取得します。LangChain ライブラリを介して行われるすべての呼び出しをトラッキングしてログし、これには prompt templates、chains、LLM calls、tools、agent steps が含まれます。トレースは Weave の Web インターフェースで確認できます。

​

calls を手動でトレースする

自動トレースに加えて、WeaveTracer コールバックまたは weave_tracing_enabled コンテキストマネージャーを使用して、calls を手動でトレースできます。これらの方法は、LangChain アプリケーションの個々の部分でリクエストコールバックを使用する場合に似ています。

注: Weave はデフォルトで LangChain の Runnables をトレースし、これは weave.init() を呼び出すと有効になります。weave.init() を呼び出す前に、環境変数 WEAVE_TRACE_LANGCHAIN を "false" に設定することで、この動作を無効にできます。これにより、アプリケーション内の特定の chain や個々のリクエストに対するトレース動作を制御できます。

​

WeaveTracer を使用する

WeaveTracer コールバックを個々の LangChain コンポーネントに渡すことで、特定のリクエストをトレースできます。

import os

os.environ["WEAVE_TRACE_LANGCHAIN"] = "false" # <- グローバルトレースを明示的に無効にする。

from weave.integrations.langchain import WeaveTracer
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import weave

# project名でWeaveを初期化する
weave.init("langchain_demo")  # <-- 環境変数が明示的に `false` に設定されているため、ここではトレースを有効にしない

weave_tracer = WeaveTracer()

config = {"callbacks": [weave_tracer]}

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

output = llm_chain.invoke({"number": 2}, config=config) # <-- このチェーンの呼び出しに対してのみトレースを有効にする。

llm_chain.invoke({"number": 4})  # <-- LangChainのCallではトレースが有効にならないが、OpenAIのCallは引き続きトレースされる

​

コンテキストマネージャー weave_tracing_enabled を使用する

または、weave_tracing_enabled コンテキストマネージャーを使用して、特定のコードブロックでトレースを有効にできます。

import os

os.environ["WEAVE_TRACE_LANGCHAIN"] = "false" # <- グローバルトレースを明示的に無効にする。

from weave.integrations.langchain import weave_tracing_enabled
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import weave

# project 名を指定して Weave を初期化する
weave.init("langchain_demo")  # <-- 環境変数が明示的に `false` に設定されているため、ここではトレースを有効にしない

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

with weave_tracing_enabled():  # <-- このチェーンの呼び出しに対してのみトレースを有効にする。
    output = llm_chain.invoke({"number": 2})


llm_chain.invoke({"number": 4})  # <-- langchain の Call ではトレースが有効にならないが、openai の Call は引き続きトレースされる

​

設定

weave.init() を呼び出すと、環境変数 WEAVE_TRACE_LANGCHAIN を "true" に設定することでトレースが有効になります。これにより、Weave は LangChain アプリケーションのトレースを自動的に取得します。この動作を無効にするには、環境変数を "false" に設定します。

​

LangChain Callbacks との関連

​

自動ロギング

weave.init() が提供する自動ロギングは、LangChain アプリケーション内のすべてのコンポーネントにコンストラクタのコールバックを渡すのに似ています。つまり、プロンプトテンプレート、チェーン、LLM calls、ツール、agent steps を含むすべてのやり取りが、アプリケーション全体でグローバルにトラッキングされます。

​

手動ログ記録

手動ログ記録の method (WeaveTracer と weave_tracing_enabled) は、LangChain アプリケーションの個々の部分でリクエストコールバックを使用するのと似ています。これらの method を使うと、アプリケーションのどの部分をトレースするかをより細かく制御できます。

• コンストラクターコールバック: チェーンまたはコンポーネント全体に適用され、すべてのやり取りを一貫してログ記録します。
• リクエストコールバック: 特定のリクエストに適用され、個別の呼び出しを詳細にトレースできます。

Weave を LangChain と統合することで、LLM アプリケーション全体のログ記録とモニタリングを確実に行えるようになり、デバッグやパフォーマンスの最適化が容易になります。

詳しくは、LangChain documentationを参照してください。

​

Modelsと評価

さまざまなユースケースのアプリケーションで LLM を整理して評価するのは、プロンプト、モデルの設定、推論パラメーターなど複数のコンポーネントが関わるため、簡単ではありません。weave.Model を使用すると、system prompt や使用するモデルといった実験の詳細を記録して整理できるため、異なるイテレーションを比較しやすくなります。

以下の例では、LangChain の チェーン を WeaveModel でラップする方法を示します。

import json
import asyncio

import weave

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# project 名でWeaveを初期化する
weave.init("langchain_demo")

class ExtractFruitsModel(weave.Model):
    model_name: str
    prompt_template: str

    @weave.op()
    async def predict(self, sentence: str) -> dict:
        llm = ChatOpenAI(model=self.model_name, temperature=0.0)
        prompt = PromptTemplate.from_template(self.prompt_template)

        llm_chain = prompt | llm
        response = llm_chain.invoke({"sentence": sentence})
        result = response.content

        if result is None:
            raise ValueError("No response from model")
        parsed = json.loads(result)
        return parsed

model = ExtractFruitsModel(
    model_name="gpt-3.5-turbo-1106",
    prompt_template='Extract fields ("fruit": <str>, "color": <str>, "flavor": <str>) from the following text, as json: {sentence}',
)
sentence = "There are many fruits that were found on the recently discovered planet Goocrux. There are neoskizzles that grow there, which are purple and taste like candy."

prediction = asyncio.run(model.predict(sentence))

# Jupyter Notebookを使用している場合は、次を実行してください:
# prediction = await model.predict(sentence)

print(prediction)

このコードは、Weave UI で可視化できるモデルを作成します。

Weave Models は、serve や 評価 でも使用できます。

​

評価

評価を使うと、モデルのパフォーマンスを測定できます。weave.Evaluation クラスを使うことで、特定のタスクやデータセットに対してモデルがどの程度うまく機能するかを把握できるため、異なるモデルやアプリケーションの各イテレーションを比較しやすくなります。以下の例では、作成したモデルを評価する方法を示します。

from weave.scorers import MultiTaskBinaryClassificationF1

sentences = [
    "There are many fruits that were found on the recently discovered planet Goocrux. There are neoskizzles that grow there, which are purple and taste like candy.",
    "Pounits are a bright green color and are more savory than sweet.",
    "Finally, there are fruits called glowls, which have a very sour and bitter taste which is acidic and caustic, and a pale orange tinge to them.",
]
labels = [
    {"fruit": "neoskizzles", "color": "purple", "flavor": "candy"},
    {"fruit": "pounits", "color": "bright green", "flavor": "savory"},
    {"fruit": "glowls", "color": "pale orange", "flavor": "sour and bitter"},
]
examples = [
    {"id": "0", "sentence": sentences[0], "target": labels[0]},
    {"id": "1", "sentence": sentences[1], "target": labels[1]},
    {"id": "2", "sentence": sentences[2], "target": labels[2]},
]

@weave.op()
def fruit_name_score(target: dict, output: dict) -> dict:
    return {"correct": target["fruit"] == output["fruit"]}


evaluation = weave.Evaluation(
    dataset=examples,
    scorers=[
        MultiTaskBinaryClassificationF1(class_names=["fruit", "color", "flavor"]),
        fruit_name_score,
    ],
)
scores = asyncio.run(evaluation.evaluate(model)))
# Jupyter Notebook で実行する場合:
# scores = await evaluation.evaluate(model)

print(scores)

このコードは、Weave UIで可視化できる評価トレースを生成します。

WeaveをLangchainと連携することで、LLMアプリケーションのログ収集と監視を包括的に行え、デバッグやパフォーマンスの最適化を容易にできます。

​

既知の問題

• 非同期 call のトレース - Langchain の AsyncCallbackManager 実装のバグにより、非同期 call が正しい順序でトレースされません。この問題を修正するための PR を提出しています。したがって、Langchain Runnables で ainvoke、astream、abatch method を使用すると、trace 内の call の順序が正確でない可能性があります。