Salta ai contenuti

Gioco di Dungeon con IA

Modulo 3: Implementazione dell’API per la storia

Lo StoryApi consiste in una singola API generate_story che, dato un Game e una lista di Action come contesto, farà progredire una storia. Questa API verrà implementata come API streaming in Python/FastAPI e dimostrerà inoltre come apportare modifiche al codice generato per adattarlo allo scopo.

Implementazione dell’API

Per creare la nostra API, dobbiamo prima installare alcune dipendenze aggiuntive.

  • boto3 verrà utilizzato per chiamare Amazon Bedrock;
  • uvicorn verrà utilizzato per avviare la nostra API in combinazione con Lambda Web Adapter (LWA).
  • copyfiles è una dipendenza npm necessaria per supportare la copia multipiattaforma di file durante l’aggiornamento del task bundle.

Per installare queste dipendenze, esegui i seguenti comandi:

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

Ora sostituiamo il contenuto dei seguenti file in 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")

Analizzando il codice sopra:

  • Utilizziamo l’impostazione x-streaming per indicare che si tratta di un’API streaming quando genereremo il nostro client SDK. Questo ci permetterà di consumare questa API in modalità streaming mantenendo la type-safety!
  • La nostra API restituisce semplicemente un flusso di testo come definito sia da media_type="text/plain" che da response_class=PlainTextResponse

Infrastruttura

L’infrastruttura configurata precedentemente presuppone che tutte le API abbiano un API Gateway che si integra con funzioni Lambda. Per la nostra story_api non vogliamo utilizzare API Gateway poiché non supporta risposte in streaming. Utilizzeremo invece una Lambda Function URL configurata con response streaming.

Per supportare questo, aggiorneremo prima i nostri costrutti CDK come segue:

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

Ora aggiorneremo lo story_api per supportare la distribuzione 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

Distribuzione e test

Prima di tutto, compiliamo la codebase:

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

Ora puoi distribuire l’applicazione eseguendo:

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

La distribuzione richiederà circa 2 minuti.

Puoi anche distribuire tutti gli stack contemporaneamente. Clicca qui per maggiori dettagli.

Al termine della distribuzione, dovresti vedere output simili a questi (alcuni valori sono oscurati):

Terminal window
dungeon-adventure-infra-sandbox
dungeon-adventure-infra-sandbox: deploying... [2/2]
dungeon-adventure-infra-sandbox
Tempo di distribuzione: 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

Possiamo testare la nostra API in due modi:

  • Avviando un’istanza locale del server FastAPI e invocando le API con curl
  • Chiamare l'API distribuita usando curl con sigv4 abilitato

Avvia il server FastAPI locale con:

Terminal window
pnpm nx run dungeon_adventure.story_api:serve

Esegui la chiamata API con:

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 il comando ha successo, dovresti vedere una risposta in streaming simile a:

UnnamedHero si ergeva maestoso, il mantello che sventolava al vento....

Congratulazioni. Hai costruito e distribuito la tua prima API con FastAPI! 🎉🎉🎉