【LangChain v1.0】Bedrockと接続してDynamoDBで会話の永続化する

Last updated at 2025-10-26Posted at 2025-10-26

LangChain v1.0がリリースされました。

僕自身、ほとんど触ったことないフレームワークなので、今更ながら触ってみようかと思います。
LangChain overview - Docs by LangChain

実行環境

LangChainバージョン：1.0.0
langchain-aws バージョン：1.0.0

langchain-awsも2025/10/18に最新バージョン v1.0.0をリリースしていました。
langchain-aws

ちなみに今回は、Lambda関数を作成して、Amazon Bedrockに接続する形でLangChainを試します。

LangChainのドキュメントによると、Claude sonnet 4 がサポートされています。
ChatBedrock - Docs by LangChain

Lambdaレイヤーの作成

デフォルトのLambda関数だと、LangChainは使用できないので、Lambdaレイヤーを関数に追加してLangChainをLambda上で使えるようにしていきます。

レイヤーの作成方法は、以前に執筆した以下の記事を参照ください。

CloudShellからLambdaレイヤーを登録する - Qiita

Lambda関数の作成

Lambda用のIAMロール作成

Bedrock呼び出し用にIAMロールを作成する。今回はAmazonBedrockFullAccessを設定している。

また、LambdaからCloudWatch Logsにログを送信したいため、CloudWatchFullAccessも設定している。

まずは簡単なLambda関数の作成

先ほど作成したIAMロールをアタッチしたLambda関数を作成します。
ランタイム：Python 3.12
アーキテクチャ：x86_64
タイムアウト：5分（デフォの3秒だと生成AIからの回答を待っている間にタイムアウトになっちゃうので今回は5分に値にしておく）

レイヤーの追加

Lambda > 関数 > 作成した関数を選択 > コードタブの下部から「レイヤーの追加」をクリックします。

カスタムレイヤーを選択して、先ほど作成＆アップロード済みのレイヤーを選択します。

ChatBedrockConverseを試す

とりあえず生成AIを呼び出すシンプルなコードを書いてみます。

（返り値はAPI Gatewayを意識して記載しているだけなので、そこまで気にしないでください）

import json
import boto3
from botocore.exceptions import ClientError
from langchain_aws import ChatBedrockConverse

def lambda_handler(event, context):
    # モデルIDの指定
    model_id = "us.anthropic.claude-sonnet-4-20250514-v1:0"
    response_text = ""

    # ユーザーメッセージを取得
    user_message = event.get('userMessage')
    messages = [
        ("system", "あなたのタスクはユーザからの質問に答えることです。"),
        ("human", user_message),
    ]

    try:
        # ChatBedrockConverseモデルを初期化
        model = ChatBedrockConverse(
            model=model_id,
            region_name="us-east-1",
            max_tokens=4096,
        )
        
        # モデルを呼び出し
        response = model.invoke(messages)
        response_text = response.content
    
    except (ClientError, Exception) as e:
            return {
                'statusCode': 500,
                'body': json.dumps({'error': f"Can't invoke '{model_id}'. Reason: {str(e)}"})
            }
    
    # API Gatewayが期待する形式でレスポンスを返す
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',                 # レスポンスボディのデータ形式(今回はJSON)を指定
            'Access-Control-Allow-Origin': '*',                 # CORS（Cross-Origin Resource Sharing）対応
            'Access-Control-Allow-Methods': 'POST, OPTIONS',    # 許可するHTTPメソッドを指定
            'Access-Control-Allow-Headers': 'Content-Type'      # クライアントが送信可能なヘッダーを指定(Content-Typeはリクエストボディの形式指定を許可)
        },
        'body':response_text
    }

テストコードも用意します。

{
  "userMessage": "日本の首都はどこですか？"
}

結果は問題なしです。

DynamoDBに保存してみる

以下を参考にしてDynamoDBにチャット履歴を保存してみます。

DynamoDBテーブルの作成

テーブル名：BedrockConversationHistory
パーティションキー：SessionId（文字列）
テーブル設定：デフォルト設定

IAMロールの編集

LambdaにアタッチしているIAMロールに対して、DynamoDBを編集できる権限も追加しておきます。

Lambda関数の編集

以下のURLを参考に、Lambdaでコードを足していきます。

ソースコードは以下になります。

import json
import boto3
from botocore.exceptions import ClientError
from langchain_aws import ChatBedrockConverse
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_message_histories.dynamodb import DynamoDBChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

