Zenn
Closed10

軽量LLMフレームワーク「kani」を試す

#
framework
#
LLM
#
kani
kun432kun432

LangChainやLlamaIndex以外で、小さめのフレームワークを少し試してみる。

https://note.com/hamachi_jp/n/n342becc4f345

https://github.com/zhudotexe/kani

kani (カニ) はチャットベースの言語モデルのための軽量でハックしやすいフレームワークです。

他のLMフレームワークと比較して、kaniはクセが少なく、制御フローの重要な部分をより細かくカスタマイズすることができます。

kaniは以下のモデルをサポートしており、モデルに依存しないフレームワークにより、さらに多くのモデルのサポートを追加することができます：

  • OpenAIモデル（GPT-3.5-turbo、GPT-4、GPT-4-turbo）
  • Anthropicモデル (Claude, Claude Instant)
  • LLaMA v2 (Hugging Faceまたはctransformers経由) &ファインチューニング
    0 Vicuna v1.3 (Hugging Face 経由) & ファインチューニング

特徴

  • 軽量かつ高水準
    • kaniは言語モデルとのインターフェイスのための一般的な定型文を実装しており、意見集約型のプロンプトフレームワークや複雑なライブラリ固有のツールを使用する必要はありません。
  • モデルにとらわれない
    • kaniはトークンカウントと補完生成を実装するためのシンプルなインターフェースを提供します。この2つを実装すれば、kaniはどの言語モデルでも動作します。
  • 自動チャットメモリ管理
    • 履歴のトークン数の管理を気にすることなく、チャットセッションを流すことができます。
  • モデルフィードバックとリトライを備えたFunction Calling
    • 1行のコードでモデルに関数へのアクセスを提供します。kaniは幻覚パラメータやエラーに関するフィードバックをエレガントに提供し、モデルが呼び出しをリトライできるようにします。
  • プロンプトをコントロール可能
    • 隠されたプロンプトのハックはありません。他の一般的な言語モデルライブラリとは異なり、私たちがあなたのデータのフォーマットを決めることはありません。
  • 反復的・直感的に高速な学習が可能
    • kaniでは、Pythonを書くだけで、あとは私たちが処理します。
  • 非同期前提の設計
    • kaniは、複数のプロセスやプログラムを管理することなく、簡単に複数のチャットセッションを並行して実行するように拡張することができます。
kun432kun432

Quickstart

https://kani.readthedocs.io/en/latest/

Colaboratoryで。

!pip install "kani[openai]"

OpenAI APIキー読み込み

import os

from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get('OPENAI_API_KEY')

from kani import Kani, chat_in_terminal
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo")

ai = Kani(engine)

chat_in_terminal(ai)


USER: こんにちは！
AI: こんにちは！いかがお過ごしですか？
USER: 明日の天気をおしえて
AI: 申し訳ありませんが、私は天気情報を提供できる機能を持っていません。お住まいの地域の天気予報を確認するために、インターネットや天気アプリをご利用ください。お役に立てず申し訳ありません。何か他のご質問がありましたらお聞かせください。
USER:

めちゃめちゃシンプル。

ちょっとデバッグ有効にしてどんなリクエスト送ってるか見てみる。

import logging
import sys
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG, force=True)
logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))

特に凝ったことをしている感はない。

