前言

2026年，AIGC应用的开发早已不是“调个API、套个聊天界面”那么简单。当你的应用需要处理复杂任务——智能对话、知识库问答、多步推理、工具调用——线性的一次性请求-response模式已经远远不够用。行业共识已经形成：AI编程彻底告别了“逐行补全”的时代，全面迈入多智能体协同软件工程的新纪元。

LangChain官方2026年的一份报告里有个数据让我印象深刻：超过60%的Agent生产事故，都和状态管理有关——这不是模型问题，是工程问题。

本文将从零开始，带你构建一个完整的AIGC全栈应用。我们将使用 Python + LangGraph 构建后端Agent工作流，React + SSE 构建前端流式交互，并完成从开发到部署的完整工程化链路。所有代码均可直接运行。

---

一、架构设计：从“调接口”到“编排工作流”

1.1 传统方案的痛点

在进入正题前，先看一个真实场景：你有一个智能客服助手，用户问“帮我查一下订单12345的状态，然后如果超时了就生成一封催单邮件”。

如果用传统的一次性API调用，你会怎么做？

• 调用LLM解析意图
• 调用订单查询API
• 判断是否超时
• 调用邮件生成API
• 返回结果

看起来没问题，但一旦中间某步失败——比如订单API超时、LLM返回格式不对——整个流程就要从头重跑。更麻烦的是，如果用户想中途修改指令（“不催单了，帮我退款”），线性流程根本无法处理。

1.2 Agent工作流：StateGraph的解决方案

LangGraph的核心思想是：把Agent的执行过程建模为一个有向图（Graph），节点是执行单元，边定义执行顺序，一个共享的State对象贯穿整个流程。

# state.py
from typing import TypedDict, List, Optional, Any
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """Agent的共享状态"""
    messages: List[dict]  # 对话历史
    user_query: str       # 用户原始问题
    order_id: Optional[str]
    order_status: Optional[str]
    is_expired: Optional[bool]
    action_result: Optional[str]
    final_answer: str
    error: Optional[str]
    retry_count: int

State是TypedDict，作为整个Graph的共享内存。每个节点读取State并返回需要更新的字段，未更新的字段保持不变。

1.3 整体架构图

┌─────────────────────────────────────────────────────────────────┐
│                         前端 (React)                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │
│  │   Chat UI   │◄─│  useSSE     │◄─│ 事件解析 & 状态更新      │ │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘ │
│         │                                                       │
│         ▼ SSE (Server-Sent Events)                              │
├─────────────────────────────────────────────────────────────────┤
│                      后端 (FastAPI + LangGraph)                 │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │                    StateGraph 工作流                        ││
│  │  ┌──────────┐    ┌──────────┐    ┌──────────┐             ││
│  │  │ 入口节点  │───▶│ 意图识别 │───▶│ 路由决策 │             ││
│  │  └──────────┘    └──────────┘    └────┬─────┘             ││
│  │                                        │                    ││
│  │              ┌─────────────────────────┼──────────────┐    ││
│  │              ▼                         ▼              ▼    ││
│  │       ┌───────────┐           ┌───────────┐   ┌─────────┐││
│  │       │ 订单查询  │           │ 邮件生成  │   │ 知识库  │││
│  │       └───────────┘           └───────────┘   └─────────┘││
│  │              │                         │              │    ││
│  │              └─────────────┬───────────┴──────────────┘    ││
│  │                            ▼                               ││
│  │                    ┌──────────────┐                        ││
│  │                    │  汇总回答节点 │                        ││
│  │                    └──────────────┘                        ││
│  └─────────────────────────────────────────────────────────────┘│
│         │                                                       │
│         ▼ 外部依赖                                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐    │
│  │  LLM (OpenAI│  │  订单API    │  │  RAG向量数据库      │    │
│  │  /DashScope)│  │  (内部服务) │  │  (Pinecone/Qdrant)  │    │
│  └─────────────┘  └─────────────┘  └─────────────────────┘    │
├─────────────────────────────────────────────────────────────────┤
│                    工程化基础设施                                │
│  Docker · GitHub Actions CI/CD · Prometheus · Grafana          │
└─────────────────────────────────────────────────────────────────┘

