AI地牢游戏

模块三：故事API实现

警告

请确保已按照本指南中的步骤授予对Anthropic Claude 3.5 Sonnet v2模型的访问权限。

StoryApi包含一个generate_story API，该API根据给定的Game和上下文Action列表推进故事情节。该API将使用Python/FastAPI实现为流式API，并演示如何修改生成代码以适应实际需求。

API实现

要创建API，首先需要安装额外依赖：

• boto3用于调用Amazon Bedrock；
• uvicorn配合Lambda Web Adapter (LWA)启动API；
• copyfiles作为npm依赖项，用于在更新bundle任务时支持跨平台文件复制。

运行以下命令安装依赖：

pnpm

pnpm nx run dungeon_adventure.story_api:add --args boto3 uvicorn

yarn

yarn nx run dungeon_adventure.story_api:add --args boto3 uvicorn

npm

npx nx run dungeon_adventure.story_api:add --args boto3 uvicorn

bun

bunx nx run dungeon_adventure.story_api:add --args boto3 uvicorn

pnpm

pnpm add -Dw copyfiles

yarn

yarn add -D copyfiles

npm

npm install --legacy-peer-deps -D copyfiles

bun

bun install -D 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")

代码分析：

• 使用x-streaming标记流式API，便于生成客户端SDK时保持类型安全
• x-query将POST请求视为查询操作，充分利用TanStack Query管理流式状态
• API通过media_type="text/plain"和response_class=PlainTextResponse返回纯文本流

注意

修改FastAPI后需重新构建项目以在网站生成的客户端中反映变更。后续还将进行更多修改后再重建。

基础设施

先前设置的基础设施假设所有API都通过API Gateway与Lambda集成。对于story_api，我们需要使用支持响应流式传输的Lambda Function URL而非API Gateway。

更新CDK构造如下：

http-api.ts

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

story-api.ts

packages/common/constructs/src/app/http-apis/story-api.ts

import { Construct } from 'constructs';
import * as url from 'url';
import { HttpApi } from '../../core/http-api.js';
import { HttpIamAuthorizer } from 'aws-cdk-lib/aws-apigatewayv2-authorizers';
import { Runtime } from 'aws-cdk-lib/aws-lambda';
import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam';

export class StoryApi extends HttpApi {
  constructor(scope: Construct, id: string) {
    super(scope, id, {
      defaultAuthorizer: new HttpIamAuthorizer(),
      apiName: 'StoryApi',
      runtime: Runtime.PYTHON_3_12,
      handler: 'story_api.main.handler',
      apiType: 'function-url-streaming',
      handler: 'run.sh',
      handlerFilePath: url.fileURLToPath(
        new URL(
          '../../../../../../dist/packages/story_api/bundle',
          import.meta.url,
        ),
      ),
    });

    this.routerFunction.addToRolePolicy(
      new PolicyStatement({
        effect: Effect.ALLOW,
        actions: ['bedrock:InvokeModelWithResponseStream'],
        resources: [
          'arn:aws:bedrock:*::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0',
        ],
      }),
    );
  }
}

更新story_api以支持Lambda Web Adapter部署：

run.sh

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

project.json

packages/story_api/project.json

{
  "name": "dungeon_adventure.story_api",
  "$schema": "../../node_modules/nx/schemas/project-schema.json",
  "projectType": "application",
  "sourceRoot": "packages/story_api/story_api",
  "targets": {
    ...
    "bundle": {
      "cache": true,
      "executor": "nx:run-commands",
      "outputs": ["{workspaceRoot}/dist/packages/story_api/bundle"],
      "options": {
        "commands": [
          "uv export --frozen --no-dev --no-editable --project story_api -o dist/packages/story_api/bundle/requirements.txt",
          "uv pip install -n --no-installer-metadata --no-compile-bytecode --python-platform x86_64-manylinux2014 --python `uv python pin` --target dist/packages/story_api/bundle -r dist/packages/story_api/bundle/requirements.txt",
          "copyfiles -f packages/story_api/run.sh dist/packages/story_api/bundle"
        ],
        "parallel": false
      },
      "dependsOn": ["compile"]
    },
    ...
  }
}

部署与测试

首先构建代码库：

pnpm

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

yarn

yarn nx run-many --target build --all

npm

npx nx run-many --target build --all

bun

bunx nx run-many --target build --all

提示

若遇到lint错误，可运行以下命令自动修复：

pnpm

pnpm nx run-many --target lint --configuration=fix --all

yarn

yarn nx run-many --target lint --configuration=fix --all

npm

npx nx run-many --target lint --configuration=fix --all

bun

bunx nx run-many --target lint --configuration=fix --all

运行以下命令部署应用：

pnpm

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

yarn

yarn nx run @dungeon-adventure/infra:deploy dungeon-adventure-infra-sandbox

npm

npx nx run @dungeon-adventure/infra:deploy dungeon-adventure-infra-sandbox

bun

bunx nx run @dungeon-adventure/infra:deploy dungeon-adventure-infra-sandbox

部署约需2分钟完成。

点击查看批量部署详情

部署完成后可见类似输出（部分值已脱敏）：

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

本地测试

运行以下命令启动FastAPI服务器：

pnpm

pnpm nx run dungeon_adventure.story_api:serve

yarn

yarn nx run dungeon_adventure.story_api:serve

npm

npx nx run dungeon_adventure.story_api:serve

bun

bunx nx run dungeon_adventure.story_api:serve

运行测试命令：

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

云端测试

acurl ap-southeast-2 lambda -N -X POST \
  https://xxx.lambda-url.ap-southeast-2.on.aws/story/generate \
  -d '{"genre":"superhero", "actions":[], "playerName":"UnnamedHero"}' \
  -H "Content-Type: application/json"

警告

使用CDK输出的dungeon-adventure-infra-sandbox.StoryApiStoryApiUrlXXX值替换高亮URL，并设置正确的区域。

成功执行后将看到流式响应：

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

祝贺！您已成功使用FastAPI构建并部署了首个流式API！🎉🎉🎉