Saltearse al contenido

Juego de Mazmorra con IA

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

La StoryApi consta de una única API generate_story que, dado un Game y una lista de Actions como contexto, generará una progresión de la historia. Esta API se implementará como una API de streaming en Python/FastAPI y demostrará cómo realizar modificaciones al código generado para adaptarlo a su propósito.

Implementación de la API

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

  • boto3 se usará para interactuar con Amazon Bedrock;
  • uvicorn se utilizará para iniciar nuestra API en combinación 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 reemplazaremos el contenido de los siguientes archivos en 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")

Analizando el código anterior:

  • Usamos la configuración x-streaming para indicar que es una API de streaming al generar nuestro SDK cliente. Esto nos permitirá consumir esta API en modo streaming manteniendo la seguridad de tipos.
  • Nuestra API simplemente devuelve un flujo de texto definido por media_type="text/plain" y response_class=PlainTextResponse

Infraestructura

La infraestructura configurada previamente asume que todas las APIs usan API Gateway integrado con funciones Lambda. Para nuestra story_api no queremos usar API Gateway ya que no soporta respuestas en streaming. En su lugar, usaremos una Lambda Function URL configurada con response streaming.

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

Ahora actualizaremos la story_api para soportar el despliegue con Lambda Web Adapter:

#!/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 la aplicación ejecutando:

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

Este despliegue tomará aproximadamente 2 minutos.

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

Una vez completado el despliegue, verás salidas similares a estas (algunos valores fueron omitidos):

Ventana de terminal
dungeon-adventure-infra-sandbox
dungeon-adventure-infra-sandbox: deploying... [2/2]
dungeon-adventure-infra-sandbox
Tiempo de despliegue: 354s
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

Podemos probar nuestra API de dos formas:

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

Inicia el servidor FastAPI localmente con:

Terminal window
pnpm nx run dungeon_adventure.story_api:serve

Una vez en ejecución, 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, verás 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. 🎉🎉🎉