上篇讲了 ReAct 循环，提到了流式输出的两阶段设计。这篇展开讲 SSE 流式对话和 Redis 会话记忆的完整实现。SSE 流式对话 和 Redis 会话记忆 是构建 AI Agent 系统的两个核心工程组件，本文将提供可直接运行的代码示例和设计思路。

为什么选 SSE 而不是 WebSocket

AI 对话场景的特点是：服务端推送为主，客户端只发一次。用户发一条消息 → Agent 处理 → 逐 token 返回。几乎不需要客户端高频推送。

特性	SSE	WebSocket
方向	单向（服务端→客户端）	双向
协议	HTTP（天然支持代理/负载均衡）	独立协议（需要 Upgrade）
重连	浏览器自动	需要自己实现
复杂度	低	高

SSE 的自动重连对 AI 对话特别有用——用户从电梯里出来网络恢复了，SSE 断掉的连接自动续上。WebSocket 得自己写重连逻辑。

流式的两阶段设计

上一篇文章提过，这里展开讲实现。

阶段一：工具调用（非流式）

POST /chat/stream → Agent Loop 启动
    ↓
LLM 返回 tool_calls → 执行工具 → 结果回传
    ↓
LLM 再返回 tool_calls → 再执行 → 再回传
    ↓
...直到 LLM 返回 finish_reason="stop"

这个阶段不能用流式——因为要拿到完整的 tool_calls JSON 才知道 LLM 想调哪个工具、传什么参数。逐 token 收的话没法解析。

阶段二：最终回答（流式）

async for chunk in adapter.chat_stream(messages, temperature, max_tokens):
    yield ("delta", chunk)  # 每收到一个 token 就推送给前端

前端收到的 SSE 事件流：

event: start        → {"session_id": "xxx"}
event: tool_call    → {"tool_name": "get_datetime", "arguments": {...}, "result": {...}}
event: delta        → {"content": "今天是"}
event: delta        → {"content": "2026年"}
event: delta        → {"content": "7月8日"}
event: done         → {"message": "完整消息...", "token_usage": {...}}

客户端断连检测

用户关掉浏览器标签页时，Agent 可能还在跑工具调用。不处理的话服务端空转浪费资源：

async for event_type, data in _run_agent_loop_stream(...):
    if await http_request.is_disconnected():
        logger.info(f"Client disconnected: {session_id}")
        return  # 立即停止生成

在每个 yield 之前检查一次。is_disconnected() 是 FastAPI 提供的方法，通过检测底层 TCP 连接状态来判断。

会话记忆：Redis 双层存储

对话记忆不是说把聊天记录存个 list 就完事了。有几个工程问题：

1. 读多写少：用户发一条消息，后续每轮对话都要读历史。读远多于写
2. 并发安全：一个会话同时来两个请求，消息 append 不能互相覆盖
3. 数据生命周期：热数据（最近 7 天）要快，冷数据（更早的）要省钱

Redis 热层

# Key: session:{session_id}
# Value: JSON
# TTL: 7天
{
  "session_id": "xxx",
  "messages": [
    {"role": "user", "content": "今天几号", "timestamp": 1234567890.0},
    {"role": "assistant", "content": "今天是7月8日", "timestamp": 1234567891.0},
    ...
  ],
  "metadata": {},
  "created_at": 1234567890.0,
  "updated_at": 1234567890.0
}

为什么选 JSON 而不是 Redis Hash？消息是个变长数组，Hash 里存数组很别扭。JSON 一个 key 存整个会话，读写一次往返。

消息追加

async def add_message(self, session_id, role, content):
    session = await self.get_session(session_id)    # 读
    session["messages"].append({role, content, ts}) # 改
    await self._save_session(session_id, session)    # 写

追加消息 = 读 → append → 写回。如果 session 不存在，自动创建——这样前端不需要先调 /session/create 再发消息，一个 API 搞定。

TTL 为什么是 7 天

7 天 = 一个工作周 + 周末。用户周一聊完，下周一来还能接着聊。超过了大概率已经忘了这回事。但数据不丢——Phase 4 会加上 PostgreSQL 冷持久化，Redis 过期的数据从 PG 恢复。

会话列表怎么查

Redis 没有"查所有 session:* 开头的 key"的原生方法，得用 SCAN：

async def list_sessions(self, limit=50):
    sessions = []
    async for key in self.redis.scan_iter(match="session:*", count=limit):
        data = await self.redis.get(key)
        sessions.append({session_id, message_count, created_at, updated_at})
    sessions.sort(key=lambda s: s["updated_at"], reverse=True)
    return sessions[:limit]

SCAN 是游标式迭代，不会阻塞 Redis 主线程。KEYS session:* 虽然简单但会阻塞——生产环境绝对不能用。

会话 API

