Saltearse al contenido
@aws/nx-plugin
Blog

Juego de Mazmorra con IA

Módulo 3: Implementación de la API de Story

El StoryApi consiste en una única API generate_story que, dado un Game y una lista de Actions como contexto, progresará una historia. Esta API se implementará como una API de streaming en Python/FastAPI y demostrará cómo se pueden modificar los códigos generados para adaptarlos a su propósito.

Implementación de la API

Para crear nuestra API, primero necesitamos instalar algunas dependencias adicionales:

  • boto3 se usará para llamar a Amazon Bedrock;
  • uvicorn se usará para iniciar nuestra API en conjunto con el Lambda Web Adapter (LWA).
  • copyfiles es una dependencia de npm necesaria para soportar la copia multiplataforma de archivos al actualizar nuestra tarea bundle.

Para instalar estas dependencias, ejecuta los siguientes comandos:

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

Ahora reemplazamos el contenido de packages/story_api/story_api/main.py con lo siguiente:

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")

Analizando el código anterior:

  • Usamos la configuración x-streaming para indicar que esta es una API de streaming cuando generemos nuestro SDK cliente. ¡Esto nos permitirá consumir esta API en modo streaming manteniendo la seguridad de tipos!
  • Usamos la configuración x-query para indicar que, aunque es una solicitud POST, la trataremos como query en lugar de mutation, permitiéndonos aprovechar al máximo la gestión de estado de streaming de TanStack Query.
  • Nuestra API simplemente devuelve un flujo de texto como lo definen media_type="text/plain" y response_class=PlainTextResponse

Infraestructura

La infraestructura que configuramos previamente asume que todas las APIs tienen un API Gateway integrado con Lambda. Para nuestro story_api no queremos usar API Gateway ya que no soporta respuestas en streaming. En su lugar, usaremos una URL de función Lambda configurada con transmisión de respuestas.

Para soportar esto, primero actualizaremos nuestros constructos 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',
          },
        },
      });
    }
  }
}

Ahora actualizaremos el story_api para soportar el despliegue con Lambda Web Adapter.

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

Despliegue y pruebas

Primero, construyamos el código base:

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

Ahora puedes desplegar tu aplicación ejecutando:

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

Este despliegue tomará aproximadamente 2 minutos en completarse.

También puedes desplegar todos los stacks a la vez. Haz clic aquí para más detalles.

Una vez completado el despliegue, deberías ver salidas similares a las siguientes (algunos valores han sido omitidos):

Ventana de terminal
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

Podemos probar nuestra API de dos formas:

  • Iniciando una instancia local del servidor FastApi e invocando las APIs usando curl.
  • Llamar a la API desplegada usando curl con Sigv4 directamente

Inicia tu servidor FastAPI local ejecutando:

Terminal window
pnpm nx run dungeon_adventure.story_api:serve

Una vez en funcionamiento, llama a la API con:

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

Si el comando se ejecuta correctamente, deberías ver una respuesta en streaming similar a:

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

¡Felicidades! Has construido y desplegado tu primera API usando FastAPI. 🎉🎉🎉