Strandsで簡単な生成AIチャットボットを作ってみた #strands #ai #llm #gemini
はじめに
生成AIチャットボットに関するこれまでの記事では LangChain, LangGraph を利用していました。例：
- 生成AIチャットボットでMCP (Model Context Protocl)を使ってみよう
- LangGraphのTool callingでOpenAI APIのFunction callingを試してみよう
- etc…
LangGraphは非常に柔軟ですが「グラフを構築する」という概念の学習コストが高く、先日の記事「Google ADKで簡単な生成AIチャットボットを作ってみた」のようにAIエージェント指向のフレームワークが用いられることも増えてきたように思います。
本稿では生成AI (Google Gemini)を利用し、セッションをS3互換ストレージ (SeaweedFS) に保存するチャットボットを、LangGraphと、Amazon Web Services (AWS)が提供するAIエージェント作成フレームワークである Strands Agents の両方で作成し、その構造的な違いを比較してみます。
生成AIチャットボットの基本仕様
作成するチャットボットの仕様は次の通り、先日の記事とほぼ同様です。
- 簡易なコマンドラインインターフェイスとする
- ユーザの入力を生成AI (Google Gemini)に渡して、回答を表示する
- 会話セッションは保存し、会話を継続できるようにする
- ボット起動時、会話セッションを特定するIDをUUIDで生成する
- ユーザがUUIDを入力すると、その会話セッションに切り替える
- ユーザが
quit,exit,qを入力すると、ボットを終了する
このようにとても単純ですが、一問一答で記憶をなくすのではなく、会話が継続できる仕組みとします。
LangGraphによる実装
まずはLangGraphによる実装です。先日の記事と同様です。
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from langchain_google_genai import ChatGoogleGenerativeAI
from psycopg import Connection
from langgraph.checkpoint.postgres import PostgresSaver
from yaspin import yaspin
import uuid
import re
import os
from dotenv import load_dotenv
load_dotenv()
# os.getenv("GOOGLE_API_KEY")
llm = ChatGoogleGenerativeAI(
model = os.getenv("GOOGLE_MODEL_NAME", "gemini-3-flash-preview"),
temperature = 0.95,
max_output_tokens = 1024,
)
class State(TypedDict):
messages: Annotated[list, add_messages]
def chatbot(state: State):
with yaspin(text="Processing", color="yellow") as spinner:
res = llm.invoke(state["messages"])
spinner.ok("✅ ")
return {"messages": res}
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")
thread_id = str(uuid.uuid4())
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)
def stream_graph_updates(user_input: str, thread_id: str):
db_uri = os.getenv("DB_URI")
with PostgresSaver.from_conn_string(db_uri) as checkpointer:
checkpointer.setup()
graph = graph_builder.compile(checkpointer=checkpointer)
events = graph.stream(
{"messages": [("user", user_input)]},
{"configurable": {"thread_id": thread_id}},
stream_mode="values"
)
for event in events:
message = event["messages"][-1]
if type(message) == AIMessage:
print(message.content)
while True:
try:
print("\nThread ID: " + thread_id)
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
elif bool(UUID_RE.match(user_input)):
thread_id = user_input
else:
stream_graph_updates(user_input, thread_id)
except Exception as e:
print(f"error: {e}")
break
State クラスによる状態管理でメッセージ履歴を自動追加、 StateGraph による chatbot ノードの追加、開始・終了ノードの追加によるグラフの構築、といった点が特徴です。また、PostgreSQLへの保存は「チェックポイント」という概念で行われています。
さまざなま処理ノードを追加したり、条件分岐ができる、というのはわかるのですが、今回のような単純なチャットボットには「お約束」や「おまじない」といった感じで少しわかりづらい、「大は小を兼ねる」といってもちょっととっつきづらい、という感想になりそうです。
Strandsによる実装
次にLangGraphによる実装をStrandsで再実装したものです。
from strands import Agent
from strands.models.gemini import GeminiModel
from strands.session import S3SessionManager
from botocore.config import Config as BotocoreConfig
from yaspin import yaspin
import asyncio
import uuid
import re
import os
from dotenv import load_dotenv
load_dotenv()
model = GeminiModel(
model_id = os.getenv("GOOGLE_MODEL_NAME", "gemini-3-flash-preview"),
params = {
"temperature": 0.95,
"max_output_tokens": 1024,
},
)
SYSTEM_PROMPT = "あなたは親切なアシスタントです。ユーザーの質問に丁寧に答えてください。"
S3_BUCKET = os.getenv("S3_BUCKET", "agent-sessions")
S3_PREFIX = os.getenv("S3_PREFIX", "")
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
boto_client_config = BotocoreConfig(s3={"addressing_style": "path"})
thread_id = str(uuid.uuid4())
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)
async def stream_runner_updates(user_input: str, thread_id: str):
session_manager = S3SessionManager(
session_id=thread_id,
bucket=S3_BUCKET,
prefix=S3_PREFIX,
region_name=AWS_REGION,
boto_client_config=boto_client_config,
)
agent = Agent(
model=model,
system_prompt=SYSTEM_PROMPT,
session_manager=session_manager,
callback_handler=None,
)
with yaspin(text="Processing", color="yellow") as spinner:
result = await agent.invoke_async(user_input)
spinner.ok("✅ ")
print(result.message["content"][0].get("text", ""))
while True:
try:
print("\nThread ID: " + thread_id)
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
elif bool(UUID_RE.match(user_input)):
thread_id = user_input
else:
asyncio.run(stream_runner_updates(user_input, thread_id))
except Exception as e:
print(f"error: {e}")
break
できる限りLangGraphでの実装と構造が変わらないように、Strandsで再実装しました。
Strandsでは「モデル」「エージェント」「セッションマネージャ」という3つの要素で構成されています。
- モデル: アプリの「頭脳」に相当。
- エージェント: アプリの「手足」に相当。
- セッションマネージャ: ユーザとエージェントのやりとりを記録する。
グラフの状態遷移を手続きに構築するLangGraphに対し、Strandsは「モデル」「セッション」といった部品を組み合わせる設計です。これにより、ロジックの流れに縛られず、必要な機能をオブジェクトとして差し替えるだけでエージェントを構成できます。
なお、Strandsの組み込みセッションマネージャは、2026年4月現在、ローカルファイルシステムとAmazon S3バケットのみの対応となっています。本稿ではローカルで実行できるS3互換ストレージとしてSeaweedFSを採用しました。テスト用のDocker Composeは次の通りです。
services:
weedmini:
container_name: weedmini
image: chrislusf/seaweedfs:4.22@sha256:dc40601b7a598dbaa0312e4aadf1cc239de2ed6a177babd2f181a6d766a20dd6
command:
- server
- -s3
restart: always
ports:
- "8333:8333"
environment:
AWS_ACCESS_KEY_ID: admin
AWS_SECRET_ACCESS_KEY: secret
volumes:
- weed-data:/data
volumes:
weed-data:
とても簡単なアクセスキーを指定しているので本番環境では利用しないでください。
これで seaweedfs を起動し、次の環境変数を設定します。
AWS_ENDPOINT_URL_S3=http://localhost:8333 AWS_ACCESS_KEY_ID=admin AWS_SECRET_ACCESS_KEY=secret AWS_REGION=us-east-1 S3_BUCKET=agent-sessions
そしてバケットの作成を行います。
aws --endpoint-url "$AWS_ENDPOINT_URL_S3" s3 mb s3://"$S3_BUCKET"
これでチャットボットを起動すれば、seaweedfsにセッションを記録して会話が継続できます。
さらに「関心の分離」を意識してリファクタリングしたコードがこちらです。
from strands import Agent
from strands.models.gemini import GeminiModel
from strands.session import S3SessionManager
from botocore.config import Config as BotocoreConfig
from yaspin import yaspin
import asyncio
import uuid
import re
import os
from dotenv import load_dotenv
load_dotenv()
SYSTEM_PROMPT = "あなたは親切なアシスタントです。ユーザーの質問に丁寧に答えてください。"
S3_BUCKET = os.getenv("S3_BUCKET", "agent-sessions")
S3_PREFIX = os.getenv("S3_PREFIX", "")
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)
def build_model() -> GeminiModel:
return GeminiModel(
model_id=os.getenv("GOOGLE_MODEL_NAME", "gemini-3-flash-preview"),
params={
"temperature": 0.95,
"max_output_tokens": 1024,
},
)
def build_agent(model: GeminiModel, session_id: str) -> Agent:
boto_client_config = BotocoreConfig(s3={"addressing_style": "path"})
session_manager = S3SessionManager(
session_id=session_id,
bucket=S3_BUCKET,
prefix=S3_PREFIX,
region_name=AWS_REGION,
boto_client_config=boto_client_config,
)
return Agent(
model=model,
system_prompt=SYSTEM_PROMPT,
session_manager=session_manager,
callback_handler=None,
)
async def chat(agent: Agent, user_input: str) -> None:
with yaspin(text="Processing", color="yellow") as spinner:
result = await agent.invoke_async(user_input)
spinner.ok("✅ ")
print(result.message["content"][0].get("text", ""))
async def main() -> None:
model = build_model()
session_id = str(uuid.uuid4())
agent = build_agent(model, session_id)
while True:
try:
print("\nThread ID: " + session_id)
user_input = await asyncio.to_thread(input, "User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
if UUID_RE.match(user_input):
session_id = user_input
agent = build_agent(model, session_id)
continue
await chat(agent, user_input)
except Exception as e:
print(f"error: {e}")
break
if __name__ == "__main__":
asyncio.run(main())
エージェントの構築、セッション管理、チャットの実行ロジックをメソッドごとに切り出しています。
まとめ
本稿では、これまでよく利用していたLangGraphでの簡単な生成AIチャットボットのコードを、Strandsで再実装することで、どのような違いがあるかを見てみました。
LangGraphでの「グラフをどのように制御するか」という低レイヤーな実装の悩みから解放され、Strandsでは「エージェントにどのような役割を与えるか」という本質的な設計に集中できるようになると思います。LangGraphが持つ柔軟なカスタマイズ性を必要としないシンプルなケースでは、Strandsのほうがコードの見通しがよく、直感的であり、メンテナンス性も高くなると言えるでしょう。
エージェントフレームワークは数多く存在するので、今後もクリエーションラインではそれぞれ確認して紹介していこうと思います。
Author
Chef・Docker・Mirantis製品などの技術要素に加えて、会議の進め方・文章の書き方などの業務改善にも取り組んでいます。「Chef活用ガイド」共著のほか、Debian Official Developerもやっています。