---

二、后端实现：基于LangGraph的Agent工作流

2.1 环境准备

# 创建项目
mkdir aigc-agent-app && cd aigc-agent-app
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install langgraph langchain-openai fastapi uvicorn sse-starlette python-dotenv httpx

创建.env文件：

OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1  # 或使用国内代理
MODEL_NAME=gpt-4o-mini

2.2 定义状态与节点

# agent/graph.py
import json
from typing import Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from .state import AgentState

# 初始化LLM
llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0.3,
    streaming=True,
)

# ----- 节点函数 -----

def entry_node(state: AgentState) -> dict:
    """入口节点：提取用户查询"""
    messages = state.get("messages", [])
    if messages:
        last_msg = messages[-1]
        user_query = last_msg.get("content", "") if isinstance(last_msg, dict) else last_msg.content
    else:
        user_query = ""
    return {"user_query": user_query}

def intent_recognition_node(state: AgentState) -> dict:
    """意图识别节点：判断用户意图"""
    query = state["user_query"]
    
    system_prompt = """你是一个意图识别专家。根据用户输入，判断意图类型。
    返回以下类别之一：order_query（订单查询）、email_generate（邮件生成）、knowledge_qa（知识库问答）、general（通用对话）。
    只返回类别名称，不要有其他内容。"""
    
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=query)
    ])
    
    intent = response.content.strip().lower()
    return {"intent": intent}

def route_by_intent(state: AgentState) -> Literal["order_query", "email_generate", "knowledge_qa", "general"]:
    """路由函数：根据意图决定下一个节点"""
    intent = state.get("intent", "general")
    return intent

def order_query_node(state: AgentState) -> dict:
    """订单查询节点"""
    query = state["user_query"]
    
    # 模拟提取订单号（实际应使用正则或NER）
    import re
    order_match = re.search(r'订单\s*[号#]?\s*([A-Z0-9]{8,})', query)
    order_id = order_match.group(1) if order_match else "未知"
    
    # 模拟调用订单API
    # 实际场景：httpx.get(f"http://order-service/api/orders/{order_id}")
    status = "已发货" if order_id != "未知" else "未找到"
    is_expired = False  # 模拟判断
    
    return {
        "order_id": order_id,
        "order_status": status,
        "is_expired": is_expired,
        "action_result": f"订单{order_id}当前状态：{status}"
    }

def email_generate_node(state: AgentState) -> dict:
    """邮件生成节点"""
    order_id = state.get("order_id", "未知")
    status = state.get("order_status", "未知")
    
    prompt = f"""根据以下信息生成一封催单邮件：
    订单号：{order_id}
    当前状态：{status}
    要求：语气礼貌、专业，包含订单号、当前状态、以及请求尽快处理的诉求。"""
    
    response = llm.invoke([
        SystemMessage(content="你是一个专业的商务邮件撰写助手。"),
        HumanMessage(content=prompt)
    ])
    
    return {"action_result": response.content}

def knowledge_qa_node(state: AgentState) -> dict:
    """知识库问答节点（RAG）"""
    query = state["user_query"]
    
    # 模拟RAG检索（实际应连接向量数据库）
    # docs = vector_store.similarity_search(query, k=3)
    # context = "\n".join([doc.page_content for doc in docs])
    
    # 简化版：直接让LLM回答
    response = llm.invoke([
        SystemMessage(content="你是一个知识渊博的客服助手，请根据你的知识回答用户问题。"),
        HumanMessage(content=query)
    ])
    
    return {"action_result": response.content}

def general_node(state: AgentState) -> dict:
    """通用对话节点"""
    query = state["user_query"]
    response = llm.invoke([HumanMessage(content=query)])
    return {"action_result": response.content}

def summary_node(state: AgentState) -> dict:
    """汇总节点：生成最终回答"""
    result = state.get("action_result", "抱歉，我无法处理您的请求。")
    return {"final_answer": result}

