workflow 型の Agent (LangGraph) を Lambda でストリーミング実行し，Reactアプリで表示する

import os

import uvicorn
from agent import stream_graph
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

# CORSミドルウェアの設定 (本番環境では適切なオリジンを指定してください)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class Input(BaseModel):
    product_detail: str


@app.post("/api/stream_graph")
def api_stream_graph(input: Input):
    return StreamingResponse(
        stream_graph(input.product_detail), media_type="application/json; charset=utf-8"
    )


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8080"))) # Default port of LWA is 8080

import json
from typing import Annotated, Any, Generator

from langchain_aws import ChatBedrock
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import START, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.types import Command
from typing_extensions import TypedDict

llm = ChatBedrock(
    model="us.anthropic.claude-3-5-haiku-20241022-v1:0",
    region="us-west-2",
)


# Define graph state
class State(TypedDict):
    product_detail: Annotated[str, "製品詳細"]
    copy: Annotated[str, "コピー"]
    target_audience: Annotated[str, "ターゲット顧客"]


# Define the nodes
def generate_copy(state: State) -> Command:
    print("Called generate_copy")
    human_prompt = """
    以下の製品情報から，商品の広告文を作成しなさい．最終的な広告文を1つのみ出力すること．
    <product_details>
    {product_detail}
    </product_details>
    """
    prompt = ChatPromptTemplate([("human", human_prompt)])
    chain = prompt | llm
    response = chain.invoke({"product_detail": state["product_detail"]})

    return Command(
        update={"copy": response.content},
        goto="analysis_target_audience",
    )


def analysis_target_audience(state: State) -> Command:
    print("Called analysis_target_audience")
    human_prompt = """
    以下の広告文に基づいて，最適なターゲット層を分析してください．最終的なターゲット層のみ出力すること．
    <copy>
    {copy}
    </copy>
    """
    prompt = ChatPromptTemplate([("human", human_prompt)])
    chain = prompt | llm
    response = chain.invoke({"copy": state["copy"]})

    return Command(
        update={"target_audience": response.content},
    )


def build_graph() -> CompiledStateGraph:
    builder = StateGraph(State)
    builder.add_edge(START, "generate_copy")
    builder.add_node("generate_copy", generate_copy)
    builder.add_node("analysis_target_audience", analysis_target_audience)
    return builder.compile()


def stream_graph(input: str) -> Generator[str, Any, None]:
    print("Starting graph")
    graph = build_graph()
    stream = graph.stream(input={"product_detail": input}, stream_mode="values")
    for state in stream:
        yield json.dumps(state, ensure_ascii=False) + "\n"

import * as cdk from 'aws-cdk-lib';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as iam from 'aws-cdk-lib/aws-iam';
import { Construct } from 'constructs';

export class FastapiLambdaWebAdapterCdkStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Lambda関数の作成
    const fastApiFunction = new lambda.DockerImageFunction(this, 'FastAPIFunction', {
      code: lambda.DockerImageCode.fromImageAsset('./app', {
        file: 'Dockerfile',
      }),
      memorySize: 512,
      timeout: cdk.Duration.seconds(300),
      tracing: lambda.Tracing.ACTIVE,
    });

    // Bedrock用のIAMポリシーを追加
    fastApiFunction.addToRolePolicy(new iam.PolicyStatement({
      effect: iam.Effect.ALLOW,
      actions: ['bedrock:InvokeModel'],
      resources: ['*'],
      sid: 'BedrockInvokePolicy',
    }));

    // Function URLの設定
    const functionUrl = fastApiFunction.addFunctionUrl({
      authType: lambda.FunctionUrlAuthType.NONE,
      invokeMode: lambda.InvokeMode.RESPONSE_STREAM,
    });

    // 出力の設定
    new cdk.CfnOutput(this, 'FastAPIFunctionUrl', {
      description: 'Function URL for FastAPI function',
      value: functionUrl.url,
    });

    new cdk.CfnOutput(this, 'FastAPIFunctionArn', {
      description: 'FastAPI Lambda Function ARN',
      value: fastApiFunction.functionArn,
    });
  }
}

import json

import requests
import streamlit as st

# APIのエンドポイントURL
LAMBDA_URL = "https://XXXXXXXXXXXXXXXXXXXXXXXXXXXXX.lambda-url.ap-northeast-1.on.aws/"
ENDPOINT = "api/stream_graph"
API_URL = LAMBDA_URL + ENDPOINT

# ページ設定
st.set_page_config(
    page_title="商品広告生成アプリ",
    page_icon="🤖",
    layout="wide",
)

# タイトルと説明
st.title("商品広告生成アプリ")
st.markdown("本アプリは、製品情報から広告文を生成し、最適なターゲット層を分析します。")

# 製品情報入力欄
product_detail = st.text_area(
    "製品情報を入力してください",
    height=150,
    placeholder="例：ラベンダーとベルガモットのやさしい香りが特徴の保湿クリームで、ヒアルロン酸とシアバターの配合により、乾燥肌に潤いを与えます。",
)

# 実行ボタン
if st.button("Agent 実行開始", type="primary", disabled=not product_detail):
    # 結果表示用のプレースホルダー
    copy_placeholder = st.empty()
    target_placeholder = st.empty()

    try:
        # APIリクエストを送信
        with st.spinner("実行中..."):
            response = requests.post(
                API_URL,
                json={"product_detail": product_detail},
                stream=True,
                headers={"Accept": "application/json"},
            )

            if response.status_code == 200:
                # ストリーミングレスポンスを処理
                for line in response.iter_lines():
                    if line:
                        # JSONデータをデコード
                        state = json.loads(line.decode("utf-8"))

                        # コピーが生成された場合
                        if "copy" in state:
                            with copy_placeholder.container():
                                st.subheader("📝 生成された広告文")
                                st.info(state["copy"])

                        # ターゲット層が分析された場合
                        if "target_audience" in state:
                            with target_placeholder.container():
                                st.subheader("🎯 ターゲット層分析")
                                st.success(state["target_audience"])
            else:
                st.error(f"APIエラー: ステータスコード {response.status_code}")

    except requests.exceptions.RequestException as e:
        st.error(f"接続エラー: {str(e)}")
    except json.JSONDecodeError as e:
        st.error(f"JSONデコードエラー: {str(e)}")
    except Exception as e:
        st.error(f"予期せぬエラーが発生しました: {str(e)}")

