【超初心者向け】AWS × AIエージェントのコードを1行ずつ徹底解説

musaka.k

はじめに

本記事ではこの記事で作成したソースコードを細かく解説するものです。

目次

  1. 全体像の把握
  2. tools.py
  3. agent.py
  4. lambda_function.py
  5. Dockerfile
  6. deploy.yml

全体像の把握

このアプリケーションは3つのPythonファイルで構成されています。

lambda_function.py  →  agent.py  →  tools.py
  • lambda_function.py: ユーザーからの質問を受け取り、回答を返す「受付係」
  • agent.py: AIが考えて、どの道具を使うか判断する「頭脳」
  • tools.py: 実際にAWSに問い合わせる「道具箱」

tools.py

まずは一番シンプルなtools.pyから見ていきましょう。このファイルには、AWSから情報を取得する関数が定義されています。

全体のコード

import boto3
from typing import Optional, List, Dict, Any
from langchain_core.tools import tool

@tool
def get_ec2_info(instance_id: Optional[str] = None, filter_key: Optional[str] = None, filter_value: Optional[str] = None) -> List[Dict[str, Any]]:
    """
    Retrieves information about EC2 instances.

    Args:
        instance_id: The ID of the specific instance to retrieve.
        filter_key: A filter key (e.g., 'tag:Name', 'instance-state-name').
        filter_value: The value for the filter key.

    Returns:
        A list of dictionaries containing instance details.
    """
    ec2 = boto3.client('ec2')

    filters = []
    if filter_key and filter_value:
        filters.append({'Name': filter_key, 'Values': [filter_value]})

    if instance_id:
        response = ec2.describe_instances(InstanceIds=[instance_id])
    elif filters:
        response = ec2.describe_instances(Filters=filters)
    else:
        response = ec2.describe_instances()

    instances = []
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            instances.append({
                'InstanceId': instance.get('InstanceId'),
                'InstanceType': instance.get('InstanceType'),
                'State': instance.get('State', {}).get('Name'),
                'PrivateIpAddress': instance.get('PrivateIpAddress'),
                'PublicIpAddress': instance.get('PublicIpAddress'),
                'VpcId': instance.get('VpcId'),
                'Tags': instance.get('Tags', [])
            })

    return instances

@tool
def get_vpc_info(vpc_id: Optional[str] = None) -> List[Dict[str, Any]]:
    """
    Retrieves information about VPCs.

    Args:
        vpc_id: The ID of the specific VPC to retrieve.

    Returns:
        A list of dictionaries containing VPC details.
    """
    ec2 = boto3.client('ec2')

    if vpc_id:
        response = ec2.describe_vpcs(VpcIds=[vpc_id])
    else:
        response = ec2.describe_vpcs()

    vpcs = []
    for vpc in response['Vpcs']:
        vpcs.append({
            'VpcId': vpc.get('VpcId'),
            'CidrBlock': vpc.get('CidrBlock'),
            'State': vpc.get('State'),
            'Tags': vpc.get('Tags', [])
        })

    return vpcs

1行ずつ解説

インポート部分

import boto3

boto3は、PythonからAWSのサービスを操作するための公式ライブラリです。これがあれば、PythonのコードからEC2やS3などのAWSサービスを自由に操作できます。
※Lambdaへのデプロイを想定していますので、Lambdaに適切な権限が付与されている前提です。

from typing import Optional, List, Dict, Any

typingモジュールから型ヒント用のクラスをインポートしています。

型 意味 例
Optional[str] 文字列か、None（何もない）のどちらか "i-12345" または None
List[Dict] 辞書のリスト（配列） [{"id": 1}, {"id": 2}]
Dict[str, Any] キーが文字列で、値は何でもOKな辞書 {"name": "test", "count": 5}

なぜこれらをインポートしているのかというと、以下ツール設計の推奨パターンによるものです。
・ユーザーの質問パターンを想像する
↓
・どんな情報を渡してくるか予測する
↓
・それを引数として定義する
AWSリソースに関する質問である以上は、インスタンスIDなどが質問文に入る可能性が高いです。
インスタンスIDを後続のツール関数の引数として定義するために、本モジュールをインポートしているわけです。

