ホームニューステックニュースClaude Codeと征く、社内情報検索Slack botのStrands Agents対応への道（検証編） #AWS

Claude Codeと征く、社内情報検索Slack botのStrands Agents対応への道（検証編） #AWS

By インモビ運営局

2025年6月3日

0

2

Claude Codeと征く、社内情報検索Slack botのStrands Agents対応への道（検証編） #AWS

こんにちは、yu_Matsuです！
先月は生成AI界隈で大きな発表が多く、頭がパンクしそうになる日々が続きましたが、皆様はいかがでしたでしょうか。

今回なその中でも Strands Agents と Claude Code のキャッチアップを兼ねて、社内情報検索Slack botのアップデートの検証を行ったため、その体験記を記事にしました。

Strands Agents は、2025年5月16日に発表されたばかりの AWS公式のAIエージェントのSDK になります。エージェントのSDKといえば、LangChain/LangGraph や MastraAI などがよく利用されていますが、遂にAWSからも提供されることとなりました！

<br />

特徴として、「わずか数行のコードで AI エージェントを構築・実行するモデル駆動型アプローチを採用したオープンソース SDK」と謳われている通り、本当に簡単にAIエージェントを実装することが出来ます。

以下は公式ドキュメントのQuickStartから拝借した、最小単位のコード例になります。たったこれだけでAIエージェントが実現出来てしまうのは驚きです！

from strands import Agent

# Create an agent with default settings
agent = Agent()

# Ask the agent a question
agent("Tell me about agentic AI")

もちろん、ツールの利用やマルチエージェントも対応しており、従来と比べるとより直感的に実装できるようになっていますので、AIエージェントの敷居がまた一段下がったのではないでしょうか。

この記事のもう一つのトピックである Claude Code ですが、こちらは先月の Anthropic社の開発者向けイベントである Claude with Code で正式リリースが発表されたエージェント型コーディングツールになります。

ベータ版ではCLIからの利用のみでしたが、正式リリースに伴い、VSCodeなどのIDE上での利用も可能になりました。

<br />

以下のコマンドでインストールすることができ、

npm install -g @anthropic-ai/claude-code

Claude Codeを起動したいプロジェクトのディレクトリに移動し、以下のコマンドを実行することで起動できます

初回起動後は認証が必要になりますので、詳しくは下記のドキュメントをご参考ください。

<br />

セットアップが完了すると、以下のようなメッセージが表示されます。Claude Codeへの指示は下の入力欄に入力して実行する感じになります。

VSCodeの例になりますが、エディタ右上の３点リーダーを押下すると「Run Claude Code」が現れるようになっており、CLIからではなくここからでも起動できるようになっています。

本題に入る前に、まずは既存のSlack botのコードのサンプルを以下に掲載します。

Slack boltを利用して実装した簡単なSlack botであり、裏側ではKnowledge bases for Amazon Bedrock（以降「Knowledge bases」） の Retriever を利用した RAG 搭載のLLMチェーンが動いています。

メッセージがストリーミングで表示されるようにしているため、少し途中の処理が複雑になっていますがご了承ください。

app.py

import os
import asyncio
import logging
from typing import Dict, List
from dotenv import load_dotenv

# slack bolt周り
from slack_bolt import App
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from slack_bolt.async_app import AsyncApp

# langchain周り
from langchain_aws import ChatBedrockConverse
from langchain_aws import AmazonKnowledgeBasesRetriever
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# Load environment variables from .env file
load_dotenv()

# 環境変数の設定
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_APP_TOKEN = os.environ["SLACK_APP_TOKEN"]
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
KNOWLEDGE_BASE_ID = os.environ["KNOWLEDGE_BASE_ID"]  # Amazon Knowledge Base ID

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 非同期Slackアプリの初期化
app = AsyncApp(token=SLACK_BOT_TOKEN)

# Amazon Knowledge Baseリトリーバーの設定
retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KNOWLEDGE_BASE_ID,
    retrieval_config={
        "vectorSearchConfiguration": {
            "numberOfResults": 4  # 検索結果の数
        }
    },
    region_name=AWS_REGION
)

