【Groq】非同期でtool_callsを利用する

概要

• 本当はlangchain_groqを使いたいところですが、langchain_groqは非同期に対応していないため、groqのネイティブライブラリを使います。
• messagesはlangchain_core.messagesではなく、groqライブラリの形式に合わせる必要があります。
• toolもgroqライブラリの形式に合わせる必要がありますが、langchain_core.toolsと形は似ているので、toolデコレーターを使います。

実装内容

toolの書き方

• 本来は以下のようなdictで定義します。
  • トップ階層にtypeとfunctionを定義、
  • function階層にname、description、parametersを定義しています。
    （この形式はlangchain_core.toolsのtoolと似ていますが、key名が"args"ではなく"parameters"です）

        tools=[
            {
                "type": "function",
                "function": {
                    "name": "get_current_weather",
                    "description": "指定された場所の現在の天気を取得します",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {
                                "type": "string",
                                "description": "都市と州、例：San Francisco, CA"
                            },
                            "unit": {
                                "type": "string",
                                "enum": ["celsius", "fahrenheit"]
                            }
                        },
                        "required": ["location"]
                    }
                }
            }
        ]

Langchain愛用の私としては、将来的にLangchain移行になった時に備え、langchain_core.tools.toolデコレーターで書いて、それをgroq用に書き直すという方法を採用しました。

from langchain_core.tools import tool

@tool
def get_current_weather(location: str, unit: Optional[Literal["celsius", "fahrenheit"]] = "celsius") -> str:
    """天気を調べるためのツールです。
    指定された場所の現在の天気を取得します"""
    # 実際の実装では天気API等を呼び出すコードをここに書きます
    print(f"天気を調べています: {location}")
    return f"{location}の天気: 晴れ、気温: 22{unit[0] if unit == 'celsius' else 'F'}"

# ツールをリスト化
tools = [get_current_weather]

# ツールスキーマを抽出してGroq APIのフォーマットに変換
groq_tools: List[ChatCompletionToolParam] = []
for tool_obj in tools:
    groq_tool: ChatCompletionToolParam = {
        "type": "function",
        "function": {
            "name": tool_obj.name,
            "description": tool_obj.description,
            "parameters": tool_obj.args,
        }
    }
    groq_tools.append(groq_tool)

messages

• 使い方はlangchain_core.messagesと似ており、roleとcontentがあります。（roleは必須っぽい）
• groq.types.chatにAI,Human,Systemのメッセージ型が定義されています

from groq.types.chat import ChatCompletionMessageParam, ChatCompletionToolParam
from groq.types.chat.chat_completion_system_message_param import ChatCompletionSystemMessageParam
from groq.types.chat.chat_completion_user_message_param import ChatCompletionUserMessageParam
from groq.types.chat.chat_completion_assistant_message_param import ChatCompletionAssistantMessageParam

messages: List[ChatCompletionMessageParam] = []

system_message = ChatCompletionSystemMessageParam(role="system", content="あなたはチャットbotです。天気に関する質問がある場合のみ、get_current_weatherツールを使用してください。それ以外の通常の会話は、ツールを使わず直接テキストで応答してください。")

messages.append(system_message)

クライアントライブラリ

• 以下のように設定すると、回答をストリーミングで利用できます。
• モデルは2025年4月現在、meta-llama/llama-4-scout-17b-16e-instructが最新。古いモデルは日本語認識力が低く、思いがけないタイミングでツール呼び出しをしてしまうことがある。

from groq import AsyncGroq

# APIリクエスト
stream = await client.chat.completions.create(
    messages=messages,
    tools=groq_tools,
    tool_choice="auto",
    model="meta-llama/llama-4-scout-17b-16e-instruct",
    temperature=0.5,
    stream=True,
)

ストリーム回答の中身と処理