def lambda_handler(event, context):
    # モデルIDの指定
    model_id = "us.anthropic.claude-sonnet-4-20250514-v1:0"
    response_text = ""

    # ユーザーメッセージを取得
    userMessage = event.get('userMessage')
    # セッションIDを取得
    sessionId = event.get('sessionId')
    # プロンプト
    prompt = ChatPromptTemplate.from_messages([
        ("system", "あなたのタスクはユーザからの質問に答えることです。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{question}"),
    ])

    try:
        # ChatBedrockConverseモデルを初期化
        model = ChatBedrockConverse(
            model=model_id,
            region_name="us-east-1",
            max_tokens=4096,
        )

        chain = prompt | model
        chain_with_history = RunnableWithMessageHistory(
             chain,
             lambda session_id: DynamoDBChatMessageHistory(
                 table_name="BedrockConversationHistory",
                 session_id=session_id,
             ),
             input_messages_key="question",
             history_messages_key="history",
         )
        
        config = {"configurable": {"session_id": sessionId}}

        # モデルを呼び出し
        response = chain_with_history.invoke({"question": userMessage}, config=config)
        response_text = response.content
    
    except (ClientError, Exception) as e:
            return {
                'statusCode': 500,
                'body': json.dumps({'error': f"Can't invoke '{model_id}'. Reason: {str(e)}"})
            }
    
    # API Gatewayが期待する形式でレスポンスを返す
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',                 # レスポンスボディのデータ形式(今回はJSON)を指定
            'Access-Control-Allow-Origin': '*',                 # CORS（Cross-Origin Resource Sharing）対応
            'Access-Control-Allow-Methods': 'POST, OPTIONS',    # 許可するHTTPメソッドを指定
            'Access-Control-Allow-Headers': 'Content-Type'      # クライアントが送信可能なヘッダーを指定(Content-Typeはリクエストボディの形式指定を許可)
        },
        'body':response_text
    }

先ほどと同じようにLambdaでテストを実行したところ、DynamoDBにデータが格納されたことを確認しました。

コード解説

備忘録がてら解説をしてきます。

    from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

    # プロンプト
    prompt = ChatPromptTemplate.from_messages([
        ("system", "あなたのタスクはユーザからの質問に答えることです。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{question}"),
    ])

MessagesPlaceholder を利用すると、過去の会話を保存しておいてくれるようになります。variable_name で渡すデータのキー名を指定する必要があるようです。

chain = prompt | model

LangChain Expression Language（LCEL）を利用することで、複数のコンポーネント（ここでは prompt と model）を組み合わせて Chain を作っています。

from langchain_community.chat_message_histories.dynamodb import DynamoDBChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

chain_with_history = RunnableWithMessageHistory(
             chain,
             lambda session_id: DynamoDBChatMessageHistory(
                 table_name="BedrockConversationHistory",
                 session_id=session_id,
             ),
             input_messages_key="question",
             history_messages_key="history",
         )

RunnableWithMessageHistory は、会話履歴を扱うためのLangChainのコンポーネントとなります。役割としては、別のRunnableをラップし、そのRunnableに対するチャットメッセージ履歴の読み書き、更新を管理します。

内容については以下のサイトが非常に参考になりました。

https://blog.serverworks.co.jp/langchain-dynamodb-chat-history#RunnableWithMessageHistoryの動き

上記のサイトを踏まえて、各パラメータについても記載しておきます。

runnable ：
第一引数。会話履歴を考慮に入れたいRunnableオブジェクト。通常は、プロンプトテンプレートとLLMを組み合わせたチェーン（例：prompt | chat_model）が使用されます。
get_session_history ：
第二引数。会話履歴を取得・保存するための関数。セッションID（str）を引数にとり、BaseChatMessageHistoryのインスタンスを返します。ここでは DynamoDBChatMessageHistory を使用してDynamoDBから会話を取得・保存しています。

lambda session_id: DynamoDBChatMessageHistory(
                 table_name="BedrockConversationHistory",
                 session_id=session_id,
             )

DynamoDBChatMessageHistory では lambda式（無名関数）を定義しています。chain_with_history 呼び出し時の{'configurable': {'session_id': sessionId}} の引数を受け取るようになっているそうです。
要は以下のコード部分で指定している sessionId の値を渡してることになります。

config = {"configurable": {"session_id": sessionId}}
response = chain_with_history.invoke({"question": userMessage}, config=config)

input_messages_key には、chain_with_history呼び出し時の {"question": userMessage} の値が入ります。
history_messages_key に指定したKeyのValueにDynamoDBChatMessageHistoryクラスのインスタンスから取得した会話履歴が含まれるらしいので、プロンプトを定義した時の MessagesPlaceholderの variable_name と値を合わせる必要があります（以下の “history” ）。プロンプトが求めるキーに対して、history_messages_key の値を合わせなきゃいけない感じです。

# プロンプト
    prompt = ChatPromptTemplate.from_messages([
        ("system", "あなたのタスクはユーザからの質問に答えることです。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{question}"),
    ])

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up