Pular para o conteúdo

Jogo de Dungeons com IA

Módulo 3: Implementação da API de História

A StoryApi consiste em uma única API generate_story que, dado um Game e uma lista de Actions como contexto, irá progredir uma história. Esta API será implementada como uma API de streaming em Python/FastAPI e também demonstrará como alterações podem ser feitas no código gerado para adequá-lo ao propósito.

Implementação da API

Para criar nossa API, primeiro precisamos instalar algumas dependências adicionais:

  • boto3 será usado para chamar o Amazon Bedrock;
  • uvicorn será usado para iniciar nossa API em conjunto com o Lambda Web Adapter (LWA);
  • copyfiles é uma dependência npm necessária para suportar a cópia multiplataforma de arquivos ao atualizar nossa tarefa bundle.

Para instalar estas dependências, execute os seguintes comandos:

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

Agora vamos substituir o conteúdo dos seguintes arquivos em 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")

Analisando o código acima:

  • Usamos a configuração x-streaming para indicar que esta é uma API de streaming ao gerar nosso client SDK. Isso permitirá consumir esta API de forma streaming mantendo a segurança de tipos!
  • Nossa API simplesmente retorna um fluxo de texto conforme definido por media_type="text/plain" e response_class=PlainTextResponse

Infraestrutura

A infraestrutura que configuramos anteriormente assume que todas as APIs têm um API Gateway integrado com funções Lambda. Para nossa story_api, não queremos usar o API Gateway pois ele não suporta respostas em streaming. Em vez disso, usaremos um Lambda Function URL configurado com response streaming.

Para suportar isso, primeiro atualizaremos nossos construtos CDK da seguinte forma:

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',
},
},
});
}
}

Agora atualizaremos a story_api para suportar a implantação do 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

Implantação e testes

Primeiro, vamos construir a base de código:

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

Sua aplicação agora pode ser implantada executando o seguinte comando:

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

Esta implantação levará aproximadamente 2 minutos para ser concluída.

Você também pode implantar todas as stacks de uma vez. Clique aqui para mais detalhes.

Após a conclusão da implantação, você verá saídas semelhantes às seguintes (alguns valores foram omitidos):

Terminal window
dungeon-adventure-infra-sandbox
dungeon-adventure-infra-sandbox: deploying... [2/2]
dungeon-adventure-infra-sandbox
Tempo de implantação: 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 testar nossa API de duas formas:

  • Iniciando uma instância local do servidor FastAPI e invocando as APIs usando curl
  • Chamando a API implantada diretamente usando curl com Sigv4

Inicie o servidor FastAPI localmente com:

Terminal window
pnpm nx run dungeon_adventure.story_api:serve

Após iniciar, execute o comando:

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"

Se executado com sucesso, você verá uma resposta em streaming similar a:

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

Parabéns. Você construiu e implantou sua primeira API usando FastAPI! 🎉🎉🎉