• ストリーミングの中身はchunk.choices[0]に格納されています。（1つしかない?）
• テキストを得たい時はchunk.choices[0].delta
• 呼び出されたツールを得たい時はchunk.choices[0].delta
• ツールはfunction名と引数が返されるだけで、自動で実行はしてくれない。toolsの中から該当するツールを見つけ出して、tool_obj.invoke(args)で実行する必要がある。
• AIの返答が終了した時、ツール呼び出し無しはchoice.finish_reason == "stop"が、ツール呼び出しした後は、choice.finish_reason == "tool_calls"が返される

async for chunk in stream:
    choice = chunk.choices[0]
    if choice.delta.content:
        ai_sentence += choice.delta.content
        print(choice.delta.content, end="|")
        messages.append(ChatCompletionAssistantMessageParam(role="assistant", content=ai_sentence))

    if choice.delta.tool_calls:
        for tool_call in choice.delta.tool_calls:
            fn = tool_call.function
            if fn and fn.name and fn.arguments:
                name = fn.name
                args = json.loads(fn.arguments)
                print(f"    ツール名: {name}, 引数: {args}")
                try:
                    # ツールを実行
                    for tool_obj in tools:
                        if tool_obj.name == name:
                            result = tool_obj.invoke(args)
                            print(f"     結果: {result}")
                            if tool_call.id:
                                messages.append(ChatCompletionToolMessageParam(role="tool", content=result,tool_call_id=tool_call.id))
                except:
                    print(f"     失敗: {name}, {args}")
    
    # ツールが呼び出されずに終了した時に発火する
    if choice.finish_reason == "stop":
        yield None

    # ツールが呼び出された後で終了した時に発火する
    if choice.finish_reason == "tool_calls":
        yield None

コードサンプル

sample.py

import json
import asyncio
from typing import List,Optional, Literal

from groq import AsyncGroq
from groq.types.chat import ChatCompletionMessageParam, ChatCompletionToolParam
from groq.types.chat.chat_completion_system_message_param import ChatCompletionSystemMessageParam
from groq.types.chat.chat_completion_user_message_param import ChatCompletionUserMessageParam
from groq.types.chat.chat_completion_assistant_message_param import ChatCompletionAssistantMessageParam
from groq.types.chat.chat_completion_tool_message_param import ChatCompletionToolMessageParam
from groq.types.chat.chat_completion_chunk import Choice, ChoiceDelta

from langchain_core.tools import tool

client = AsyncGroq()

messages: List[ChatCompletionMessageParam] = [
    ChatCompletionSystemMessageParam(role="system", content="あなたはチャットbotです。天気に関する質問がある場合のみ、get_current_weatherツールを使用してください。それ以外の通常の会話は、ツールを使わず直接テキストで応答してください。"),
]

# LangChainのtoolデコレータを使用して関数をツールとして定義
@tool
def get_current_weather(location: str, unit: Optional[Literal["celsius", "fahrenheit"]] = "celsius") -> str:
    """天気を調べるためのツールです。
    指定された場所の現在の天気を取得します"""
    # 実際の実装では天気API等を呼び出すコードをここに書きます
    print(f"天気を調べています: {location}")
    return f"{location}の天気: 晴れ、気温: 22{unit[0] if unit == 'celsius' else 'F'}"



# ツールをリスト化
tools = [get_current_weather]

# ツールスキーマを抽出してGroq APIのフォーマットに変換
groq_tools: List[ChatCompletionToolParam] = []
for tool_obj in tools:
    groq_tool: ChatCompletionToolParam = {
        "type": "function",
        "function": {
            "name": tool_obj.name,
            "description": tool_obj.description,
            "parameters": tool_obj.args,
        }
    }
    groq_tools.append(groq_tool)


async def human_talk():
    """ユーザーの発言"""
    text = "こんにちは〜。今日はポカポカいい天気ですね。昨日は雨で桜が散ってしまって残念です。さっき私は天気を何と言いましたか。ところで、明日は大阪城公園に行く予定です。大阪城公園の天気を教えてください。"

    sentence = ""
    for chunk in text:
        await asyncio.sleep(0.05)
        sentence += chunk
        if chunk == "。" or chunk == "？" or chunk == "！":
            yield sentence
            sentence = ""