from langchain_core.tools import tool

LangChainというAIアプリケーション開発フレームワークから、toolデコレータをインポートしています。このデコレータを使うと、普通のPython関数をAIが使える「ツール」に変換できます。

@toolデコレータ

@tool
def get_ec2_info(...):

@toolはデコレータと呼ばれる特殊な書き方です。関数の上に@マークをつけて書くことで、その関数に特別な機能を追加します。

この場合、「この関数はAIエージェントが呼び出せるツールですよ」とマークしています。AIは関数名とdocstring（説明文）を読んで、いつこのツールを使うべきかを自動的に判断します。

関数の引数（詳細解説）

def get_ec2_info(instance_id: Optional[str] = None, filter_key: Optional[str] = None, filter_value: Optional[str] = None) -> List[Dict[str, Any]]:

この1行には多くの情報が詰まっています。

引数の詳細

instance_id: Optional[str] = None の分解：

部分 意味
instance_id 引数名（変数名）
: 型ヒントの開始を示す記号
Optional[str] 「文字列（str）か、None のどちらか」を受け取れる
= None デフォルト値。引数を省略した場合は None が入る

Optional[str] とは？

  • str = 文字列型（例："i-0123456789abcdef0"）
  • Optional[X] = 「X型 または None」という意味
  • つまり Optional[str] = str | None（文字列かNone）

使用例：

# 引数を指定する場合
get_ec2_info(instance_id="i-0123456789abcdef0")

# 引数を省略すると None になる
get_ec2_info()  # instance_id は None

3つの引数の役割：

引数 説明 例
instance_id 特定のインスタンスIDを指定 "i-0123456789abcdef0"
filter_key フィルタの種類（タグ名や状態名） "tag:Name", "instance-state-name"
filter_value フィルタの値 "production", "running"
-> – 戻り値の型ヒント開始

「この関数は何を返すか」を示す記号です。

List[Dict[str, Any]] – 戻り値の型
部分 意味
List[...] リスト（配列）を返す
Dict[str, Any] キーが文字列(str)、値は何でもOK(Any)の辞書

つまり： 辞書のリストを返します。

戻り値の具体例：

[
    {
        "InstanceId": "i-0123456789abcdef0",  # str
        "InstanceType": "t2.micro",           # str
        "State": "running",                   # str
        "PrivateIpAddress": "10.0.0.5",       # str
        "Tags": [{"Key": "Name", "Value": "web-server"}]  # list
    },
    {
        "InstanceId": "i-abcdef1234567890",
        # ... 他のインスタンス情報
    }
]
全体の構造図

整理すると以下のようになります。

def get_ec2_info(instance_id: Optional[str] = None, ...) -> List[Dict[str, Any]]:
    │              │         │            │   │        │    └── 戻り値の型
    │              │         │            │   │        └── 戻り値の型ヒント開始
    │              │         │            │   └── デフォルト値
    │              │         │            └── 代入演算子
    │              │         └── 型ヒント（str または None）
    │              └── 引数名
    └── 関数定義キーワード

docstring（説明文）

    """
    Retrieves information about EC2 instances.

    Args:
        instance_id: The ID of the specific instance to retrieve.
        ...
    """