2.3 构建Graph并添加检查点

# agent/graph.py (续)
def build_agent_graph():
    """构建完整的Agent工作流图"""
    # 创建StateGraph
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("entry", entry_node)
    workflow.add_node("intent", intent_recognition_node)
    workflow.add_node("order_query", order_query_node)
    workflow.add_node("email_generate", email_generate_node)
    workflow.add_node("knowledge_qa", knowledge_qa_node)
    workflow.add_node("general", general_node)
    workflow.add_node("summary", summary_node)
    
    # 设置入口
    workflow.set_entry_point("entry")
    
    # 添加边
    workflow.add_edge("entry", "intent")
    
    # 条件路由：根据意图分流
    workflow.add_conditional_edges(
        "intent",
        route_by_intent,
        {
            "order_query": "order_query",
            "email_generate": "email_generate",
            "knowledge_qa": "knowledge_qa",
            "general": "general",
        }
    )
    
    # 所有分支最终汇聚到summary
    workflow.add_edge("order_query", "summary")
    workflow.add_edge("email_generate", "summary")
    workflow.add_edge("knowledge_qa", "summary")
    workflow.add_edge("general", "summary")
    
    workflow.add_edge("summary", END)
    
    # 使用MemorySaver实现状态持久化（支持断点续传）
    memory = MemorySaver()
    
    return workflow.compile(checkpointer=memory)

# 全局Agent实例
agent = build_agent_graph()

LangGraph的StateGraph用一张有向图把“分支判断、循环重试、人工审批、断点续传”这四件事一次性解决了。工作流启动后，LangGraph用状态机管理执行：从入口节点开始，调用节点函数并传入当前状态，节点返回增量更新而非替换整个状态，LangGraph拿到更新后原子性地合并。

2.4 FastAPI服务端：流式响应

# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
from typing import List
import asyncio
import json

from agent.graph import agent, AgentState

app = FastAPI(title="AIGC Agent Service")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

class ChatRequest(BaseModel):
    messages: List[dict]
    thread_id: str  # 会话ID，用于状态隔离

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    """SSE流式对话接口"""
    
    async def event_generator():
        # 提取用户最新消息
        user_msg = request.messages[-1]["content"] if request.messages else ""
        
        # 构建初始状态
        initial_state = {
            "messages": request.messages,
            "user_query": user_msg,
            "order_id": None,
            "order_status": None,
            "is_expired": None,
            "action_result": None,
            "final_answer": "",
            "error": None,
            "retry_count": 0,
        }
        
        config = {"configurable": {"thread_id": request.thread_id}}
        
        try:
            # 发送开始事件
            yield {
                "event": "start",
                "data": json.dumps({"status": "processing", "thread_id": request.thread_id})
            }
            
            # 执行Agent工作流
            # 使用astream_events获取流式事件
            async for event in agent.astream_events(initial_state, config, version="v2"):
                kind = event["event"]
                
                if kind == "on_chat_model_stream":
                    # LLM流式输出
                    chunk = event["data"]["chunk"]
                    if hasattr(chunk, "content") and chunk.content:
                        yield {
                            "event": "token",
                            "data": json.dumps({"content": chunk.content})
                        }
                
                elif kind == "on_chain_end":
                    # 节点执行完成
                    if "output" in event["data"]:
                        output = event["data"]["output"]
                        if "final_answer" in output:
                            # 最终答案已生成
                            yield {
                                "event": "done",
                                "data": json.dumps({"final_answer": output["final_answer"]})
                            }
            
            # 发送结束事件
            yield {
                "event": "end",
                "data": json.dumps({"status": "completed"})
            }
            
        except Exception as e:
            yield {
                "event": "error",
                "data": json.dumps({"error": str(e)})
            }
    
    return EventSourceResponse(event_generator())

@app.get("/api/health")
async def health():
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

三、前端实现：React + SSE流式交互

3.1 项目初始化

