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

AI 던전 게임

모듈 3: 스토리 API 구현

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

API 구현

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

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

다음 명령어로 종속성을 설치합니다:

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

이제 packages/story_api/story_api/main.py 내용을 다음으로 교체합니다:

packages/story_api/story_api/main.py
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, 'x-query': True},
          response_class=PlainTextResponse)
def generate_story(request: StoryRequest) -> str:
    return StreamingResponse(bedrock_stream(request), media_type="text/plain")

위 코드 분석:

  • 클라이언트 SDK 생성 시 스트리밍 API임을 나타내기 위해 x-streaming 설정 사용. 이를 통해 타입 안전성을 유지하면서 스트리밍 방식으로 API 소비 가능!
  • POST 요청이지만 mutation 대신 query로 처리하기 위해 x-query 설정 사용. TanStack Query가 스트리밍 상태를 관리할 수 있도록 함
  • media_type="text/plain"과 response_class=PlainTextResponse로 단순 텍스트 스트림 반환

인프라

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

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

packages/common/constructs/src/core/http-api.ts
import { Construct } from 'constructs';
import { CfnOutput, Duration, Stack } from 'aws-cdk-lib';
import {
  CorsHttpMethod,
  HttpApi as _HttpApi,
  HttpMethod,
  IHttpRouteAuthorizer,
} from 'aws-cdk-lib/aws-apigatewayv2';
import { HttpLambdaIntegration } from 'aws-cdk-lib/aws-apigatewayv2-integrations';
import {
  Code,
  Function,
  FunctionUrl,
  FunctionUrlAuthType,
  InvokeMode,
  LayerVersion,
  Runtime,
  Tracing,
} from 'aws-cdk-lib/aws-lambda';
import { Grant, IGrantable } from 'aws-cdk-lib/aws-iam';
import { RuntimeConfig } from './runtime-config.js';


export interface HttpApiProps {
  readonly apiName: string;
  readonly handler: string;
  readonly handlerFilePath: string;
  readonly runtime: Runtime;
  readonly defaultAuthorizer: IHttpRouteAuthorizer;
  readonly apiType?: 'api-gateway' | 'function-url-streaming';
  readonly allowedOrigins?: string[];
}


export class HttpApi extends Construct {
  public readonly api?: _HttpApi;
  public readonly routerFunctionUrl?: FunctionUrl;
  public readonly routerFunction: Function;


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


    this.routerFunction = new Function(this, `${id}Handler`, {
      timeout: Duration.seconds(30),
      runtime: props.runtime,
      handler: props.handler,
      code: Code.fromAsset(props.handlerFilePath),
      tracing: Tracing.ACTIVE,
      environment: {
        AWS_CONNECTION_REUSE_ENABLED: '1',
      },
    });


    let apiUrl;
    if (props.apiType === 'function-url-streaming') {
      const stack = Stack.of(this);
      this.routerFunction.addLayers(
        LayerVersion.fromLayerVersionArn(
          this,
          'LWALayer',
          `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
        ),
      );
      this.routerFunction.addEnvironment('PORT', '8000');
      this.routerFunction.addEnvironment(
        'AWS_LWA_INVOKE_MODE',
        'response_stream',
      );
      this.routerFunction.addEnvironment(
        'AWS_LAMBDA_EXEC_WRAPPER',
        '/opt/bootstrap',
      );
      this.routerFunctionUrl = this.routerFunction.addFunctionUrl({
        authType: FunctionUrlAuthType.AWS_IAM,
        invokeMode: InvokeMode.RESPONSE_STREAM,
        cors: {
          allowedOrigins: props.allowedOrigins ?? ['*'],
          allowedHeaders: [
            'authorization',
            'content-type',
            'x-amz-content-sha256',
            'x-amz-date',
            'x-amz-security-token',
          ],
        },
      });
      apiUrl = this.routerFunctionUrl.url;
    } else {
      this.api = new _HttpApi(this, id, {
        corsPreflight: {
          allowOrigins: props.allowedOrigins ?? ['*'],
          allowMethods: [CorsHttpMethod.ANY],
          allowHeaders: [
            'authorization',
            'content-type',
            'x-amz-content-sha256',
            'x-amz-date',
            'x-amz-security-token',
          ],
        },
        defaultAuthorizer: props.defaultAuthorizer,
      });


      this.api.addRoutes({
        path: '/{proxy+}',
        methods: [
          HttpMethod.GET,
          HttpMethod.DELETE,
          HttpMethod.POST,
          HttpMethod.PUT,
          HttpMethod.PATCH,
          HttpMethod.HEAD,
        ],
        integration: new HttpLambdaIntegration(
          'RouterIntegration',
          this.routerFunction,
        ),
      });
      apiUrl = this.api.url;
    }


    new CfnOutput(this, `${props.apiName}Url`, { value: apiUrl! });


    RuntimeConfig.ensure(this).config.httpApis = {
      ...RuntimeConfig.ensure(this).config.httpApis!,
      [props.apiName]: apiUrl,
    };
  }


  public grantInvokeAccess(grantee: IGrantable) {
    if (this.api) {
      Grant.addToPrincipal({
        grantee,
        actions: ['execute-api:Invoke'],
        resourceArns: [this.api.arnForExecuteApi('*', '/*', '*')],
      });
    } else if (this.routerFunction) {
      Grant.addToPrincipal({
        grantee,
        actions: ['lambda:InvokeFunctionUrl'],
        resourceArns: [this.routerFunction.functionArn],
        conditions: {
          StringEquals: {
            'lambda:FunctionUrlAuthType': 'AWS_IAM',
          },
        },
      });
    }
  }
}

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

packages/story_api/run.sh
#!/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


✨  Deployment time: 354s


Outputs:
dungeon-adventure-infra-sandbox.ElectroDbTableTableNameXXX = dungeon-adventure-infra-sandbox-ElectroDbTableXXX-YYY
dungeon-adventure-infra-sandbox.GameApiGameApiUrlXXX = https://xxx.region.amazonaws.com/
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를 구축하고 배포했습니다! 🎉🎉🎉