端点	功能
POST /session/create	创建会话（前端也可以不调，发消息时自动创建）
GET /session/list	会话列表（按更新时间倒序）
DELETE /session/{id}	删除会话
GET /chat/{id}/history	查某会话的历史消息

完整代码示例

下面是一个完整的 FastAPI 应用示例，包含 SSE 流式对话和 Redis 会话记忆：

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
import redis.asyncio as redis
import json
import time
import asyncio
from typing import AsyncGenerator, Optional

app = FastAPI()
redis_client = redis.Redis(host="localhost", port=6379, db=0)

class SessionManager:
    """Redis 会话管理器"""
    
    def __init__(self, redis_client, ttl=604800):  # 7天 = 604800秒
        self.redis = redis_client
        self.ttl = ttl
    
    async def get_session(self, session_id: str) -> Optional[dict]:
        data = await self.redis.get(f"session:{session_id}")
        if data:
            return json.loads(data)
        return None
    
    async def create_session(self, session_id: str) -> dict:
        session = {
            "session_id": session_id,
            "messages": [],
            "metadata": {},
            "created_at": time.time(),
            "updated_at": time.time()
        }
        await self._save_session(session_id, session)
        return session
    
    async def add_message(self, session_id: str, role: str, content: str):
        session = await self.get_session(session_id)
        if not session:
            session = await self.create_session(session_id)
        session["messages"].append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
        session["updated_at"] = time.time()
        await self._save_session(session_id, session)
    
    async def _save_session(self, session_id: str, session: dict):
        await self.redis.setex(
            f"session:{session_id}",
            self.ttl,
            json.dumps(session)
        )
    
    async def list_sessions(self, limit: int = 50) -> list:
        sessions = []
        async for key in self.redis.scan_iter(match="session:*", count=limit):
            data = await self.redis.get(key)
            if data:
                session = json.loads(data)
                sessions.append({
                    "session_id": session["session_id"],
                    "message_count": len(session["messages"]),
                    "created_at": session["created_at"],
                    "updated_at": session["updated_at"]
                })
        sessions.sort(key=lambda s: s["updated_at"], reverse=True)
        return sessions[:limit]
    
    async def delete_session(self, session_id: str):
        await self.redis.delete(f"session:{session_id}")

session_manager = SessionManager(redis_client)

async def simulate_llm_stream(messages: list) -> AsyncGenerator[tuple, None]:
    """模拟 LLM 流式输出"""
    yield ("start", {"session_id": "demo_session"})
    
    # 模拟工具调用阶段
    yield ("tool_call", {
        "tool_name": "get_datetime",
        "arguments": {},
        "result": {"datetime": "2026年7月8日"}
    })
    
    # 模拟流式输出
    response_text = "今天是2026年7月8日，星期四。"
    for char in response_text:
        await asyncio.sleep(0.05)  # 模拟延迟
        yield ("delta", {"content": char})
    
    yield ("done", {
        "message": response_text,
        "token_usage": {"prompt_tokens": 50, "completion_tokens": 10}
    })

@app.get("/chat/stream")
async def chat_stream(request: Request, session_id: str, message: str):
    """SSE 流式对话端点"""
    # 保存用户消息
    await session_manager.add_message(session_id, "user", message)
    
    async def event_generator():
        async for event_type, data in simulate_llm_stream([]):
            if await request.is_disconnected():
                break
            yield f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
        
        # 保存助手回复
        await session_manager.add_message(session_id, "assistant", data.get("message", ""))
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"
        }
    )

@app.post("/session/create")
async def create_session(session_id: str):
    """创建新会话"""
    session = await session_manager.create_session(session_id)
    return {"session_id": session["session_id"], "created": True}

@app.get("/session/list")
async def list_sessions(limit: int = 50):
    """获取会话列表"""
    sessions = await session_manager.list_sessions(limit)
    return {"sessions": sessions, "count": len(sessions)}

@app.delete("/session/{session_id}")
async def delete_session(session_id: str):
    """删除会话"""
    await session_manager.delete_session(session_id)
    return {"deleted": True}

@app.get("/chat/{session_id}/history")
async def get_chat_history(session_id: str):
    """获取会话历史"""
    session = await session_manager.get_session(session_id)
    if not session:
        raise HTTPException(status_code=404, detail="Session not found")
    return {"session_id": session_id, "messages": session["messages"]}

总结

SSE + 会话记忆这块大约 430 行 Python，包括：

• SSE 两阶段流式设计
• 客户端断连检测
• Redis JSON 会话存储
• SCAN 游标列表查询
• TTL 7 天生命周期

配合 LLM 适配层和 ReAct 循环，Phase 1 的"用户输入 → Agent 处理 → 流式返回"闭环就完整了。

下一篇写知识库——怎么把企业文档变成 AI 能查的向量。