컨텐츠로 건너뛰기
@aws/nx-plugin
Blog

AI 던전 게임

모듈 3: 스토리 API 구현

StoryApi는 Game과 컨텍스트용 Action 목록을 입력받아 스토리를 진행하는 단일 API generate_story로 구성됩니다. 이 API는 Python/FastAPI로 스트리밍 API로 구현되며, 생성된 코드를 목적에 맞게 수정하는 방법을 추가로 시연합니다.

API 구현

API를 생성하기 전에 먼저 몇 가지 추가 종속성을 설치해야 합니다.

  • boto3는 Amazon Bedrock 호출에 사용됩니다.
  • uvicorn은 Lambda Web Adapter (LWA)와 함께 사용할 때 API 시작에 사용됩니다.
  • copyfiles는 bundle 작업 업데이트 시 크로스 플랫폼 파일 복사를 지원하기 위한 npm 종속성입니다.

이 종속성들을 설치하려면 다음 명령어를 실행하세요:

Terminal window
pnpm nx run dungeon_adventure.story_api:add --args boto3 uvicorn
Terminal window
pnpm add -Dw copyfiles

이제 packages/story_api/story_api의 다음 파일 내용을 교체합니다:

import json


from boto3 import client
from fastapi.responses import PlainTextResponse, StreamingResponse
from pydantic import BaseModel


from .init import app, lambda_handler


handler = lambda_handler


bedrock = client('bedrock-runtime')


class Action(BaseModel):
    role: str
    content: str


class StoryRequest(BaseModel):
    genre: str
    playerName: str
    actions: list[Action]


async def bedrock_stream(request: StoryRequest):
    messages = [
        {"role": "user", "content": "Continue or create a new story..."}
    ]


    for action in request.actions:
        messages.append({"role": action.role, "content": action.content})


    response = bedrock.invoke_model_with_response_stream(
        modelId='anthropic.claude-3-sonnet-20240229-v1:0',
        body=json.dumps({
            "system":f"""
            You are running an AI text adventure game in the {request.genre} genre.
            Player: {request.playerName}. Return less than 200 characters of text.
            """,
            "messages": messages,
            "max_tokens": 1000,
            "temperature": 0.7,
            "anthropic_version": "bedrock-2023-05-31"
        })
    )


    stream = response.get('body')
    if stream:
        for event in stream:
            chunk = event.get('chunk')
            if chunk:
                message = json.loads(chunk.get("bytes").decode())
                if message['type'] == "content_block_delta":
                    yield message['delta']['text'] or ""
                elif message['type'] == "message_stop":
                    yield "\n"


@app.post("/story/generate",
          openapi_extra={'x-streaming': True},
          response_class=PlainTextResponse)
def generate_story(request: StoryRequest) -> str:
    return StreamingResponse(bedrock_stream(request), media_type="text/plain")

위 코드 분석:

  • 클라이언트 SDK 생성 시 스트리밍 API임을 나타내기 위해 x-streaming 설정을 사용합니다. 이는 타입 안전성을 유지하면서 스트리밍 방식으로 API를 사용할 수 있게 합니다.
  • API는 media_type="text/plain"과 response_class=PlainTextResponse로 정의된 텍스트 스트림을 반환합니다.

인프라

이전에 설정한 인프라는 모든 API가 Lambda 함수와 통합하는 API Gateway를 사용한다고 가정합니다. story_api의 경우 스트리밍 응답을 지원하지 않는 API Gateway 대신 응답 스트리밍이 구성된 Lambda Function URL을 사용합니다.

이를 지원하기 위해 CDK 구성을 다음과 같이 업데이트합니다:

import { Duration, Stack, CfnOutput } from 'aws-cdk-lib';
import { IGrantable, Grant } from 'aws-cdk-lib/aws-iam';
import {
  Runtime,
  Code,
  Tracing,
  LayerVersion,
  FunctionUrlAuthType,
  InvokeMode,
  Function,
} from 'aws-cdk-lib/aws-lambda';
import { Construct } from 'constructs';
import url from 'url';
import { RuntimeConfig } from '../../core/runtime-config.js';


export class StoryApi extends Construct {
  public readonly handler: Function;


