第 21 章 · 实战三：全栈 Agent 服务

本章目标：
掌握 FastAPI + LangGraph 后端 Agent 服务的完整架构
学会使用 AI SDK 的 useChat 构建流式聊天前端
实现基于 Redis 的消息持久化与 DynamoDB 长时记忆
完成 Docker 容器化并部署到 AWS Fargate
接入 CloudWatch 监控与 Langfuse 可观测性

📌 本章综合了本课程的多个章节：FastAPI 生产基础（FastAPI ch22）、AI SDK Provider 管理（AI SDK ch03）、LangGraph 流式输出（Agent 工程 ch08）、可观测性（Langfuse（即将发布））。如遇到不熟悉的知识点，可回看对应章节。

21.1 项目架构与目录结构

整体架构

agent-prod-app/
├── backend/                    # FastAPI + LangGraph 后端
│   ├── app/
│   │   ├── __init__.py
│   │   ├── main.py            # FastAPI 入口
│   │   ├── agent.py           # LangGraph Agent 定义
│   │   ├── tools.py           # 工具函数
│   │   ├── persistence.py     # Redis/DynamoDB 持久化
│   │   └── config.py          # 配置管理
│   ├── tests/
│   ├── Dockerfile
│   └── requirements.txt
├── frontend/                   # React + useChat 前端
│   ├── src/
│   │   ├── App.tsx
│   │   ├── components/
│   │   │   ├── ChatBot.tsx    # 聊天界面组件
│   │   │   └── MessageList.tsx
│   │   └── index.tsx
│   ├── Dockerfile
│   └── package.json
├── docker-compose.yml          # 本地开发编排
└── deploy/                     # AWS 部署脚本
    └── deploy.sh

技术栈选择

层	技术	版本
后端框架	FastAPI	0.141+
Agent 编排	LangGraph	1.2.x
AI 调用	ai SDK	v6+（兼容 Gateway）
缓存/队列	Redis	7.x
数据库	DynamoDB	按需
前端框架	React + TypeScript	18+
聊天 UI	useChat（AI SDK）	v6+
容器化	Docker	latest
部署	AWS Fargate	—
CDN	CloudFront	—
监控	CloudWatch + Langfuse	—

21.2 阶段一：后端 FastAPI + LangGraph Agent

21.2.1 安装依赖

bash

cd backend
pip install fastapi uvicorn langgraph ai langchain-openai redis boto3 pydantic-settings python-dotenv

21.2.2 配置管理

app/config.py

python

"""配置管理：统一环境变量与默认值"""
import os
from functools import lru_cache
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # AI Provider
    AI_GATEWAY_API_KEY: str = ""
    MODEL_ID: str = "openai/gpt-4o-mini"
    
    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"
    
    # DynamoDB
    AWS_REGION: str = "us-east-1"
    TABLE_NAME: str = "agent-messages"
    
    # Langfuse
    LANGFUSE_PUBLIC_KEY: str = ""
    LANGFUSE_SECRET_KEY: str = ""
    LANGFUSE_HOST: str = "https://cloud.langfuse.com"
    
    # 应用
    APP_NAME: str = "Agent Production App"
    DEBUG: bool = False
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


@lru_cache()
def get_settings() -> Settings:
    return Settings()

21.2.3 LangGraph Agent 定义

app/agent.py

python

"""LangGraph Agent 定义：包含工具调用与状态管理"""
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langgraph.tools import ToolNode
from ai import createGateway
from .config import get_settings
from .tools import get_weather, search_docs


class AgentState(TypedDict):
    """Agent 状态：继承 MessagesState 并扩展工具结果"""
    messages: Annotated[list, add_messages]
    tool_results: list[str]  # 记录工具调用结果用于上下文
    iteration: int  # 防止无限循环


def llm_node(state: AgentState) -> dict:
    """LLM 调用节点：处理用户消息并调用工具"""
    settings = get_settings()
    gateway = createGateway(apiKey=settings.AI_GATEWAY_API_KEY)
    model = gateway(settings.MODEL_ID)
    
    result = await model.bind_tools([get_weather, search_docs]).ainvoke(state["messages"])
    
    # 检查是否需要调用工具
    if hasattr(result, "tool_calls") and result.tool_calls:
        return {
            "messages": [result],
            "iteration": state.get("iteration", 0) + 1,
        }
    return {"messages": [result]}


