Week 11：实现 + 调试

🎯 本周目标：把 Week 9-10 设计的骨架填充成可运行的完整应用。重点攻克流式输出、会话记忆、Prompt 调优。

1. 流式输出实现（重点！）

这是让用户体验从"等几秒"变成"逐字显示"的关键。

FastAPI SSE + LangGraph astream

# app/routes/chat.py
import json
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from app.models.schemas import ChatRequest
from app.agent.graph import app as agent_app

router = APIRouter(prefix="/api/chat", tags=["对话"])

@router.post("/stream")
async def stream_chat(req: ChatRequest):
    """Agent 流式对话接口"""

    async def event_generator():
        # 初始状态
        state = {
            "messages": [{"role": "user", "content": req.user_input}],
            "session_id": req.session_id or "new_session",
        }

        # LangGraph astream 会 yield 每个节点的输出事件
        # 需要转换成 SSE 格式：data: {json}\n\n
        async for event in agent_app.astream_events(
            state,
            version="v2",
        ):
            event_type = event.get("event")

            # 只关心 LLM 的 on_chat_model_stream（逐 token 流出）
            if event_type == "on_chat_model_stream":
                chunk = event["data"]["chunk"]
                content = chunk.content
                if content:
                    # SSE 格式
                    yield f"data: {json.dumps({'type': 'token', 'content': content}, ensure_ascii=False)}\n\n"

            # 工具调用事件
            elif event_type == "on_tool_start":
                tool_name = event["name"]
                yield f"data: {json.dumps({'type': 'tool_start', 'tool': tool_name}, ensure_ascii=False)}\n\n"

            elif event_type == "on_tool_end":
                yield f"data: {json.dumps({'type': 'tool_end'}, ensure_ascii=False)}\n\n"

        # 结束标记
        yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",    # Nginx 反代需要
        },
    )

前端 TS 调用示例

// TypeScript / 前端消费 SSE
async function streamChat(userInput: string, sessionId?: string) {
  const response = await fetch("/api/chat/stream", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ user_input: userInput, session_id: sessionId }),
  });

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n\n");  // SSE 用 \n\n 分隔事件
    buffer = lines.pop() || "";

    for (const line of lines) {
      const dataLine = line.split("\n").find(l => l.startsWith("data: "));
      if (!dataLine) continue;

      const payload = JSON.parse(dataLine.slice(6));

      switch (payload.type) {
        case "token":
          appendToChat(payload.content);    // 逐字追加到 UI
          break;
        case "tool_start":
          showToolIndicator(payload.tool);   // 显示"正在查询..."
          break;
        case "done":
          finalizeChat();
          break;
      }
    }
  }
}

💡 TS 对照：这就是 fetch + ReadableStream 的典型用法。SSE 是 HTTP 协议的一种长连接模式，不需要 WebSocket 那么重。

后端调试技巧

# 用 curl 测试流式接口（Windows PowerShell）
curl -X POST http://localhost:8000/api/chat/stream ^
  -H "Content-Type: application/json" ^
  -d "{\"user_input\": \"BTC 现在多少钱？\"}"

如果一切正常，你会看到输出是逐行追加的（不是等几秒后整块出来）。

2. 会话记忆（LangGraph Checkpointer）

让 Agent 能记住多轮对话的上下文。

SQLite 存储（轻量）

.venv\Scripts\python -m pip install langgraph-checkpoint-sqlite

给 LangGraph 加 Checkpointer

# app/agent/graph.py（修改 Week 9-10 的编译部分）

from langgraph.checkpoint.sqlite import SqliteSaver
from app.config import settings

# 创建 SQLite Checkpointer
checkpointer = SqliteSaver.from_conn_string(str(settings.session_db_path))

# 编译时把 checkpointer 传进去
app = graph.compile(checkpointer=checkpointer)

调用时传入 session_id

async def run_agent_with_memory(user_input: str, session_id: str):
    # session_id 对应对话线程 ID
    config = {"configurable": {"thread_id": session_id}}
    
    async for chunk in app.astream(
        {"messages": [{"role": "user", "content": user_input}]},
        config,
        stream_mode="messages",
    ):
        # 处理每个 chunk
        ...

TS 对照

LangGraph Checkpointer前端类比
thread_id = session_idlocalStorage.setItem('sessionId', xxx)
SQLite 持久化写数据库
下次调用传同样的 thread_id从 localStorage 读回 sessionId

3. Prompt 工程（持续调优）

好的 System Prompt 比调模型还重要。

推荐的 System Prompt 模板

# app/agent/prompts.py

TRADING_SYSTEM_PROMPT = """你是一个专业的量化交易 AI 助手，名字叫小A。

## 你的能力
1. 查询交易标的（股票/加密货币）的实时价格
2. 执行买入/卖出交易（Mock 模式，不会真的下单）
3. 获取市场新闻和分析
4. 回答关于本项目技术栈的问题
5. 和用户闲聊

## 你的行为规范
- 必要时主动调用工具获取真实数据，不要瞎编
- 回答要简洁、专业、友好
- 如果工具返回的数据不清楚，诚实地告诉用户
- 对交易相关的建议要加上风险提示

## 输出格式
- 用 Markdown 格式回答
- 价格数字用 `$` 符号包裹，如 `$99,999.99`
- 订单信息用列表展示

## 当前时间
{current_time}
"""

Prompt 版本管理

prompts/
├── system_v1.txt      # 第一版
├── system_v2.txt      # 加了风险提示
├── system_v3.txt      # 改了输出格式
└── current.txt        # 当前使用（gitignore 里可以排除）

在 prompts.py 里读取对应版本的文件，改 prompt 只改文件不碰代码。

4. 调试清单

#检查项方法
1FastAPI 启动成功访问 http://localhost:8000/docs 能看到 Swagger UI
2/health 返回正确curl http://localhost:8000/health
3Agent 非流式对话正常POST /api/chat 能返回完整回答
4Agent 流式对话正常POST /api/chat/stream 能逐字吐出
5工具调用正常问 "BTC 多少钱" 能触发 get_stock_price
6RAG 正常问项目相关问题能查到文档内容
7记忆正常同一个 session_id 多轮对话能记住上下文
8CORS 正常前端页面能正常调用后端

常见 Bug 和解决方案

Bug原因解决
流式不流，整块吐出Nginx 缓冲或没有设置 X-Accel-Buffering: no加 header + 关闭缓冲
工具调用后 Agent 不返回结果tools_condition 路由错了检查条件边的配置
多轮对话没记忆thread_id 每次都不一样前端要保存 session_id 并传回
LLM 一直调同一个工具死循环没设 tool_choice 限制加 recursion_limit 或检查 LLM 的判断逻辑
异步报错 run method is not supported with async db混用了同步/异步链路统一用 arun() / astream()
Swagger UI 调接口报 CORSFastAPI CORS 中间件没加加 CORSMiddleware

5. 本周任务清单

#任务验收
1实现 SSE 流式接口curl 能看到逐行输出
2加 LangGraph Checkpointer（SQLite）多轮对话能记住
3写 System Prompt 模板按模板对话风格一致
4把 Week 9-10 骨架里的所有文件填上实现所有接口都能用
5写一个简单的前端（HTML + vanilla JS）测试 SSE浏览器里能逐字看到回答
6通读一遍代码，把同步/异步调用全部统一没有混用