はじめに
いつもは ChatBedrock を使っていたのですが、 ChatBedrockConverse を使えばストリーミング応答することができる、ということを聞いたので動かしてみました。
Bedrockに質問を送信した後、すぐに返答が始まり、少しずつ文字が出力されること、が期待値です。
こちらも参考にして下さい。
ChatBedrock の使い方
代表的なパラメータの説明
region_name： AWS のリージョン (us-east-1 など)
model： 利用するAIモデルのID
max_tokens： 応答の最大トークン数
temperature： 応答のランダム性
実装例
モデルにNovaを指定しています。
invoke("質問内容")をすることで、Bedrockに質問を投げます。
これを動かすと、回答を得られるまで少し時間がかかり、全ての回答がまとめて出力されます。
rom langchain.prompts import PromptTemplate
from langchain_aws import ChatBedrock
# プロンプトテンプレートの定義
template = PromptTemplate(
input_variables=["topic"],
template="あなたは {name}のアーキテクトです。{topic} について詳しく教えてください。",
partial_variables={"name": "AWS"},
)
formatted_prompt = template.format(topic="BedrockのNova")
print(formatted_prompt)
llm = ChatBedrock(
region_name="us-east-1",
model="amazon.nova-micro-v1:0",
max_tokens=200,
temperature=0,
)
follow_up_result = llm.invoke(formatted_prompt)
print(follow_up_result)
ChatBedrockConverse の使い方
代表的なパラメータの説明
※ChatBedrockと同じです。
region_name： AWS のリージョン (us-east-1 など)
model： 利用するAIモデルのID
max_tokens： 応答の最大トークン数
temperature： 応答のランダム性
ChatBedrock との違い
ChatBedrockConverseは、invoke()とstream()で、Bedrockに質問を投げることができます。
実装例 invoke()を使用
ChatBedrockと同じ方法で実行することができます。
rom langchain.prompts import PromptTemplate
from langchain_aws import ChatBedrockConverse
# プロンプトテンプレートの定義
template = PromptTemplate(
input_variables=["topic"],
template="あなたは {name}のアーキテクトです。{topic} について詳しく教えてください。",
partial_variables={"name": "AWS"},
)
formatted_prompt = template.format(topic="BedrockのNova")
print(formatted_prompt)
llm = ChatBedrockConverse(
region_name="us-east-1",
model="amazon.nova-micro-v1:0",
max_tokens=200,
temperature=0,
)
follow_up_result = llm.invoke(formatted_prompt)
print(follow_up_result)
実装例 stream()を使用
stream("質問内容")をすることで、Bedrockに質問を投げます。
これを動かすと、すぐに返答が始まります。
rom langchain.prompts import PromptTemplate
from langchain_aws import ChatBedrockConverse
# プロンプトテンプレートの定義
template = PromptTemplate(
input_variables=["topic"],
template="あなたは {name}のアーキテクトです。{topic} について詳しく教えてください。",
partial_variables={"name": "AWS"},
)
formatted_prompt = template.format(topic="BedrockのNova")
print(formatted_prompt)
llm = ChatBedrockConverse(
region_name="us-east-1",
model="amazon.nova-micro-v1:0",
max_tokens=200,
temperature=0,
)
for chunk in llm.stream(formatted_prompt):
print(chunk.content, end="", flush=True)
動かしてみると、ストリーミング応答を得られますが、次のようなjson形式の応答となるため、回答が分かりづらいです。
[][{'type': 'text', 'text': 'Amazon', 'index': 0}][{'type': 'text', 'text': ' Bed', 'index': 0}][{'type': 'text', 'text': 'rock', 'index': 0}][{'type': 'text', 'text': 'は', 'index': 0}][{'type': 'text', 'text': '、', 'index': 0}][{'type': 'text', 'text': 'AWS', 'index': 0}],・・・
ChatBedrockConverseのストリーミング応答を整形する
以下のコードで、ChatBedrockConverse を使用したストリーミング応答を、text部分だけを抽出して整形した形で出力することができます。
rom langchain.prompts import PromptTemplate
from langchain_aws import ChatBedrockConverse
# プロンプトテンプレートの定義
template = PromptTemplate(
input_variables=["topic"],
template="あなたは {name}のアーキテクトです。{topic} について詳しく教えてください。",
partial_variables={"name": "AWS"},
)
formatted_prompt = template.format(topic="BedrockのNova")
print(formatted_prompt)
llm = ChatBedrockConverse(
region_name="us-east-1",
model="amazon.nova-micro-v1:0",
max_tokens=200,
temperature=0,
)
response_text = ""
for chunk in llm.stream(formatted_prompt):
if hasattr(chunk, "content") and isinstance(chunk.content, list):
# 'text' キーが存在する場合にのみ処理
text_chunk = "".join([part["text"] for part in chunk.content if "text" in part])
response_text += text_chunk
print(text_chunk, end="", flush=True)
応答を解析して text の部分だけを抽出し、リアルタイムで表示するようにしました。
この工夫により、ストリーミング応答がスムーズに表示されるようになりました。
まとめ
-
ChatBedrockConverseを使うことで、Bedrock でストリーミング応答が可能になる - そのままの出力は見づらいので、整形することで対応する
自身で出力の工夫をせずとも、LangChainの出力クラスがあるような・・・
今後は、その辺りも調べていきたいと思います。