def tool_node(state: AgentState) -> dict:
    """工具执行节点：调用外部工具并返回结果"""
    tool_calls = state["messages"][-1].tool_calls
    tool_results = await ToolNode([get_weather, search_docs]).ainvoke(tool_calls)
    return {
        "messages": tool_results,
        "tool_results": [r.content for r in tool_results],
    }


def should_continue(state: AgentState) -> str:
    """条件边：判断是否继续调用工具或结束"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        if state.get("iteration", 0) >= 5:  # 防止无限循环
            return "end"
        return "tools"
    return "end"


# 构建图
graph = StateGraph(AgentState)
graph.add_node("llm", llm_node)
graph.add_node("tools", tool_node)
graph.add_conditional_edges("llm", should_continue, {"tools": "tools", "end": END})
graph.add_edge(START, "llm")
compiled_graph = graph.compile()

21.2.4 FastAPI 路由

app/main.py

python

"""FastAPI 主应用：路由与中间件"""
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from .agent import compiled_graph
from .persistence import save_message, get_conversation_history
from .config import get_settings
import asyncio


app = FastAPI(title="Agent Production App")

settings = get_settings()

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应限制为前端域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class ChatRequest(BaseModel):
    message: str
    conversation_id: str | None = None


class ChatResponse(BaseModel):
    response: str
    conversation_id: str
    messages: list[dict]


@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "ok", "service": settings.APP_NAME}


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """聊天接口：接收用户消息，返回 Agent 响应"""
    # 获取历史对话
    history = await get_conversation_history(request.conversation_id)
    
    # 准备消息列表
    messages = history + [{"role": "user", "content": request.message}]
    
    # 运行 Agent
    try:
        result = await compiled_graph.ainvoke({
            "messages": messages,
            "iteration": 0,
        })
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    
    # 提取最后一条 AI 消息
    ai_message = result["messages"][-1]
    response_text = ai_message.content if hasattr(ai_message, "content") else str(ai_message)
    
    # 保存消息到持久化存储
    await save_message(
        conversation_id=request.conversation_id or "default",
        user_message=request.message,
        ai_message=response_text,
        metadata={
            "tool_calls": getattr(ai_message, "tool_calls", []),
            "iterations": result.get("iteration", 0),
        },
    )
    
    return ChatResponse(
        response=response_text,
        conversation_id=request.conversation_id or "default",
        messages=[{"role": m.type, "content": m.content} for m in result["messages"]],
    )


@app.get("/chat/{conversation_id}/history")
async def get_history(conversation_id: str):
    """获取对话历史"""
    history = await get_conversation_history(conversation_id)
    return {"messages": history}

21.3 阶段二：前端 React + useChat

21.3.1 安装依赖

bash

cd frontend
npm install ai react react-dom
npm install -D typescript @types/react @types/node

21.3.2 聊天组件

frontend/src/components/ChatBot.tsx

tsx

/**
 * 聊天界面组件：基于 AI SDK 的 useChat hook
 * 参考：/ai-sdk/ ch16
 */
import { useChat } from "ai/react";
import { MessageList } from "./MessageList";

export function ChatBot({ conversationId }: { conversationId?: string }) {
  const {
    messages,
    input,
    handleInputChange,
    handleSubmit,
    isLoading,
    error,
  } = useChat({
    api: "http://localhost:8000/chat",
    body: { conversation_id: conversationId },
    onResponse: (response) => {
      console.log("Server response:", response.status);
    },
    onError: (error) => {
      console.error("Chat error:", error);
    },
  });

  return (
    <div className="chat-container">
      <div className="chat-header">
        <h2>Agent 助手</h2>
        <span className="status">{isLoading ? "思考中..." : "在线"}</span>
      </div>

      <MessageList messages={messages} />

      {error && (
        <div className="error-message">
          请求出错：{error.message}
        </div>
      )}

      <form onSubmit={handleSubmit} className="chat-input">
        <input
          type="text"
          value={input}
          onChange={handleInputChange}
          placeholder="输入消息..."
          disabled={isLoading}
        />
        <button type="submit" disabled={isLoading || !input.trim()}>
          发送
        </button>
      </form>
    </div>
  );
}

21.3.3 消息列表组件

frontend/src/components/MessageList.tsx

tsx

/**
 * 消息列表：渲染用户与 AI 消息
 */
import { Message } from "ai";

interface MessageListProps {
  messages: Message[];
}

export function MessageList({ messages }: MessageListProps) {
  return (
    <div className="message-list">
      {messages.map((msg) => (
        <div
          key={msg.id}
          className={`message ${msg.role}`}
        >
          <div className="message-content">
            {msg.content}
          </div>
        </div>
      ))}
      {messages.length === 0 && (
        <div className="empty-state">
          开始对话吧！
        </div>
      )}
    </div>
  );
}

21.3.4 主应用

frontend/src/App.tsx

tsx

/**
 * 主应用：挂载聊天组件
 */
import { ChatBot } from "./components/ChatBot";

export default function App() {
  return (
    <div className="app">
      <header>
        <h1>全栈 Agent 服务</h1>
      </header>
      <main>
        <ChatBot />
      </main>
    </div>
  );
}

21.4 阶段三：消息持久化

21.4.1 Redis 短期记忆

app/persistence.py

python

"""消息持久化：Redis 短期记忆 + DynamoDB 长期记忆"""
import json
import uuid
from datetime import datetime
from typing import Optional
import redis
import boto3
from .config import get_settings


class MessagePersistence:
    """消息持久化管理器"""
    
    def __init__(self):
        self.settings = get_settings()
        self.redis = redis.from_url(self.settings.REDIS_URL, decode_responses=True)
        self.dynamodb = boto3.resource(
            "dynamodb",
            region_name=self.settings.AWS_REGION,
        )
        self.table = self.dynamodb.Table(self.settings.TABLE_NAME)
    
    async def save_message(
        self,
        conversation_id: str,
        user_message: str,
        ai_message: str,
        metadata: dict | None = None,
    ) -> str:
        """保存对话消息"""
        message_id = str(uuid.uuid4())
        timestamp = datetime.utcnow().isoformat()
        
        # Redis：短期记忆（带 TTL）
        key = f"conv:{conversation_id}:messages"
        message_data = {
            "id": message_id,
            "role": "user",
            "content": user_message,
            "timestamp": timestamp,
        }
        await self.redis.lpush(key, json.dumps(message_data))
        
        ai_data = {
            "id": str(uuid.uuid4()),
            "role": "assistant",
            "content": ai_message,
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": metadata or {},
        }
        await self.redis.lpush(key, json.dumps(ai_data))
        
        # 设置 TTL（7 天）
        await self.redis.expire(key, 7 * 24 * 3600)
        
        # DynamoDB：长期记忆
        await self.table.put_item(
            Item={
                "conversation_id": conversation_id,
                "message_id": message_id,
                "role": "user",
                "content": user_message,
                "timestamp": timestamp,
            }
        )
        await self.table.put_item(
            Item={
                "conversation_id": conversation_id,
                "message_id": ai_data["id"],
                "role": "assistant",
                "content": ai_message,
                "timestamp": ai_data["timestamp"],
                "metadata": metadata or {},
            }
        )
        
        return message_id
    
    async def get_conversation_history(
        self,
        conversation_id: str,
        limit: int = 50,
    ) -> list[dict]:
        """获取对话历史"""
        key = f"conv:{conversation_id}:messages"
        messages = await self.redis.lrange(key, 0, limit - 1)
        
        # 按时间排序（Redis LPUSH 是倒序）
        return sorted(
            [json.loads(m) for m in messages],
            key=lambda x: x["timestamp"],
        )


# 全局单例
persistence = MessagePersistence()

21.4.2 更新路由使用持久化

修改 app/main.py 中的导入：

python

from .persistence import persistence as persist
# 替换 save_message 和 get_conversation_history 调用为 persist.save_message()

21.5 阶段四：Docker 容器化

21.5.1 后端 Dockerfile

backend/Dockerfile

dockerfile

FROM python:3.12-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

21.5.2 前端 Dockerfile

frontend/Dockerfile

dockerfile

FROM node:20-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .
RUN npm run build

# 使用 nginx  serving static files
FROM nginx:alpine
COPY --from=0 /app/build /usr/share/nginx/html
COPY nginx.conf /etc/nginx/conf.d/default.conf

EXPOSE 80
CMD ["nginx", "-g", "daemon off;"]

21.5.3 docker-compose.yml

yaml

version: "3.9"

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  backend:
    build: ./backend
    ports:
      - "8000:8000"
    environment:
      - AI_GATEWAY_API_KEY=${AI_GATEWAY_API_KEY}
      - REDIS_URL=redis://redis:6379/0
      - AWS_REGION=${AWS_REGION:-us-east-1}
      - LANGFUSE_PUBLIC_KEY=${LANGFUSE_PUBLIC_KEY}
      - LANGFUSE_SECRET_KEY=${LANGFUSE_SECRET_KEY}
      - LANGFUSE_HOST=${LANGFUSE_HOST:-https://cloud.langfuse.com}
    depends_on:
      - redis

  frontend:
    build: ./frontend
    ports:
      - "3000:80"
    depends_on:
      - backend

volumes:
  redis_data:

21.5.4 本地运行

bash

# 设置环境变量
cp .env.example .env
# 编辑 .env 填入 AI_GATEWAY_API_KEY

# 启动所有服务
docker-compose up -d

# 访问
# 前端：http://localhost:3000
# API：http://localhost:8000/docs
# Redis 监控：http://localhost:8001（需要 redis-commander）

21.6 阶段五：部署到 AWS Fargate

21.6.1 ECR 镜像仓库

bash

# 登录 ECR
aws ecr get-login-password --region us-east-1 | \
  docker login --username AWS --password-stdin <account>.dkr.ecr.us-east-1.amazonaws.com

# 构建并推送后端镜像
cd backend
docker build -t agent-backend:latest .
docker tag agent-backend:latest <account>.dkr.ecr.us-east-1.amazonaws.com/agent-backend:latest
docker push <account>.dkr.ecr.us-east-1.amazonaws.com/agent-backend:latest

# 构建并推送前端镜像
cd ../frontend
docker build -t agent-frontend:latest .
docker tag agent-frontend:latest <account>.dkr.ecr.us-east-1.amazonaws.com/agent-frontend:latest
docker push <account>.dkr.ecr.us-east-1.amazonaws.com/agent-frontend:latest

21.6.2 ECS Task Definition

deploy/task-definition.json

json

{
  "family": "agent-service",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "512",
  "memory": "1024",
  "executionRoleArn": "arn:aws:iam::<account>:role/ecs-task-execution-role",
  "taskRoleArn": "arn:aws:iam::<account>:role/ecs-task-role",
  "containerDefinitions": [
    {
      "name": "backend",
      "image": "<account>.dkr.ecr.us-east-1.amazonaws.com/agent-backend:latest",
      "portMappings": [
        {
          "containerPort": 8000,
          "protocol": "tcp"
        }
      ],
      "environment": [
        {
          "name": "AI_GATEWAY_API_KEY",
          "value": "${AI_GATEWAY_API_KEY}"
        },
        {
          "name": "REDIS_URL",
          "value": "redis://<redis-endpoint>:6379/0"
        }
      ],
      "secrets": [
        {
          "name": "AI_GATEWAY_API_KEY",
          "valueFrom": "arn:aws:secretsmanager:us-east-1:<account>:secret:agent/ai-key"
        }
      ],
      "logConfiguration": {
        "logDriver": "awslogs",
        "options": {
          "awslogs-group": "/ecs/agent-service",
          "awslogs-region": "us-east-1",
          "awslogs-stream-prefix": "backend"
        }
      }
    },
    {
      "name": "frontend",
      "image": "<account>.dkr.ecr.us-east-1.amazonaws.com/agent-frontend:latest",
      "portMappings": [
        {
          "containerPort": 80,
          "protocol": "tcp"
        }
      ],
      "dependsOn": [
        {
          "containerName": "backend",
          "condition": "START"
        }
      ]
    }
  ]
}

21.6.3 CloudFront 分发

bash

# 创建 CloudFront 分发（前端静态资源）
aws cloudfront create-distribution \
  --origin-domain-name <alb-dns>.amazonaws.com \
  --default-cache-behavior Enabled=true \
  --paths "/*"

21.6.4 完整部署脚本

deploy/deploy.sh

bash

#!/bin/bash
set -e

ACCOUNT="<account>"
REGION="us-east-1"
REPO="agent-service"

echo "=== 构建并推送镜像 ==="
docker build -t ${REPO}-backend:latest ./backend
docker build -t ${REPO}-frontend:latest ./frontend

docker tag ${REPO}-backend:latest ${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com/${REPO}-backend:latest
docker tag ${REPO}-frontend:latest ${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com/${REPO}-frontend:latest

aws ecr get-login-password --region ${REGION} | \
  docker login --username AWS --password-stdin ${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com

docker push ${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com/${REPO}-backend:latest
docker push ${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com/${REPO}-frontend:latest

echo "=== 更新 ECS 服务 ==="
aws ecs update-service \
  --cluster ${REPO}-cluster \
  --service ${REPO}-backend-service \
  --force-new-deployment

echo "=== 部署完成 ==="
echo "访问地址：https://<cloudfront-domain>"

21.7 阶段六：监控与可观测性

21.7.1 CloudWatch Logs

app/main.py 中添加日志中间件：

python

import logging
import time
from contextlib import asynccontextmanager

logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时初始化
    logger.info("Application starting up")
    yield
    # 关闭时清理
    logger.info("Application shutting down")

app = FastAPI(lifespan=lifespan)

21.7.2 Langfuse 集成

app/main.py

python

from langfuse.decorators import langfuse_context, observe
from .config import get_settings

settings = get_settings()

# 初始化 Langfuse
langfuse_context.update_user_metadata(
    public_key=settings.LANGFUSE_PUBLIC_KEY,
    secret_key=settings.LANGFUSE_SECRET_KEY,
    base_url=settings.LANGFUSE_HOST,
)

@app.post("/chat", response_model=ChatResponse)
@observe()  # 自动追踪
async def chat(request: ChatRequest):
    # 原有的 chat 逻辑...
    pass

21.7.3 CloudWatch 告警

bash

# 创建 CloudWatch 告警
aws cloudwatch put-metric-alert \
  --alert-name "AgentBackendHighError" \
  --alarm-actions arn:aws:sns:us-east-1:<account>:alerts \
  --metric-name CPUUtilization \
  --namespace AWS/ECS \
  --statistic Average \
  --period 300 \
  --threshold 80 \
  --comparison-operator GreaterThanThreshold

21.8 完整项目示例

项目结构

agent-prod-app/
├── backend/
│   ├── app/
│   │   ├── __init__.py
│   │   ├── main.py
│   │   ├── agent.py
│   │   ├── tools.py
│   │   ├── persistence.py
│   │   └── config.py
│   ├── tests/
│   │   ├── test_agent.py
│   │   └── test_persistence.py
│   ├── Dockerfile
│   └── requirements.txt
├── frontend/
│   ├── src/
│   │   ├── App.tsx
│   │   └── components/
│   │       ├── ChatBot.tsx
│   │       └── MessageList.tsx
│   ├── Dockerfile
│   └── package.json
├── docker-compose.yml
├── deploy/
│   └── deploy.sh
└── README.md

requirements.txt

fastapi==0.141.0
uvicorn==0.34.0
langgraph==1.2.11
ai==6.6.2
langchain-openai==0.3.0
redis==5.2.1
boto3==1.36.0
pydantic-settings==2.7.1
python-dotenv==1.0.1
langfuse==2.53.0

package.json

json

{
  "name": "agent-frontend",
  "version": "1.0.0",
  "scripts": {
    "dev": "next dev",
    "build": "next build",
    "start": "next start"
  },
  "dependencies": {
    "ai": "^6.6.2",
    "react": "^18.3.1",
    "react-dom": "^18.3.1",
    "next": "^15.1.0"
  },
  "devDependencies": {
    "@types/react": "^18.3.0",
    "@types/node": "^20.0.0",
    "typescript": "^5.0.0"
  }
}

本章小结

本章完成了全栈 Agent 服务的完整构建：

后端：FastAPI + LangGraph Agent，支持工具调用与条件边
前端：React + useChat，实现流式聊天界面
持久化：Redis 短期记忆 + DynamoDB 长期记忆
部署：Docker 容器化 → ECR → Fargate → CloudFront
可观测性：CloudWatch 日志监控 + Langfuse AI 追踪

📌 关键设计模式：
分层架构：API 层 → Agent 层 → 工具层 → 持久化层
状态隔离：短期对话用 Redis，长期用户数据用 DynamoDB
部署自动化：一键脚本从构建到部署全流程
可观测性优先：Langfuse 追踪每个 Agent 决策

🛠️ 动手实践

扩展工具集：为 Agent 添加搜索工具（调用 Tavily API）和数据库查询工具
实现断点续传：在前端添加"恢复对话"功能，从 DynamoDB 加载历史消息
性能压测：使用 Locust（参考 Locust 教程）测试 Agent 服务的并发能力

参考资料

FastAPI 生产部署 — Docker 与部署
AI SDK Provider 管理 — Gateway 与自定义 Provider
LangGraph 持久化 — Checkpointer 与 Thread
Langfuse 可观测性（即将发布） — 追踪与评估

⚠️ 注：Agent 工程实战课程的 ch07、ch14 章节正在建设中，暂链接到已完成章节。

第 21 章 · 实战三：全栈 Agent 服务 ​

21.1 项目架构与目录结构 ​

整体架构 ​

技术栈选择 ​

21.2 阶段一：后端 FastAPI + LangGraph Agent ​

21.2.1 安装依赖 ​

21.2.2 配置管理 ​

21.2.3 LangGraph Agent 定义 ​

21.2.4 FastAPI 路由 ​

21.3 阶段二：前端 React + useChat ​

21.3.1 安装依赖 ​

21.3.2 聊天组件 ​

21.3.3 消息列表组件 ​

21.3.4 主应用 ​

21.4 阶段三：消息持久化 ​

21.4.1 Redis 短期记忆 ​

21.4.2 更新路由使用持久化 ​

21.5 阶段四：Docker 容器化 ​

21.5.1 后端 Dockerfile ​

21.5.2 前端 Dockerfile ​

21.5.3 docker-compose.yml ​

21.5.4 本地运行 ​

21.6 阶段五：部署到 AWS Fargate ​

21.6.1 ECR 镜像仓库 ​

21.6.2 ECS Task Definition ​

21.6.3 CloudFront 分发 ​

21.6.4 完整部署脚本 ​

21.7 阶段六：监控与可观测性 ​

21.7.1 CloudWatch Logs ​

21.7.2 Langfuse 集成 ​

21.7.3 CloudWatch 告警 ​

21.8 完整项目示例 ​

项目结构 ​

requirements.txt ​

package.json ​

本章小结 ​

🛠️ 动手实践 ​

参考资料 ​

第 21 章 · 实战三：全栈 Agent 服务

21.1 项目架构与目录结构

整体架构

技术栈选择

21.2 阶段一：后端 FastAPI + LangGraph Agent

21.2.1 安装依赖

21.2.2 配置管理

21.2.3 LangGraph Agent 定义

21.2.4 FastAPI 路由

21.3 阶段二：前端 React + useChat

21.3.1 安装依赖

21.3.2 聊天组件

21.3.3 消息列表组件

21.3.4 主应用

21.4 阶段三：消息持久化

21.4.1 Redis 短期记忆

21.4.2 更新路由使用持久化

21.5 阶段四：Docker 容器化

21.5.1 后端 Dockerfile

21.5.2 前端 Dockerfile

21.5.3 docker-compose.yml

21.5.4 本地运行

21.6 阶段五：部署到 AWS Fargate

21.6.1 ECR 镜像仓库

21.6.2 ECS Task Definition

21.6.3 CloudFront 分发

21.6.4 完整部署脚本

21.7 阶段六：监控与可观测性

21.7.1 CloudWatch Logs

21.7.2 Langfuse 集成

21.7.3 CloudWatch 告警

21.8 完整项目示例

项目结构

requirements.txt

package.json

本章小结

🛠️ 动手实践

参考资料