async def main():
    async for human_sentence in human_talk():
        print("==========ユーザーの発言==========")
        print(human_sentence)
        messages.append(ChatCompletionUserMessageParam(role="user", content=human_sentence))

    
        # APIリクエスト
        stream = await client.chat.completions.create(
            messages=messages,
            tools=groq_tools,
            tool_choice="auto",
            model="meta-llama/llama-4-scout-17b-16e-instruct",
            temperature=0.5,
            stream=True,
        )

        # LLMから返された増分デルタを表示します
        print("==========AI応答==========")
        ai_sentence = ""
        try:
            async for chunk in stream:
                choice = chunk.choices[0]
                if choice.delta.content:
                    ai_sentence += choice.delta.content
                    print(choice.delta.content, end="|")
                    messages.append(ChatCompletionAssistantMessageParam(role="assistant", content=ai_sentence))

                if choice.delta.tool_calls:
                    for tool_call in choice.delta.tool_calls:
                        fn = tool_call.function
                        if fn and fn.name and fn.arguments:
                            name = fn.name
                            args = json.loads(fn.arguments)
                            print(f"    ツール名: {name}, 引数: {args}")
                            try:
                                # ツールを実行
                                for tool_obj in tools:
                                    if tool_obj.name == name:
                                        result = tool_obj.invoke(args)
                                        print(f"     結果: {result}")
                                        if tool_call.id:
                                            messages.append(ChatCompletionToolMessageParam(role="tool", content=result,tool_call_id=tool_call.id))
                            except:
                                print(f"     失敗: {name}, {args}")
                
                # ツールが呼び出されずに終了した時に発火する
                if choice.finish_reason == "stop":
                    yield None

                # ツールが呼び出された後で終了した時に発火する
                if choice.finish_reason == "tool_calls":
                    yield None
            
            print("")

        except Exception as e:
            print(f"ストリーミングエラー: {e}")
            yield None


if __name__ == "__main__":
    async def run_main():
        async for _ in main():
            pass
    asyncio.run(run_main())

実行結果

==========ユーザーの発言==========
こんにちは〜。
==========AI応答==========
こ|ん|に|ち|は|！|どう|ぞ|よ|ろ|しく|お願い|します|。|天気|についての|質問|が|あれば|、お|気|軽|にお|聞|き|ください|。|
==========ユーザーの発言==========
今日はポカポカいい天気ですね。
==========AI応答==========
そう|ですね|！|今日|は|とても|気持ち|の|良い|天気|ですね|。|どこ|か|お|出|かけ|予定|ですか|？|
==========ユーザーの発言==========
昨日は雨で桜が散ってしまって残念です。
==========AI応答==========
そう|ですね|。|雨|で|桜|が|散|る|のは|早く|て|残|念|です|よね|。| <|function|=get|_current|_weather|>{"|location|":| "\|u|677|1|\u|4|e|ac|\u|90|f|1|",| "|unit|":| "|c|elsius|"}| </|function|>|
==========ユーザーの発言==========
さっき私は天気を何と言いましたか。
==========AI応答==========
あなた|は|「|今日|は|ポ|カ|ポ|カ|いい|天気|ですね|。」|と言|いました|。|
==========ユーザーの発言==========
ところで、明日は大阪城公園に行く予定です。
==========AI応答==========
大阪|城|公園|に行|く|ん|ですね|。|楽し|み|ですね|！|
==========ユーザーの発言==========
大阪城公園の天気を教えてください。
==========AI応答==========
    ツール名: get_current_weather, 引数: {'location': '大寿城公园', 'unit': 'celsius'}
天気を調べています: 大寿城公园
     結果: 大寿城公园の天気: 晴れ、気温: 22c