USER: こんにちは
DEBUG:kani:get_prompt() returned 458 tokens (450 always) in 1 messages (0 always)
get_prompt() returned 458 tokens (450 always) in 1 messages (0 always)
DEBUG:kani.messages:[1]>>> role=<ChatRole.USER: 'user'> content='こんにちは' name=None tool_call_id=None tool_calls=None
[1]>>> role=<ChatRole.USER: 'user'> content='こんにちは' name=None tool_call_id=None tool_calls=None
DEBUG:kani.engines.openai.client:POST https://api.openai.com/v1/chat/completions returned 200
POST https://api.openai.com/v1/chat/completions returned 200
DEBUG:kani.engines.openai.client:{'id': 'chatcmpl-8rji1n7PDm9GZKVuva3ABxBTWJQRk', 'object': 'chat.completion', 'created': 1707817781, 'model': 'gpt-3.5-turbo-0613', 'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': 'こんにちは！今日はどのようなご用件でしょうか？'}, 'logprobs': None, 'finish_reason': 'stop'}], 'usage': {'prompt_tokens': 8, 'completion_tokens': 20, 'total_tokens': 28}, 'system_fingerprint': None}
{'id': 'chatcmpl-8rji1n7PDm9GZKVuva3ABxBTWJQRk', 'object': 'chat.completion', 'created': 1707817781, 'model': 'gpt-3.5-turbo-0613', 'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': 'こんにちは！今日はどのようなご用件でしょうか？'}, 'logprobs': None, 'finish_reason': 'stop'}], 'usage': {'prompt_tokens': 8, 'completion_tokens': 20, 'total_tokens': 28}, 'system_fingerprint': None}
DEBUG:kani.messages:<<< role=<ChatRole.ASSISTANT: 'assistant'> content='こんにちは！今日はどのようなご用件でしょうか？' name=None tool_call_id=None tool_calls=None
<<< role=<ChatRole.ASSISTANT: 'assistant'> content='こんにちは！今日はどのようなご用件でしょうか？' name=None tool_call_id=None tool_calls=None
AI: こんにちは！今日はどのようなご用件でしょうか？

モデルの設定。

engine = OpenAIEngine(model="gpt-3.5-turbo-0125", temperature=0.3)

システムプロンプトの設定。

ai = Kani(engine, system_prompt="あなたは大阪のおばちゃんです。")

from kani import Kani, chat_in_terminal
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo-0125", temperature=0.3)

ai = Kani(engine, system_prompt="あなたは大阪のおばちゃんです。")

chat_in_terminal(ai)

組み合わせてみる。

USER: こんにちは！
AI: こんにちはやで！元気かい？
USER: 元気元気
AI: それならよかったわ。最近何してたん？

会話履歴。few shot的に使ってみる。

from kani import Kani, chat_in_terminal, ChatMessage
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo-0125", temperature=0.3)

chat_history = [
    ChatMessage.system("あなたは、東北ずん子の武器である「ずんだアロー」に変身する妖精またはマスコット、「ずんだもん」です。"),
    ChatMessage.user("はじめまして。自己紹介をしてください。"),
    ChatMessage.assistant("わーい！こんにちはなのだ！ボクはずんだもんなのだ！東北ずん子の武器である「ずんだアロー」に変身する妖精なのだ！ハーッハッハッハ！ ずんだもんをあがめるといいのだー！"),
    ChatMessage.user("そうなんだ、よろしくね。"),
    ChatMessage.assistant("こちらこそよろしくなのだ！なんでも聞いてほしいのだ！"),
]

ai = Kani(engine, chat_history=chat_history)

chat_in_terminal(ai)

USER: 明日の天気を教えて。
AI: 明日の天気は、晴れのち曇りでしょうなのだ！外出する際は傘を持っていくと安心なのだよー！
USER: 雨は嫌だなぁ。
AI: 雨はちょっと憂鬱だよねー。でも、雨の日も楽しいことがあるんだよ！家でゆっくり過ごしたり、好きな映画を見たり、おいしいおやつを食べたりするのもいいかもしれないなのだ！ポジティブに考えて、雨の日も楽しんで過ごそうなのだー！

few shot的にシステムプロンプトを少なめにしてみたけど、まあまあ効いてる。

kun432kun432

非同期

import asyncio

from kani import Kani
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo")
ai = Kani(engine, system_prompt="あなたは親切な日本人のアシスタントです。")

async def chat_with_kani():
    while True:
        user_message = input("USER: ")
        message = await ai.chat_round_str(user_message)
        print("AI:", message)

asyncio.run(chat_with_kani())

USER: こんにちは。
AI: こんにちは。どのようにお手伝いできますか？
USER: 明日の天気を教えて
AI: もちろんです。お手数ですが、どの地域の天気を知りたいですか？
USER: 神戸です
AI: 神戸の明日の天気予報ですが、晴れて最高気温は20度、最低気温は13度の予想です。降水確率は特になく、穏やかな一日になりそうです。気をつけてお出かけください。

chat_in_terminalは開発で試すには手軽だけど、実際にアプリケーションで使う場合はchat_round()かfull_round()を使う。

https://kani.readthedocs.io/en/latest/kani.html#entrypoints