# 使用Vite创建React项目
npm create vite@latest frontend -- --template react-ts
cd frontend
npm install
npm install -D tailwindcss postcss autoprefixer
npx tailwindcss init -p

3.2 自定义SSE Hook

// hooks/useSSE.ts
import { useState, useCallback, useRef, useEffect } from 'react';

interface SSEMessage {
  event: string;
  data: any;
}

interface UseSSEOptions {
  onMessage?: (msg: SSEMessage) => void;
  onError?: (error: Event) => void;
  onOpen?: () => void;
}

export function useSSE(options: UseSSEOptions = {}) {
  const [isConnected, setIsConnected] = useState(false);
  const [error, setError] = useState<Event | null>(null);
  const eventSourceRef = useRef<EventSource | null>(null);
  const reconnectTimerRef = useRef<number | null>(null);

  const connect = useCallback((url: string) => {
    // 关闭已有连接
    if (eventSourceRef.current) {
      eventSourceRef.current.close();
    }

    const eventSource = new EventSource(url);
    eventSourceRef.current = eventSource;

    eventSource.onopen = () => {
      setIsConnected(true);
      setError(null);
      options.onOpen?.();
    };

    eventSource.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        options.onMessage?.({ event: 'message', data });
      } catch (e) {
        console.error('Failed to parse SSE message:', e);
      }
    };

    eventSource.onerror = (err) => {
      setError(err);
      setIsConnected(false);
      options.onError?.(err);
      
      // 自动重连（指数退避）
      if (reconnectTimerRef.current) {
        window.clearTimeout(reconnectTimerRef.current);
      }
      reconnectTimerRef.current = window.setTimeout(() => {
        connect(url);
      }, 3000);
    };

    return () => {
      eventSource.close();
    };
  }, [options]);

  const disconnect = useCallback(() => {
    if (eventSourceRef.current) {
      eventSourceRef.current.close();
      eventSourceRef.current = null;
    }
    if (reconnectTimerRef.current) {
      window.clearTimeout(reconnectTimerRef.current);
      reconnectTimerRef.current = null;
    }
    setIsConnected(false);
  }, []);

  useEffect(() => {
    return () => {
      disconnect();
    };
  }, [disconnect]);

  return { connect, disconnect, isConnected, error };
}

3.3 流式对话组件

对于流式对话场景，更推荐使用Fetch API的ReadableStream来解析SSE事件，因为它支持POST请求携带更多参数。

// hooks/useChatStream.ts
import { useState, useCallback, useRef } from 'react';

interface Message {
  role: 'user' | 'assistant' | 'system';
  content: string;
}

export function useChatStream() {
  const [messages, setMessages] = useState<Message[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const [currentResponse, setCurrentResponse] = useState('');
  const abortControllerRef = useRef<AbortController | null>(null);

  const sendMessage = useCallback(async (content: string, threadId: string) => {
    // 添加用户消息
    const userMessage: Message = { role: 'user', content };
    setMessages(prev => [...prev, userMessage]);
    setIsStreaming(true);
    setCurrentResponse('');

    abortControllerRef.current = new AbortController();

    try {
      const response = await fetch('http://localhost:8000/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          messages: [...messages, userMessage],
          thread_id: threadId,
        }),
        signal: abortControllerRef.current.signal,
      });

      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }

      const reader = response.body?.getReader();
      const decoder = new TextDecoder();
      let buffer = '';

      // 解析SSE流
      while (reader) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const dataStr = line.slice(6);
            try {
              const data = JSON.parse(dataStr);
              
              if (data.content) {
                // 累积token
                setCurrentResponse(prev => prev + data.content);
              }
              
              if (data.final_answer) {
                // 最终答案
                const assistantMessage: Message = {
                  role: 'assistant',
                  content: data.final_answer,
                };
                setMessages(prev => [...prev, assistantMessage]);
                setCurrentResponse('');
              }
              
              if (data.error) {
                console.error('Server error:', data.error);
              }
            } catch (e) {
              // 忽略非JSON数据
            }
          }
        }
      }
    } catch (error: any) {
      if (error.name === 'AbortError') {
        console.log('Request aborted');
      } else {
        console.error('Stream error:', error);
      }
    } finally {
      setIsStreaming(false);
      abortControllerRef.current = null;
    }
  }, [messages]);

  const stopStream = useCallback(() => {
    if (abortControllerRef.current) {
      abortControllerRef.current.abort();
    }
  }, []);

  return { messages, isStreaming, currentResponse, sendMessage, stopStream };
}