// API設定
export const API_CONFIG = {
    LAMBDA_URL: "https://XXXXXXXXXXXXXXXXXXXXXXXXXXXXX.lambda-url.ap-northeast-1.on.aws/",
    ENDPOINT: "api/stream_graph",
    get API_URL() {
      return `${this.LAMBDA_URL}${this.ENDPOINT}`;
    },
    HEADERS: {
      'Content-Type': 'application/json',
      'Accept': 'text/event-stream'
    }
  };

import React from 'react';
import useProductAnalysis from './hooks/useProductAnalysis';
import ProductForm from './components/ProductForm';
import ResultCard from './components/Result/ResultCard';
import './styles/components/App.css';

const App = () => {
  const {
    copy,
    targetAudience,
    isLoading,
    error,
    analyzeProduct
  } = useProductAnalysis();

  return (
    <div className="container">
      <div className="card">
        <h1 className="title">
          <span className="title-highlight">商品広告生成アプリ</span>
        </h1>
        <h2 className="subtitle">商品詳細</h2>

        <ProductForm
          onSubmit={analyzeProduct}
          isLoading={isLoading}
        />

        {error && (
          <div className="error">
            {error}
          </div>
        )}

        {copy && (
          <ResultCard
            title="生成された広告文"
            content={copy}
            emoji="📝"
          />
        )}

        {targetAudience && (
          <ResultCard
            title="ターゲット層分析"
            content={targetAudience}
            emoji="🎯"
            isTargetAudience={true}
          />
        )}
      </div>
    </div>
  );
};

export default App;

import { useState } from 'react';
import { API_CONFIG } from '../config/api';

const useProductAnalysis = () => {
  const [copy, setCopy] = useState('');
  const [targetAudience, setTargetAudience] = useState('');
  const [isLoading, setIsLoading] = useState(false);
  const [error, setError] = useState(null);

  const analyzeProduct = async (productDetail) => {
    setIsLoading(true);
    setCopy('');
    setTargetAudience('');
    setError(null);

    try {
      const response = await fetch(API_CONFIG.API_URL, {
        method: 'POST',
        headers: API_CONFIG.HEADERS,
        credentials: 'omit',
        mode: 'cors',
        body: JSON.stringify({ product_detail: productDetail })
      });

      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }

      await processStream(response.body.getReader());
    } catch (error) {
      console.error('Error:', error);
      setError(error.message || '予期せぬエラーが発生しました');
    } finally {
      setIsLoading(false);
    }
  };

  const processStream = async (reader) => {
    const decoder = new TextDecoder();
    let buffer = '';

    try {
      while (true) {
        const { done, value } = await reader.read();

        if (done) {
          processBuffer(buffer);
          break;
        }

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        lines.forEach(line => processBuffer(line));
      }
    } catch (error) {
      console.error('Stream processing error:', error);
      throw error;
    }
  };

  const processBuffer = (buffer) => {
    if (!buffer.trim()) return;

    try {
      const data = JSON.parse(buffer);
      if (data.copy) setCopy(data.copy);
      if (data.target_audience) setTargetAudience(data.target_audience);
    } catch (error) {
      console.error('JSON parsing error:', error);
    }
  };

  return {
    copy,
    targetAudience,
    isLoading,
    error,
    analyzeProduct
  };
};

export default useProductAnalysis;

import * as cdk from 'aws-cdk-lib';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as cloudfront from 'aws-cdk-lib/aws-cloudfront';
import * as cloudfront_origins from 'aws-cdk-lib/aws-cloudfront-origins';
import * as s3deploy from 'aws-cdk-lib/aws-s3-deployment';
import * as path from 'path';
import { RemovalPolicy } from 'aws-cdk-lib';

export class ReactAppStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // S3バケットの作成
    const websiteBucket = new s3.Bucket(this, 'ReactAppBucket', {
      removalPolicy: RemovalPolicy.DESTROY, // 本番では RETAIN にすべき
      autoDeleteObjects: true, // 本番では false にすべき
    });

    // CloudFront Distributionの作成
    const distribution = new cloudfront.Distribution(this, 'ReactAppDistribution', {
      defaultBehavior: {
        origin:
          // S3 バケットへの OAC によるアクセス制御を設定
          cloudfront_origins.S3BucketOrigin.withOriginAccessControl(
            websiteBucket
          ),
      },
      defaultRootObject: 'index.html',
      errorResponses: [
        {
          httpStatus: 403,
          responseHttpStatus: 200,
          responsePagePath: '/index.html',
        },
        {
          httpStatus: 404,
          responseHttpStatus: 200,
          responsePagePath: '/index.html',
        },
      ],
    });

    // S3へのデプロイ
    new s3deploy.BucketDeployment(this, 'DeployReactApp', {
      sources: [s3deploy.Source.asset(path.join(__dirname, '../../react/build'))],
      destinationBucket: websiteBucket,
      distribution,
      distributionPaths: ['/*'],
    });

    // CloudFront URLの出力
    new cdk.CfnOutput(this, 'DistributionDomainName', {
      value: distribution.distributionDomainName,
      description: 'CloudFront Distribution Domain Name',
    });
  }
}