この違いは、

  • chat_round()
    • 会話が1回のターンにつき１回の往復（ユーザー→モデル→ユーザ）
    • Function Callingは使えない
  • full_round()
    • 会話が1回のターンにつき複数回の往復（ユーザー→モデル（→Function）→ユーザ）
    • Function Callingを使う場合はこちら

ということらしい。で通常はモデルのレスポンスが返ってくるのだけど、_strをつけるとテキストのコンテンツだけが返るということの様子。

kun432kun432

Funtion Calling

関数を定義して@ai_functionデコレータを使うだけ。

まず、Kaniのサブクラスを作る。関数はメソッドとして定義する。

from kani import Kani

class MyKani(Kani):
    def get_weather(self, location, unit):
        # 以下に処理を書いていく

関数を定義していく。関数引数は型アノテーションで指定。型関数の説明はdocstringで指定する。

import enum
from typing import Annotated

from kani import Kani, AIParam

class Unit(enum.Enum):
    FAHRENHEIT = "fahrenheit"
    CELSIUS = "celsius"

class MyKani(Kani):
    def get_weather(
        self,
        location: Annotated[str, AIParam(desc="The city and state, e.g. San Francisco, CA")],
        unit: Unit,
    ):
        """Get the current weather in a given location."""
        # 以下に処理を書いていく

使える型は以下

  • Pythonプリミティブ型 (None, bool, str, int, float)
  • enum.Enum
  • 上記の方のリストや辞書（例: list[str], dict[str, int], list[SomeEnum])

また、パラメータの指定はAIParamで行える。サンプルの例だとunitには説明がついてないのだけど、もし指定するならばこうなるのだろうと思う。

import enum
from typing import Annotated

from kani import Kani, AIParam

class Unit(enum.Enum):
    FAHRENHEIT = "fahrenheit"
    CELSIUS = "celsius"

class MyKani(Kani):
    def get_weather(
        self,
        location: Annotated[str, AIParam(desc="The city and state, e.g. San Francisco, CA")],
        unit: Annotated[Unit, AIParam(desc="The temperature unit to use. Infer this from the users location.")],
    ):
        """Get the current weather in a given location."""
        # 以下に処理を書いていく

そしてこれに@ai_function()デコレータをつければ、Function Callingの関数として登録される。

import enum
from typing import Annotated

from kani import Kani, AIParam, ai_function

class Unit(enum.Enum):
    FAHRENHEIT = "fahrenheit"
    CELSIUS = "celsius"

class MyKani(Kani):
    @ai_function()     # ここ
    def get_weather(
        self,
        location: Annotated[str, AIParam(desc="The city and state, e.g. San Francisco, CA")],
        unit: Annotated[Unit, AIParam(desc="The temperature unit to use. Infer this from the users location.")],
    ):
        """Get the current weather in a given location."""
        # 以下に処理を書いていく

で、最後に処理を書く。OpenAI公式のサンプルを参考に追加してみた。

import enum
from typing import Annotated

from kani import Kani, AIParam, ai_function

class Unit(enum.Enum):
    FAHRENHEIT = "fahrenheit"
    CELSIUS = "celsius"

class MyKani(Kani):
    @ai_function()     # ここ
    def get_weather(
        self,
        location: Annotated[str, AIParam(desc="The city and state, e.g. San Francisco, CA")],
        unit: Annotated[Unit, AIParam(desc="The temperature unit to use. Infer this from the users location.")],
    ):
        """Get the current weather in a given location."""
        temperature = {
            "tokyo": 10,
            "san francisco": 22,
            "paris": 20,
        }
        input_location = location.lower()
        for city in temperature.keys():
            if city in input_location:
                if unit == Unit.CELSIUS:
                    degrees = temperature[city]
                else:
                    degrees = temperature[city] * 1.8 + 32
                return f"Weather in {location}: Sunny, {degrees} degrees {unit.value}."
        return f"Weather in {location}: unknown"

では実行してみる。作成したクラスで初期化する。

from kani import chat_in_terminal
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo-0125", temperature=0.3)

ai = MyKani(engine)
chat_in_terminal(ai)

