当前位置：首页 > news >正文

使用FastAPI微服务在AWS EKS中构建上下文增强型AI问答系统

news 来源：原创 2025/5/10 7:39:45

系统概述

本文介绍如何使用FastAPI在AWS Elastic Kubernetes Service (EKS)上构建一个由多个微服务组成的AI问答系统。该系统能够接收用户输入的提示(prompt)，通过调用其他微服务从AWS ElastiCache on Redis和Amazon DynamoDB获取相关上下文，然后利用AWS Bedrock的Meta Llama 4模型和Azure OpenAI GPT-4 API生成最终答案返回给用户。
这个架构展示了如何利用FastAPI的轻量级特性构建微服务系统，结合AWS和Azure的AI能力，同时利用云原生技术实现可扩展性和安全性。系统设计考虑了高可用性和容错能力，通过多模型集成提高回答质量，并通过云服务集成实现快速上下文检索。

架构设计

整个系统架构如下：

用户 → [网关服务 - FastAPI] → [AI核心服务 - FastAPI]├──> Redis (AWS ElastiCache) ← 快速上下文├──> DynamoDB ← 结构化数据或备用数据├──> AWS Bedrock (Llama 4)└──> Azure OpenAI (GPT-4)

微服务详解

1. 网关服务 (prompt-gateway-service)

职责：

接收用户通过REST API提交的提示(prompt)
将提示转发给AI核心服务
返回最终响应给用户

示例代码：

# prompt_gateway_service/main.py
from fastapi import FastAPI, HTTPException
import httpxapp = FastAPI()
AI_CORE_SERVICE_URL = "http://ai-core-service:8000/process"@app.post("/prompt")
async def receive_prompt(prompt: str):async with httpx.AsyncClient() as client:response = await client.post(AI_CORE_SERVICE_URL, json={"prompt": prompt})if response.status_code != 200:raise HTTPException(status_code=500, detail="AI服务失败")return response.json()

2. AI核心服务 (ai-core-service)

职责：

使用提示从Redis和DynamoDB获取相关上下文
将上下文与原始提示结合
同时调用Llama 4(Bedrock)和GPT-4(Azure)生成响应
聚合两个模型的结果并返回最终答案

示例代码：

# ai_core_service/main.py
from fastapi import FastAPI, Request
import boto3
import redis
import httpx
import os
import jsonapp = FastAPI()
redis_client = redis.Redis(host=os.getenv("REDIS_HOST"), port=6379, decode_responses=True)
dynamodb = boto3.resource("dynamodb", region_name="us-west-2")
table = dynamodb.Table("YourTableName")@app.post("/process")
async def process_prompt(request: Request):data = await request.json()prompt = data["prompt"]# 从Redis获取上下文context = redis_client.get(prompt)if not context:# 回退到DynamoDBresponse = table.get_item(Key={"prompt": prompt})context = response.get("Item", {}).get("context", "")combined_prompt = f"Context: {context}\n\nPrompt: {prompt}"# 调用Llama 4 (Bedrock)bedrock = boto3.client("bedrock-runtime", region_name="us-west-2")llama_response = bedrock.invoke_model(body=json.dumps({"prompt": combined_prompt, "max_tokens": 300}),modelId="meta.llama4-70b-chat-v1")llama_output = json.loads(llama_response['body'].read())['output']# 调用Azure OpenAI GPT-4azure_url = "https://<your-azure-openai-endpoint>/openai/deployments/<deployment>/chat/completions?api-version=2023-05-15"headers = {"api-key": os.getenv("AZURE_API_KEY")}payload = {"messages": [{"role": "user", "content": combined_prompt}],"model": "gpt-4"}async with httpx.AsyncClient() as client:azure_response = await client.post(azure_url, headers=headers, json=payload)gpt_output = azure_response.json()['choices'][0]['message']['content']return {"llama": llama_output, "gpt4": gpt_output}