"""で囲まれた部分はdocstringと呼ばれ、関数の説明を書きます。

重要: AIエージェントはこのdocstringを読んで、「この関数は何をするのか」「いつ使うべきか」を判断します。英語で書いているのは、AIモデル（GPT）が英語の方が理解しやすいためです。

boto3クライアントの作成

    ec2 = boto3.client('ec2')

AWSのEC2サービスと通信するための「クライアント」を作成しています。これを使って、EC2に関する様々な操作（インスタンス一覧の取得、起動、停止など）ができます。

フィルタの構築

    filters = []
    if filter_key and filter_value:
        filters.append({'Name': filter_key, 'Values': [filter_value]})

AWS APIにはフィルタ機能があります。例えば「稼働中のインスタンスだけ取得」といった絞り込みができます。

filter_keyとfilter_valueの両方が指定されている場合のみ、フィルタ条件を作成します。

例
{'Name': 'instance-state-name', 'Values': ['running']}

↑ これは「状態がrunningのものだけ」という条件を意味します。

AWS APIの呼び出し

    if instance_id:
        response = ec2.describe_instances(InstanceIds=[instance_id])
    elif filters:
        response = ec2.describe_instances(Filters=filters)
    else:
        response = ec2.describe_instances()

条件に応じて、3パターンの呼び出し方を使い分けています。

条件 動作
instance_idが指定されている そのIDのインスタンスだけ取得
フィルタが指定されている フィルタ条件に合うものだけ取得
何も指定されていない すべてのインスタンスを取得

レスポンスの整形

    instances = []
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            instances.append({
                'InstanceId': instance.get('InstanceId'),
                'InstanceType': instance.get('InstanceType'),
                'State': instance.get('State', {}).get('Name'),
                'PrivateIpAddress': instance.get('PrivateIpAddress'),
                'PublicIpAddress': instance.get('PublicIpAddress'),
                'VpcId': instance.get('VpcId'),
                'Tags': instance.get('Tags', [])
            })

AWSからのレスポンスは以下のような階層構造になっています。
そのため2重ループをして、最終的にフラットなリストに変換しています。

# 実際のAWSレスポンス（イメージ）
response = {
    'Reservations': [
        {
            'ReservationId': 'r-xxxxx',
            'Instances': [
                {'InstanceId': 'i-111', 'InstanceType': 't2.micro', ...},
                {'InstanceId': 'i-222', 'InstanceType': 't2.small', ...}
            ]
        },
        {
            'ReservationId': 'r-yyyyy',
            'Instances': [
                {'InstanceId': 'i-333', 'InstanceType': 't3.medium', ...}
            ]
        }
    ]
}

# 整形後
instances = [
    {'InstanceId': 'i-111', ...},
    {'InstanceId': 'i-222', ...},
    {'InstanceId': 'i-333', ...}
]

instance.get('InstanceId')という書き方は、辞書から値を安全に取得する方法です。キーが存在しない場合はNoneを返します（エラーにならない）。

instance.get('State', {}).get('Name')は、ネストした辞書から値を取得しています。Stateがない場合は空の辞書{}を使い、その後の.get('Name')でも安全に処理できます。

agent.py

次に、AIエージェントの中核となるagent.pyを見ていきましょう。

全体のコード

from typing import TypedDict, Annotated, List
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import BaseMessage, HumanMessage
import operator
from tools import get_ec2_info, get_vpc_info

# Define the state
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

# Initialize tools
tools = [get_ec2_info, get_vpc_info]

# Initialize LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# Define nodes
def agent(state: AgentState):
    messages = state['messages']
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

# Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", agent)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")

def should_continue(state: AgentState):
    messages = state['messages']
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

workflow.add_conditional_edges(
    "agent",
    should_continue,
)

workflow.add_edge("tools", "agent")

app = workflow.compile()

def run_agent(query: str):
    """
    Runs the agent with a given query.
    """
    inputs = {"messages": [HumanMessage(content=query)]}
    result = app.invoke(inputs)

    last_message = result['messages'][-1]
    return last_message.content

1行ずつ解説

インポート部分

from typing import TypedDict, Annotated, List
クラス 説明
TypedDict 辞書の各キーにどんな型の値が入るか定義できる
Annotated 型に追加情報（メタデータ）をつけられる
List リスト（配列）の型 
from langchain_openai import ChatOpenAI

OpenAIのチャットモデル（GPT-3.5、GPT-4など）を使うためのクラスです。

from langgraph.graph import StateGraph, END

LangGraphはLangChainの一部で、AIエージェントの処理フローをグラフ（ノードとエッジの組み合わせ）として定義できます。

  • StateGraph: 状態を持つグラフを作成するクラス
  • END: グラフの終了を表す特別な値
from langgraph.prebuilt import ToolNode

ツールを実行するためのノード（処理単位）です。LangGraphがあらかじめ用意してくれています。

from langchain_core.messages import BaseMessage, HumanMessage

チャットのメッセージを表すクラスです。

  • BaseMessage: すべてのメッセージの基底クラス
  • HumanMessage: ユーザー（人間）からのメッセージ
import operator

Pythonの標準ライブラリで、演算子を関数として使えます。ここではoperator.add（足し算）を使います。

from tools import get_ec2_info, get_vpc_info

さきほど解説したtools.pyから、2つの関数をインポートしています。

状態の定義

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

エージェントの使いまわしできる入れ物を定義しています。
またmessagesキーには、BaseMessageまたはその子クラスのインスタンスだけがリストで入るという型ヒントを定義しています。
この入れ物は処理の途中で更新されながら引き継がれます。

複雑なので細かく説明します。

まずTypedDictクラスを継承し、AgentStateクラスを作成しています。
これにより、キーと値の型を固定することができます。

以下TypeDictの使い方の例

from typing import TypedDict, List

# 通常の辞書（何でも入る）
data = {}
data["name"] = "Alice"
data["age"] = 25
data[123] = "何でもOK"  # キーが数字でもOK
data["age"] = "文字列もOK"  # 型も自由

# TypedDictを使った辞書（型が決まっている）
class UserState(TypedDict):
    name: str
    age: int
    hobbies: List[str]

user = UserState(name="Alice", age=25, hobbies=["読書", "映画"])
user["name"] = "Bob"          # ✅ OK（strだから）
user["age"] = 30              # ✅ OK（intだから）
user["age"] = "三十"          # ❌ IDEが警告（intじゃない）
user["other"] = "something"   # ❌ IDEが警告（定義にないキー）

そして、List[BaseMessage]はBaseMessage型（またはその子クラス）の要素だけを持つリストを意味します。

List[BaseMessage]
│     └── 要素の型（BaseMessageまたはその子クラス）
└── リスト型

BaseMessageとは
LangChainドキュメントによると、Base class for all types of messages in a conversation
（会話におけるすべてのメッセージ種類の基底クラス）となり、チャットのメッセージを表す親クラスです。

BaseMessage（抽象クラス）
    │
    ├── HumanMessage      # ユーザーのメッセージ
    ├── AIMessage         # LLMの応答
    ├── SystemMessage     # システム指示
    ├── ToolMessage       # ツール実行結果
    ├── FunctionMessage   # 関数呼び出し結果（旧式）
    ├── ChatMessage       # 汎用チャットメッセージ
    └── RemoveMessage     # メッセージ削除用

具体例を用いて説明します。
よくあるチャットボットとの会話は、上記クラスのcontentに格納されています。

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

# HumanMessage（BaseMessageの子クラス）
human_msg = HumanMessage(content="EC2の情報を教えて")
print(human_msg.content)  # "EC2の情報を教えて"
print(human_msg.type)     # "human"

# AIMessage（BaseMessageの子クラス）
ai_msg = AIMessage(content="EC2インスタンスは3つあります")
print(ai_msg.content)     # "EC2インスタンスは3つあります"
print(ai_msg.type)        # "ai"

# ToolMessage（BaseMessageの子クラス）
tool_msg = ToolMessage(
    content='{"InstanceId": "i-123"}',
    tool_call_id="call_abc123"
)
print(tool_msg.content)   # '{"InstanceId": "i-123"}'
print(tool_msg.type)      # "tool"

つまり、messagesキーには、BaseMessageまたはその子クラスのインスタンスだけがリストで入ることになります。
※あくまで型ヒントなので、実行時に強制されるわけでありません。

messages: List[BaseMessage] = [
    HumanMessage(content="質問"),      # ✅ OK（子クラス）
    AIMessage(content="回答"),         # ✅ OK（子クラス）
    ToolMessage(content="結果", ...),  # ✅ OK（子クラス）
    SystemMessage(content="指示"),     # ✅ OK（子クラス）
]

messages: List[BaseMessage] = [
    "ただの文字列",                    # ❌ NG（strはBaseMessageじゃない）
    123,                              # ❌ NG（intはBaseMessageじゃない）
    {"key": "value"},                 # ❌ NG（dictはBaseMessageじゃない）
    SomeOtherClass(),                 # ❌ NG（無関係なクラス）
]

Annotated[List[BaseMessage], operator.add]の細かな文法説明は省きます（というかすみません、私も若干理解が怪しいです。）

公式のLangGraph Quickstartでも使用されてますので、一般的な状態の定義方法と覚えてしまってよいかと思います。

このmessageに以下のように会話履歴がたまっていくことになります。

# 現在の状態
state = {"messages": [HumanMessage("こんにちは")]}

# agentノードが返す値
return {"messages": [AIMessage("はい、何でしょう？")]}

# 結果（operator.addで連結）
state = {"messages": [
    HumanMessage("こんにちは"),
    AIMessage("はい、何でしょう？")  # 追加される
]}

ツールとLLMの初期化

tools = [get_ec2_info, get_vpc_info]

使用するツールをリストにまとめます。

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

OpenAIのGPT-3.5モデルを使うLLM（大規模言語モデル）を作成します。

  • model="gpt-3.5-turbo": 使用するモデル名
  • temperature=0: 出力のランダム性。0だと最も確定的な回答になります
llm_with_tools = llm.bind_tools(tools)

LLMにツールを紐づけます。これにより、LLMは「このツールを使いたい」と判断したとき、適切な形式でツール呼び出しを返せるようになります。

エージェントノードの定義

def agent(state: AgentState):
    messages = state['messages']
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

エージェントノードは、現在のメッセージ履歴をLLMに渡し、回答を取得します。

  1. state['messages']で今までのメッセージ履歴を取得
  2. llm_with_tools.invoke(messages)でLLMに問い合わせ
  3. 結果を新しいメッセージとして返す

ちなみにこのstateですが、公式ドキュメントによるとフレームワーク側で作成・管理されており、関数に渡されます。

グラフの構築

workflow = StateGraph(AgentState)

AgentState で定められた「状態の形と更新ルール」に従って、StateGraph がワークフローを実行します。

初期状態: messages = [HumanMessage("EC2の情報を教えて")]
    ↓ agentノード実行
状態更新: messages = [HumanMessage(...), AIMessage(tool_calls=[...])]
    ↓ toolsノード実行
状態更新: messages = [..., AIMessage(...), ToolMessage(結果)]
    ↓ agentノード実行
最終状態: messages = [..., ..., ..., AIMessage("EC2は3台あります")]

このワークフローは動きとしては、上記のようなイメージです。
なお、以降の説明でノード（node）とエッジ（Edge）という言葉が登場しますが、
エッジ（Edge）：何をするかを定義
エッジ（Edge）：どこへ行くかを定義
というイメージです。

workflow.add_node("agent", agent)
workflow.add_node("tools", ToolNode(tools))

グラフに2つのノード（処理単位）を追加します。ワークフロー実行のためにはまずはノードをワークフローに追加する必要があります。

  • "agent": AIが考えるノード
  • "tools": ツールを実行するノード
workflow.set_entry_point("agent")

処理の開始点を"agent"ノードに設定します。

条件分岐の定義

def should_continue(state: AgentState):
    messages = state['messages']
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

この関数は、次にどのノードに進むかを決めます。

  • LLMの回答にtool_calls（ツール呼び出し要求）があれば → "tools"ノードへ
  • なければ → END（終了）
workflow.add_conditional_edges(
    "agent",
    should_continue,
)

"agent"ノードの後に、should_continue関数で次の行き先を決める条件分岐を追加します。
add_conditional_edgesは第一引数が分岐元のノード名で、第二引数が分岐先を決める関数名です。
本来は第三引数に関数の戻り値と次に呼び出すノードのマッピングを記載しますが、両者が一致しているため、省略しています。

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",  # "tools" → toolsノード
        END: END           # END → 終了
    }
)

上記は省略しないバージョンです。

workflow.add_edge("tools", "agent")

add_edgeでノード
"tools"ノードの後は、必ず"agent"ノードに戻ります。これにより、ツール実行結果をAIが確認し、追加のツール呼び出しが必要か判断できます。

グラフのコンパイル

app = workflow.compile()

定義したグラフを実行可能な形にコンパイルします。

実行関数

def run_agent(query: str):
    inputs = {"messages": [HumanMessage(content=query)]}
    result = app.invoke(inputs)

    last_message = result['messages'][-1]
    return last_message.content

外部から呼び出すための関数です。
AgentStateの形式に合わせた辞書を作成します。

query = "EC2インスタンスを教えて"
↓
inputs = {"messages": [HumanMessage(content="EC2インスタンスを教えて")]}

このようなイメージです。

last_message = result['messages'][-1]
return last_message.content

最終回答を抽出します。result['messages'] には全会話履歴が蓄積されており、最後のメッセージ（[-1]）がLLMの最終回答となります。

lambda_function.py

AWS Lambdaのエントリーポイントとなるファイルです。

全体のコード

import json
from agent import run_agent

def lambda_handler(event, context):
    """
    AWS Lambda handler function.
    Expects an event with a 'query' key.
    """
    try:
        if 'body' in event:
            try:
                body = json.loads(event['body'])
                query = body.get('query')
            except (TypeError, json.JSONDecodeError):
                query = event.get('body', {}).get('query') if isinstance(event.get('body'), dict) else None
        else:
            query = event.get('query')

        if not query:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'No query provided'})
            }

        response = run_agent(query)

        return {
            'statusCode': 200,
            'body': json.dumps({'response': response})
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

1行ずつ解説

インポート

import json
from agent import run_agent
  • json: JSONの読み書きをするPython標準ライブラリ
  • run_agent: さきほど解説した、エージェントを実行する関数

Lambda ハンドラー

def lambda_handler(event, context):

AWS Lambdaのお約束: 関数名はlambda_handlerで、引数はeventとcontextの2つ。

  • event: Lambdaに渡されたデータ（質問など）
  • context: 実行環境の情報（タイムアウト時間など）

リクエストの解析

        if 'body' in event:
            try:
                body = json.loads(event['body'])
                query = body.get('query')
            except (TypeError, json.JSONDecodeError):
                query = event.get('body', {}).get('query') if isinstance(event.get('body'), dict) else None
        else:
            query = event.get('query')

Lambdaは2つの呼び出し方があるため、両方に対応しています。

パターン1: API Gateway経由

{
  "body": "{\"query\": \"EC2一覧を教えて\"}"
}

bodyは文字列なので、json.loads()でパースが必要。

パターン2: 直接呼び出し

{
  "query": "EC2一覧を教えて"
}

直接queryキーにアクセスできる。

バリデーション

        if not query:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'No query provided'})
            }

質問がなければ、HTTPステータス400（Bad Request）を返します。

エージェントの実行

        response = run_agent(query)

        return {
            'statusCode': 200,
            'body': json.dumps({'response': response})
        }

エージェントを実行し、結果をHTTPステータス200（成功）で返します。

エラーハンドリング

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

予期しないエラーが発生した場合、HTTPステータス500（サーバーエラー）を返します。

Dockerfile

Dockerfileは、アプリケーションを動かすための環境を定義するファイルです。

全体のコード

FROM public.ecr.aws/lambda/python:3.12

COPY requirements.txt ${LAMBDA_TASK_ROOT}

RUN pip install -r requirements.txt

COPY lambda_function.py ${LAMBDA_TASK_ROOT}
COPY agent.py ${LAMBDA_TASK_ROOT}
COPY tools.py ${LAMBDA_TASK_ROOT}

CMD [ "lambda_function.lambda_handler" ]

1行ずつ解説

ベースイメージ

FROM public.ecr.aws/lambda/python:3.12

FROMは、どのイメージをベースにするか指定します。

public.ecr.aws/lambda/python:3.12は、AWSが公式に提供しているLambda用のPython 3.12イメージです。これを使うことで、Lambda環境と同じ環境でコンテナを作れます。

依存関係のコピー

COPY requirements.txt ${LAMBDA_TASK_ROOT}

COPYは、ローカルのファイルをコンテナ内にコピーします。

${LAMBDA_TASK_ROOT}は、Lambdaベースイメージで定義されている環境変数で、Lambdaのコードを置くディレクトリ（/var/task）を指します。

依存関係のインストール

RUN pip install -r requirements.txt

RUNは、コンテナ内でコマンドを実行します。

ここでは、pip install -r requirements.txtで必要なPythonパッケージ（boto3、langchainなど）をインストールしています。

アプリケーションコードのコピー

COPY lambda_function.py ${LAMBDA_TASK_ROOT}
COPY agent.py ${LAMBDA_TASK_ROOT}
COPY tools.py ${LAMBDA_TASK_ROOT}

3つのPythonファイルをコンテナにコピーします。

エントリーポイント

CMD [ "lambda_function.lambda_handler" ]

CMDは、コンテナ起動時に実行するコマンドを指定します。

"lambda_function.lambda_handler"は、「lambda_function.pyファイルのlambda_handler関数を実行する」という意味です。これはLambdaの規約に従った書き方です。

deploy.yml

GitHub Actionsのワークフローファイルです。コードをプッシュすると自動でデプロイされます。

全体のコード

name: Deploy to AWS Lambda test4

on:
  push:
    branches:
      - master
      - main
    paths-ignore:
      - '**.md'
      - 'docs/**'

permissions:
  id-token: write
  contents: read

env:
  AWS_REGION: ${{ secrets.AWS_REGION }}
  ECR_REPOSITORY: ${{ secrets.ECR_REPOSITORY }}
  LAMBDA_FUNCTION_NAME: ${{ secrets.LAMBDA_FUNCTION_NAME }}
  AWS_ROLE_ARN: ${{ secrets.AWS_ROLE_ARN }}

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Debug OIDC Context
        run: |
          echo "GitHub Repository: ${{ github.repository }}"
          echo "GitHub Ref: ${{ github.ref }}"
          echo "GitHub Actor: ${{ github.actor }}"
          echo "Please verify these values match your IAM Role Trust Policy."

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ env.AWS_ROLE_ARN }}
          aws-region: ${{ env.AWS_REGION }}

      - name: Login to Amazon ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v2

      - name: Build, tag, and push image to Amazon ECR
        id: build-image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
          IMAGE_TAG: ${{ github.sha }}
        run: |
          docker build --platform linux/amd64 --provenance=false -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
          docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
          echo "image=$ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG" >> $GITHUB_OUTPUT

      - name: Update Lambda Function
        env:
          IMAGE_URI: ${{ steps.build-image.outputs.image }}
        run: |
          aws lambda update-function-code \
            --function-name $LAMBDA_FUNCTION_NAME \
            --image-uri $IMAGE_URI

動きとしては
・master/main に push
・コードをチェックアウト
・OIDC認証でAWSにログイン
・ECR（コンテナ保管庫）にログイン
・Dockerイメージをビルド → ECRにプッシュ
・Lambda関数を新しいイメージで更新
という感じです。

1行ずつ解説

ワークフロー名

name: Deploy to AWS Lambda test4

GitHubのActionsタブに表示される名前です。

トリガー条件

on:
  push:
    branches:
      - master
      - main
    paths-ignore:
      - '**.md'
      - 'docs/**'

いつこのワークフローを実行するかを定義しています。

設定 意味
push: プッシュされたとき
branches: [master, main] masterまたはmainブランチへのプッシュのみ
paths-ignore: これらのファイルが変更されただけでは実行しない
**.md すべてのMarkdownファイル
docs/** docsフォルダ以下のすべて

READMEを更新しただけでデプロイが走らないようにしています。

パーミッション

permissions:
  id-token: write
  contents: read

OIDC認証に必要な権限を設定しています。アクセスキーを使用しないセキュアな認証方法となります。

  • id-token: write: OIDC トークンの発行を許可
  • contents: read: リポジトリの内容を読み取り許可

環境変数

env:
  AWS_REGION: ${{ secrets.AWS_REGION }}
  ECR_REPOSITORY: ${{ secrets.ECR_REPOSITORY }}
  LAMBDA_FUNCTION_NAME: ${{ secrets.LAMBDA_FUNCTION_NAME }}
  AWS_ROLE_ARN: ${{ secrets.AWS_ROLE_ARN }}

GitHubのSecretsに保存した値を環境変数として使用します。
OIDC認証ではAWS側で定義したロールにスイッチして、AWSリソースの操作を行います。
そのため、ここでロールを指定しています。

ジョブの定義

jobs:
  deploy:
    runs-on: ubuntu-latest
  • jobs:: 実行するジョブ（処理のまとまり）を定義
  • deploy:: ジョブの名前
  • runs-on: ubuntu-latest: 最新のUbuntu環境で実行

ステップ1: コードのチェックアウト

      - name: Checkout code
        uses: actions/checkout@v4

リポジトリのコードを取得します。actions/checkoutはGitHub公式のアクションです。

ステップ2: デバッグ情報の出力

      - name: Debug OIDC Context
        run: |
          echo "GitHub Repository: ${{ github.repository }}"
          #リポジトリの「オーナー/リポジトリ名」
          echo "GitHub Ref: ${{ github.ref }}"
          #ブランチまたはタグのフルパス
          echo "GitHub Actor: ${{ github.actor }}"
          #pushを実行したユーザー名
          echo "Please verify these values match your IAM Role Trust Policy."

GitHub上でログが確認できるのでトラブルシューティング用に、GitHub Actionsの実行コンテキスト情報を出力します。
${{ github.xxx }}はGitHubが自動的に設定する変数で、リポジトリ名やブランチ名などが入っています。

ステップ3: AWS認証

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ env.AWS_ROLE_ARN }}
          aws-region: ${{ env.AWS_REGION }}

OIDC認証でAWSにログインします。

従来のアクセスキー方式と違い、一時的なトークンを使用するため、より安全です。

role-to-assumeで指定したIAMロールの権限を一時的に取得します。

ステップ4: ECRログイン

      - name: Login to Amazon ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v2

Amazon ECR（Elastic Cntainer Registry） にログインします。ECRはAWSのDockerイメージ保管サービスです。
前のステップでAWS認証済なので、スイッチ先のロールに適切な権限があれば成功します。
id: login-ecrを設定することで、後のステップでこのステップの出力を参照できます。

ステップ5: イメージのビルド＆プッシュ

      - name: Build, tag, and push image to Amazon ECR
        id: build-image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
          IMAGE_TAG: ${{ github.sha }}
        run: |
          docker build --platform linux/amd64 --provenance=false -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
          docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
          echo "image=$ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG" >> $GITHUB_OUTPUT

Dockerイメージをビルドして、ECRにプッシュします。

オプション 説明
--platform linux/amd64 x86_64アーキテクチャ向けにビルド（Lambdaの動作環境に合わせる）
--provenance=false メタデータを無効化（Lambdaでエラーになるのを防ぐ）
-t $ECR_REGISTRY/... イメージにタグ（名前）をつける

ECR_REGISTRYとECR_REPOSITORYの関係性は以下になります。

123456789012.dkr.ecr.us-east-1.amazonaws.com/aws-rag-agent:a1b2c3d
└──────────── ECR_REGISTRY ────────────────┘└ECR_REPOSITORY┘:└TAG┘

なお、Dockerfileは現在のディレクトリにある必要があります。

docker build ... .
              └── ここでDockerfileを探し、
                  ここにあるファイルをビルドに使う

${{ github.sha }}はコミットのハッシュ値で、イメージのバージョン識別に使用します。

echo "image=..." >> $GITHUB_OUTPUTで、次のステップにイメージURIを渡しています。

ステップ6: Lambda更新

      - name: Update Lambda Function
        env:
          IMAGE_URI: ${{ steps.build-image.outputs.image }}
        run: |
          aws lambda update-function-code \
            --function-name $LAMBDA_FUNCTION_NAME \
            --image-uri $IMAGE_URI

AWS CLIを使って、Lambda関数のコードを新しいイメージに更新します。

${{ steps.build-image.outputs.image }}で、前のステップで保存したイメージURIを取得しています。

まとめ

この記事では、以下のファイルを初心者向けに解説しました。

ファイル 役割
tools.py AWSと通信する関数（EC2/VPC情報取得）
agent.py AIエージェントの思考ループを定義
lambda_function.py Lambdaのエントリーポイント
Dockerfile コンテナ環境の定義
deploy.yml GitHub Actionsで自動デプロイ

これらを組み合わせることで、「自然言語でAWSインフラを検索できるAIエージェント」が完成します。

各ファイルの役割を理解することで、カスタマイズや機能追加も容易になります。ぜひ自分のプロジェクトに活用してみてください。

参考リンク

Last modified: 2025-12-13

Author