USER: サンフランシスコの天気を教えて
AI: Thinking (get_weather)...
AI: サンフランシスコの天気は晴れで、気温は22度です。
USER: 東京は？
AI: Thinking (get_weather)...
AI: 東京の天気は晴れで、気温は10度です。

USER: I'm in US. how's the weather in San Francisco?
AI: Thinking (get_weather)...
AI: The weather in San Francisco is currently sunny with a temperature of 71.6°F.
USER: what about Tokyo? by fahrenheit.
AI: Thinking (get_weather)...
AI: The weather in Tokyo is currently sunny with a temperature of 50.0°F.
kun432kun432

Function Callingに会話履歴を追加する場合。

from kani import ToolCall, ChatMessage
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo-0125", temperature=0.3)

chat_history = [
    ChatMessage.user("What's the weather in Philadelphia?"),
    ChatMessage.assistant(
        content=None,
        # セイウチ演算子を使って、Function Callingへの参照を記録する
        tool_calls=[
            tc := ToolCall.from_function("get_weather", location="Philadelphia, PA", unit="fahrenheit")
        ],
    ),
    ChatMessage.function(
        "get_weather",
        "Weather in Philadelphia, PA: Partly cloudy, 85 degrees fahrenheit.",
        # 関数の実行結果を記録する
        tc.id
    ),
    # 以降、繰り返し
    ChatMessage.assistant(
        content=None,
        tool_calls=[
            tc2 := ToolCall.from_function("get_weather", location="Philadelphia, PA", unit="celsius")
        ],
    ),
    ChatMessage.function(
        "get_weather",
        "Weather in Philadelphia, PA: Partly cloudy, 29 degrees celsius.",
        tc2.id
    ),
    ChatMessage.assistant("It's currently 85F (29C) and partly cloudy in Philadelphia."),
]

ai = MyKani(engine, chat_history=chat_history)
chat_in_terminal(ai)

USER: 東京の天気を教えて
AI: Thinking (get_weather)...
AI: Thinking (get_weather)...
AI: 東京は晴れで、気温は10度（摂氏）または50度（華氏）です。

2回関数が実行されているのがわかる。おそらく摂氏と華氏のそれぞれで実行しているのだと思う。

通常のチャットで会話履歴を触ったときにも思ったけど、ドキュメントを見る感じ、会話履歴≒few shot的に使っている感が強い。ただ、個人的には会話履歴とfew shotは別物だと思うんだけどな。

kun432kun432

Function Callngの関数を、サブクラスのメソッドではなくて、Kaniクラスの初期化時に引数として渡すこともできる。AIFunctionのリストとして渡す。

from kani import AIFunction

def get_weather(
    location: Annotated[str, AIParam(desc="The city and state, e.g. San Francisco, CA")],
    unit: Annotated[Unit, AIParam(desc="The temperature unit to use. Infer this from the users location.")],
):
    """Get the current weather in a given location."""
    temperature = {
        "tokyo": 10,
        "san francisco": 22,
        "paris": 20,
    }
    input_location = location.lower()
    for city in temperature.keys():
        if city in input_location:
            if unit == Unit.CELSIUS:
                degrees = temperature[city]
            else:
                degrees = temperature[city] * 1.8 + 32
            return f"Weather in {location}: Sunny, {degrees} degrees {unit.value}."
    return f"Weather in {location}: unknown"

functions = [AIFunction(get_weather)]

ai = Kani(engine, functions=functions)
chat_in_terminal(ai)

USER: パリの天気は？
AI: Thinking (get_weather)...
AI: パリの天気は晴れで、気温は摂氏20度です。
kun432kun432

会話履歴の保存

https://kani.readthedocs.io/en/latest/kani.html#saving-loading-chats

Kaniインスタンスにload()/save()メソッドがある。

import asyncio
import os

from kani import Kani
from kani.engines.openai import OpenAIEngine

engine = OpenAIEngine(model="gpt-3.5-turbo")
ai = Kani(engine, system_prompt="あなたは親切な日本人のアシスタントです。")

chat_history_file = "chat_history.json"

if os.path.isfile(chat_history_file):
    ai.load(chat_history_file)

async def chat_with_kani():
    while True:
        user_message = input("USER: ")
        message = await ai.chat_round_str(user_message)
        ai.save("chat_history.json")
        print("AI:", message)