  constructor(scope: Construct, id: string) {
    super(scope, id);


    this.handler = new Function(this, 'Handler', {
      runtime: Runtime.PYTHON_3_12,
      handler: 'run.sh',
      code: Code.fromAsset(
        url.fileURLToPath(
          new URL(
            '../../../../../../dist/packages/story_api/bundle',
            import.meta.url,
          ),
        ),
      ),
      timeout: Duration.seconds(30),
      tracing: Tracing.ACTIVE,
      environment: {
        AWS_CONNECTION_REUSE_ENABLED: '1',
      },
    });


    const stack = Stack.of(this);
    this.handler.addLayers(
      LayerVersion.fromLayerVersionArn(
        this,
        'LWALayer',
        `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
      ),
    );
    this.handler.addEnvironment('PORT', '8000');
    this.handler.addEnvironment('AWS_LWA_INVOKE_MODE', 'response_stream');
    this.handler.addEnvironment('AWS_LAMBDA_EXEC_WRAPPER', '/opt/bootstrap');
    const functionUrl = this.handler.addFunctionUrl({
      authType: FunctionUrlAuthType.AWS_IAM,
      invokeMode: InvokeMode.RESPONSE_STREAM,
      cors: {
        allowedOrigins: ['*'],
        allowedHeaders: [
          'authorization',
          'content-type',
          'x-amz-content-sha256',
          'x-amz-date',
          'x-amz-security-token',
        ],
      },
    });


    new CfnOutput(this, 'StoryApiUrl', { value: functionUrl.url });


    // Register the API URL in runtime configuration for client discovery
    RuntimeConfig.ensure(this).config.apis = {
      ...RuntimeConfig.ensure(this).config.apis!,
      StoryApi: functionUrl.url,
    };
  }


  public grantInvokeAccess(grantee: IGrantable) {
    Grant.addToPrincipal({
      grantee,
      actions: ['lambda:InvokeFunctionUrl'],
      resourceArns: [this.handler.functionArn],
      conditions: {
        StringEquals: {
          'lambda:FunctionUrlAuthType': 'AWS_IAM',
        },
      },
    });
  }
}

이제 Lambda Web Adapter 배포를 지원하도록 story_api를 업데이트합니다.

#!/bin/bash


PATH=$PATH:$LAMBDA_TASK_ROOT/bin \
    PYTHONPATH=$PYTHONPATH:/opt/python:$LAMBDA_RUNTIME_DIR \
    exec python -m uvicorn --port=$PORT story_api.main:app

배포 및 테스트

먼저 코드베이스를 빌드합니다:

Terminal window
pnpm nx run-many --target build --all

이제 다음 명령어로 애플리케이션을 배포할 수 있습니다:

Terminal window
pnpm nx run @dungeon-adventure/infra:deploy dungeon-adventure-infra-sandbox

이 배포는 완료까지 약 2분 정도 소요됩니다.

한 번에 모든 스택을 배포할 수도 있습니다. 자세한 내용을 보려면 여기를 클릭하세요.

배포가 완료되면 다음과 유사한 출력을 확인할 수 있습니다(일부 값은 생략됨):

Terminal window
dungeon-adventure-infra-sandbox
dungeon-adventure-infra-sandbox: deploying... [2/2]


 ✅  dungeon-adventure-infra-sandbox


✨  배포 시간: 354초


Outputs:
dungeon-adventure-infra-sandbox.ElectroDbTableTableNameXXX = dungeon-adventure-infra-sandbox-ElectroDbTableXXX-YYY
dungeon-adventure-infra-sandbox.GameApiEndpointXXX = https://xxx.execute-api.region.amazonaws.com/prod/
dungeon-adventure-infra-sandbox.GameUIDistributionDomainNameXXX = xxx.cloudfront.net
dungeon-adventure-infra-sandbox.StoryApiStoryApiUrlXXX = https://xxx.lambda-url.ap-southeast-2.on.aws/
dungeon-adventure-infra-sandbox.UserIdentityUserIdentityIdentityPoolIdXXX = region:xxx
dungeon-adventure-infra-sandbox.UserIdentityUserIdentityUserPoolIdXXX = region_xxx

다음 방법으로 API를 테스트할 수 있습니다:

  • FastApi 서버를 로컬에서 시작하고 curl로 API 호출
  • 배포된 API를 sigv4 활성화 curl로 직접 호출

다음 명령어로 로컬 FastAPI 서버를 시작하세요:

Terminal window
pnpm nx run dungeon_adventure.story_api:serve

서버가 실행되면 다음 명령어로 호출합니다:

Terminal window
curl -N -X POST http://127.0.0.1:8000/story/generate \
  -d '{"genre":"superhero", "actions":[], "playerName":"UnnamedHero"}' \
  -H "Content-Type: application/json"

명령어가 성공적으로 실행되면 다음과 유사한 스트리밍 응답을 확인할 수 있습니다:

UnnamedHero stood tall, his cape billowing in the wind....

축하합니다. FastAPI를 사용한 첫 번째 API를 성공적으로 구축하고 배포했습니다! 🎉🎉🎉