3.4 聊天界面组件

// components/ChatInterface.tsx
import React, { useState, useRef, useEffect } from 'react';
import { useChatStream } from '../hooks/useChatStream';

const ChatInterface: React.FC = () => {
  const [input, setInput] = useState('');
  const [threadId] = useState(() => `session_${Date.now()}`);
  const { messages, isStreaming, currentResponse, sendMessage, stopStream } = useChatStream();
  const messagesEndRef = useRef<HTMLDivElement>(null);

  const scrollToBottom = () => {
    messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
  };

  useEffect(() => {
    scrollToBottom();
  }, [messages, currentResponse]);

  const handleSend = async () => {
    if (!input.trim() || isStreaming) return;
    const msg = input.trim();
    setInput('');
    await sendMessage(msg, threadId);
  };

  const handleKeyDown = (e: React.KeyboardEvent) => {
    if (e.key === 'Enter' && !e.shiftKey) {
      e.preventDefault();
      handleSend();
    }
  };

  return (
    <div className="flex flex-col h-screen max-w-3xl mx-auto p-4">
      <div className="flex-1 overflow-y-auto space-y-4 mb-4">
        {messages.map((msg, idx) => (
          <div
            key={idx}
            className={`flex ${msg.role === 'user' ? 'justify-end' : 'justify-start'}`}
          >
            <div
              className={`max-w-[80%] rounded-lg px-4 py-2 ${
                msg.role === 'user'
                  ? 'bg-blue-500 text-white'
                  : 'bg-gray-200 dark:bg-gray-700 text-gray-900 dark:text-white'
              }`}
            >
              <div className="whitespace-pre-wrap">{msg.content}</div>
            </div>
          </div>
        ))}
        
        {/* 流式响应中的临时内容 */}
        {isStreaming && currentResponse && (
          <div className="flex justify-start">
            <div className="max-w-[80%] rounded-lg px-4 py-2 bg-gray-200 dark:bg-gray-700">
              <div className="whitespace-pre-wrap">
                {currentResponse}
                <span className="inline-block w-2 h-4 ml-1 bg-current animate-pulse">|</span>
              </div>
            </div>
          </div>
        )}
        
        <div ref={messagesEndRef} />
      </div>
      
      <div className="flex gap-2">
        <textarea
          className="flex-1 rounded-lg border border-gray-300 dark:border-gray-600 p-2 resize-none focus:outline-none focus:ring-2 focus:ring-blue-500"
          rows={2}
          value={input}
          onChange={(e) => setInput(e.target.value)}
          onKeyDown={handleKeyDown}
          placeholder="输入您的问题..."
          disabled={isStreaming}
        />
        <button
          className={`px-4 py-2 rounded-lg font-medium ${
            isStreaming
              ? 'bg-red-500 hover:bg-red-600 text-white'
              : 'bg-blue-500 hover:bg-blue-600 text-white'
          }`}
          onClick={isStreaming ? stopStream : handleSend}
        >
          {isStreaming ? '停止' : '发送'}
        </button>
      </div>
      
      <div className="text-xs text-gray-400 mt-2">
        会话ID: {threadId}
      </div>
    </div>
  );
};

export default ChatInterface;

---

四、工程化落地：从开发到生产

4.1 Docker容器化部署

# Dockerfile (后端)
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

ENV PYTHONUNBUFFERED=1
EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# Dockerfile (前端)
FROM node:20-alpine as builder
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build

FROM nginx:alpine
COPY --from=builder /app/dist /usr/share/nginx/html
EXPOSE 80
CMD ["nginx", "-g", "daemon off;"]