# BedrockのAIモデルの初期化
bedrock_model = ChatBedrockConverse(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    region_name=AWS_REGION
)

# ドキュメントをフォーマットする関数
def format_docs(docs):
    """取得したドキュメントを文字列に整形"""
    return "\n\n".join(doc.page_content for doc in docs)

# プロンプトテンプレート
prompt = ChatPromptTemplate.from_template("""
あなたは専門的なアシスタントです。与えられたコンテキストに基づいて質問に答えてください。

コンテキスト:
{context}

質問: {question}

回答は日本語で行ってください。
""")

# RAGのLLMチェーン
chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | prompt
    | bedrock_model
    | StrOutputParser()
)

# メンションされたときの処理
@app.event("app_mention")
async def handle_mention(event, say):
    """Botがメンションされたときの処理"""
    user_id = event["user"]
    channel_id = event["channel"]
    text = event["text"]
    
    # の部分を削除してメッセージのみを取得
    message_text = text.split(">", 1)[1].strip() if ">" in text else text
    
    try:
        # スレッドで初期メッセージ送信（タイピング中の表示代わり）
        response = await say(
            text="考え中...",
            thread_ts=event["ts"]
        )
        message_ts = response["ts"]
        
        # ストリーミング処理用の状態管理
        full_response = ""
        update_interval = 1.0  # 更新間隔（秒）
        last_update = 0
        
        # RAGチェーンを非同期ストリーミングで実行
        async for chunk in rag_chain.astream(message_text):
            # 新しいチャンクを追加（文字列として処理）
            if isinstance(chunk, str):
                full_response += chunk
            else:
                full_response += str(chunk)
            
            # 一定間隔でSlackメッセージを更新
            current_time = asyncio.get_event_loop().time()
            if current_time - last_update >= update_interval and full_response.strip():
                await app.client.chat_update(
                    channel=channel_id,
                    ts=message_ts,
                    text=full_response
                )
                last_update = current_time
        
        # 最終的な完全な応答を更新
        if full_response.strip():
            await app.client.chat_update(
                channel=channel_id,
                ts=message_ts,
                text=full_response
            )
        else:
            await app.client.chat_update(
                channel=channel_id,
                ts=message_ts,
                text="応答を生成できませんでした。"
            )
    except Exception as e:
        logger.error(f"エラーが発生しました: {e}")
        await say(f"エラーが発生しました: {str(e)}")

# メイン関数
async def main():
    # AsyncSocketModeHandlerの初期化
    handler = AsyncSocketModeHandler(app, SLACK_APP_TOKEN)
    # アプリの起動
    await handler.start_async()

# スクリプトとして実行された場合
if __name__ == "__main__":
    asyncio.run(main())

RAGのデータソースとなるドキュメントは、サンプルとして「広報東京都2025年5月号」のPDF版を利用しました。

<br />

動作イメージは以下のような感じです。（Slack botの設定等は本題から逸れるため省略します。）

Slack bot（今回は「テスト用ボット」という名前）にメンション付きでメッセージを送ると、スレッドが作成されて会話が開始します。

上記PDFの内容に関して答えてくれるかどうか試してみます。『「夢のみち」道路愛護ポスターコンクール』というコンクールの応募締め切りについて聞いてみます。

正解は「6月27日」ですが、問題なく回答しています。

以降で、このSlack botのコードをStrands Agentsで書き換えつつ、MCPサーバーも利用できるようにしていきます！

早速 Claude Code と共に進めていきたいのですが、まずはお互いにStrands Agentsに慣れたいと思います。

Strands Agentsの説明でも触れた通り、非常に簡単なプログラムでAIエージェントを実装できます。以下のような test.py というベースラインのプログラム（ほぼQuickStartのもの）を作成し、機能を付け加えてみたいと思います。

test.py

import boto3
from strands import Agent
from strands.models import BedrockModel

# Create a BedrockModel
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    region_name='us-east-1',
    temperature=1.0,
)

agent = Agent(model=bedrock_model)

if __name__ == "__main__":
    result = agent("こんにちは！")
    print(result.message)