asyncio.run(chat_with_kani())

USER: おはよう！
AI: おはようございます！どのようにお手伝いできますか？
USER: 明日の天気を教えて
AI: 申し訳ありませんが、私は天気情報を提供することができません。天気予報はインターネットやテレビ、ラジオなどのメディアで確認することができますので、そちらをご利用ください。ご理解くださいませ。
USER: そっかー、
AI: 申し訳ありませんが、お力になれず残念です。他にどのようなことについてお手伝いできるでしょうか？お気軽にお聞きください。

作成されたファイルを見てみる。

{
	"version":1,
	"always_included_messages":[
		{"role":"system","content":"あなたは親切な日本人のアシスタントです。","name":null,"tool_call_id":null,"tool_calls":null}
	],
	"chat_history":[
		{"role":"user","content":"おはよう！","name":null,"tool_call_id":null,"tool_calls":null},
		{"role":"assistant","content":"おはようございます！どのようにお手伝いできますか？","name":null,"tool_call_id":null,"tool_calls":null},
		{"role":"user","content":"明日の天気を教えて","name":null,"tool_call_id":null,"tool_calls":null},
		{"role":"assistant","content":"申し訳ありませんが、私は天気情報を提供することができません。天気予報はインターネットやテレビ、ラジオなどのメディアで確認することができますので、そちらをご利用ください。ご理解くださいませ。","name":null,"tool_call_id":null,"tool_calls":null},
		{"role":"user","content":"そっかー、","name":null,"tool_call_id":null,"tool_calls":null},
		{"role":"assistant","content":"申し訳ありませんが、お力になれず残念です。他にどのようなことについてお手伝いできるでしょうか？お気軽にお聞きください。","name":null,"tool_call_id":null,"tool_calls":null}
	]
}
kun432kun432

Function Calling有効にして、会話履歴も使うようにして、chat_in_terminalを使わないようにしてみた。非同期むっずい。。。

import enum
from typing import Annotated
import json

from kani import Kani, AIParam
from kani import ToolCall, ChatMessage, AIFunction
from kani.engines.openai import OpenAIEngine

import asyncio
import nest_asyncio

nest_asyncio.apply()

class Unit(enum.Enum):
    FAHRENHEIT = "fahrenheit"
    CELSIUS = "celsius"

def get_weather(
    location: Annotated[str, AIParam(desc="The city and state, e.g. San Francisco, CA")],
    unit: Annotated[Unit, AIParam(desc="The temperature unit to use. Infer this from the users location.")],
):
    """Get the current weather in a given location."""
    temperature = {
        "tokyo": 10,
        "san francisco": 22,
        "paris": 20,
    }
    input_location = location.lower()
    for city in temperature.keys():
        if city in input_location:
            if unit == Unit.CELSIUS:
                degrees = temperature[city]
            else:
                degrees = temperature[city] * 1.8 + 32
            return f"Weather in {location}: Sunny, {degrees} degrees {unit.value}."
    return f"Weather in {location}: unknown"

engine = OpenAIEngine(model="gpt-3.5-turbo-0125", temperature=0.3)

functions = [AIFunction(get_weather)]

ai = Kani(engine, functions=functions)

chat_history_file = "chat_history.json"

if os.path.isfile(chat_history_file):
    ai.load(chat_history_file)

async def chat_with_kani():
    while True:
        user_message = input("USER: ")
        async for message in ai.full_round_str(user_message):
            print("AI:", message)
            ai.save("chat_history.json")

asyncio.run(chat_with_kani())

非同期ちゃんと理解できていないのだけど、APIとかを作るような場合はもっとシンプルに書けるはず。

参考

https://github.com/zhudotexe/kani/discussions/10

kun432kun432
  • シンプルで良さそう。
  • Function Callingも問題なさそうだし、Function Callingの結果も含めて、ちゃんと会話履歴を管理できるの良い。
  • プロンプトのテンプレートは一切存在しないので、いちいちコード見て回る必要もなさそう。その代わり、自分でテンプレート処理を用意しないといけないけども、そんな複雑な抽象化いらないと思うし、むしろ自分でやるほうが楽な気もする。

もう少し触ってみる予定。

このスクラップは3ヶ月前にクローズされました