# docker-compose.yml
version: '3.8'
services:
  backend:
    build: ./backend
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./backend:/app
    restart: unless-stopped

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

4.2 CI/CD流水线

# .github/workflows/deploy.yml
name: Deploy AIGC Agent App

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          pip install pytest pytest-asyncio
          pip install -r backend/requirements.txt
      - name: Run tests
        run: pytest backend/tests/
      - name: Setup Node
        uses: actions/setup-node@v4
        with:
          node-version: '20'
      - name: Run frontend tests
        run: |
          cd frontend
          npm ci
          npm run test

  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Login to Docker Hub
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
      - name: Build and push backend
        uses: docker/build-push-action@v5
        with:
          context: ./backend
          push: true
          tags: ${{ secrets.DOCKER_USERNAME }}/aigc-agent-backend:latest
      - name: Build and push frontend
        uses: docker/build-push-action@v5
        with:
          context: ./frontend
          push: true
          tags: ${{ secrets.DOCKER_USERNAME }}/aigc-agent-frontend:latest

4.3 监控与告警

# monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 请求计数
request_counter = Counter(
    'agent_requests_total',
    'Total number of agent requests',
    ['intent', 'status']
)

# 响应时间
request_duration = Histogram(
    'agent_request_duration_seconds',
    'Agent request duration in seconds',
    ['intent']
)

# 当前活跃会话
active_sessions = Gauge(
    'agent_active_sessions',
    'Number of active agent sessions'
)

# 在节点中埋点
def track_request(intent: str, duration: float, status: str):
    request_counter.labels(intent=intent, status=status).inc()
    request_duration.labels(intent=intent).observe(duration)

4.4 日志与可观测性

# logging_config.py
import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "module": record.module,
            "message": record.getMessage(),
            "thread_id": getattr(record, "thread_id", None),
            "intent": getattr(record, "intent", None),
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

# 配置日志
logger = logging.getLogger("aigc-agent")
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

---

五、踩坑与最佳实践

5.1 状态管理的坑

超过60%的Agent生产事故都和状态管理有关。建议：

1. 使用检查点（Checkpointer） ：MemorySaver用于开发，生产环境使用SqliteSaver或PostgresSaver持久化状态
2. 状态版本控制：每次状态变更记录版本号，便于回滚和审计
3. 状态大小限制：避免在State中存储过大的数据（如完整文档），用引用ID替代

5.2 流式响应的坑

1. SSE连接超时：设置合适的heartbeat或发送keepalive事件
2. 前端解析鲁棒性：SSE数据可能分片到达，需要正确处理buffer
3. 错误恢复：连接断开后实现自动重连，使用指数退避策略

5.3 成本控制

1. Token计数：每次请求记录输入/输出token数，设置每日预算告警
2. 缓存策略：对常见问题使用Redis缓存LLM响应
3. 模型降级：复杂任务用GPT-4，简单任务用GPT-4o-mini

5.4 安全与合规

1. API Key管理：使用环境变量或Secrets Manager，绝不提交到代码仓库
2. 输入过滤：对用户输入进行敏感词过滤和注入防护
3. 审计日志：记录每次Agent调用的完整链路，满足合规要求

---

六、总结

本文从零构建了一个完整的AIGC全栈应用，核心 takeaways：

1. Agent工作流编排：LangGraph的StateGraph提供了比线性Chain更强大的分支、循环、状态管理能力，是生产级Agent应用的基础。

2. 全栈工程化：从后端的FastAPI+LangGraph到前端的React+SSE，再到Docker+CI/CD的部署链路，构成了完整的工程化闭环。

3. 流式交互体验：SSE是实现AIGC应用实时反馈的关键技术，配合前端的状态管理，能提供流畅的用户体验。

4. 可观测性：日志、监控、链路追踪是生产系统的必备能力，在开发阶段就应纳入考量。

AIGC应用开发正在从“调API”走向“编排工作流”，从“写代码”走向“工程化”。掌握Agent工作流编排和全栈工程化能力，将是AI时代全栈工程师的核心竞争力。