03 · 后端：FastAPI 与分层架构

一、为什么选择 FastAPI

三个理由：

1. 原生 async + SSE：Agent 需要流式输出，FastAPI 的 StreamingResponse 最好用。
2. 自带 OpenAPI：可直接通过 /docs 调试，面试演示也很方便。
3. Pydantic 校验：请求体自动校验，省去大量 if 判断。

二、目录即架构

app/ ├── config.py ├── llm/ │ ├── providers.py │ └── prompts.py ├── rag/ │ ├── loader.py │ ├── embedding.py │ ├── vectorstore.py │ └── retriever.py ├── agent/ │ ├── state.py │ ├── graph.py │ ├── tools.py │ ├── guardrails.py │ └── memory.py ├── observability/tracer.py ├── api/routes.py └── main.py

分层原则：llm / rag / agent 三层彼此不知道 HTTP 的存在，api 层只负责协议转换。好处是——即使换成 Java 实现，业务逻辑也是一模一样的（见第 09 篇）。

三、配置中心：一处读取环境变量

@dataclass class Settings: llm_provider: str = "openai" llm_base_url: str = "https://api.deepseek.com/v1" ... @classmethod def from_env(cls) -> "Settings": ...

settings = Settings.from_env()

三个细节：

1. 自动加载 .env：

   from dotenv import load_dotenv _BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(file))) load_dotenv(os.path.join(_BASE_DIR, ".env"), override=False)

2. 自动降级：未配置 Key 时，llm_provider 会自动从 openai 切换为 mock，embed_provider 会自动从 openai 切换为 hashing，从而保证永远不会因缺少配置而启动失败。

3. 所有环境变量均可覆盖：在 Docker / K8s 中可直接注入 env，无须修改代码。

四、LLM 适配层：屏蔽厂商差异

class BaseLLM: def chat(self, messages, *, temperature=None, json_mode=False, trace_id="", span_name="llm"): ... def stream(self, messages, *, temperature=None, trace_id=""): ...

OpenAICompatibleLLM 使用 OpenAI SDK，兼容 DeepSeek / Qwen / 通义 / vLLM / Ollama（它们都实现了 /v1/chat/completions）。

重试与主备降级

candidates = [self.model] + self.fallback_models for model in candidates: for attempt in range(self.max_retries + 1): try: return self._client.chat.completions.create(**kwargs) except Exception as e: if attempt < self.max_retries: time.sleep(0.6 * (2 ** attempt))

为什么需要主备降级：线上曾遇到主模型限流或故障，导致整个 Agent 不可用。配置 LLM_FALLBACK_MODELS=qwen-plus,deepseek-chat 后，即可自动切换。

Token 统计

每次调用都会将 usage 上报给 Tracer：

tracer.record_llm(trace_id, u["prompt"], u["completion"], model=resp.model)

前端右上角实时显示累计 Token——成本可见，才谈得上治理。

五、Prompt 模板：五段式 + 集中管理

PLANNER_TMPL = """{system_role}

【可用工具】 {tools}

【历史对话摘要】 {summary}

【用户问题】 {question}

【任务】 把问题拆解为 1~3 个可执行步骤……严格输出 JSON……” """

我将所有模板放在 llm/prompts.py，原因如下：

• 可版本化：修改 Prompt 无须修改业务代码，可灰度发布、可进行 A/B 测试。
• 可测试：抽离模板后，可以对同一模板进行回归测试。
• 可复用：Python 侧和 Java 侧使用同一套模板文本。

五段式结构：角色 → 目标 → 约束 → 输出格式 → 少样本示例。规划节点强制输出 JSON，并配备容错解析器（移除 Markdown 代码块、截取第一个 JSON 对象）：

def parse_json_safe(text): m = re.search(r"(?:json)?\s*(.*?)", text, re.S) ... start, end = raw.find("{"), raw.rfind("}")

模型偶尔会输出 ... ，这个解析器曾多次发挥作用。

六、API 设计

接口	方法	说明
/api/chat	POST	SSE 流式对话
/api/chat/sync	POST	非流式调用（curl / 定时任务 / 单测）
/api/knowledge/ingest	POST	知识库重建索引
/api/knowledge/search	POST	检索调试
/api/traces、/api/traces/{id}	GET	Trace 查询
/api/tools	GET	工具清单（向前端展示）
/api/sessions/{id}	GET/DELETE	会话记忆

SSE 事件类型：

trace_start / plan / tool_start / tool_end / reflect / hitl token(增量文本) / llm_usage / final / error / stream_end

七、一个踩坑：同步图 + async 服务会阻塞事件循环

LangGraph 的 invoke 是同步的，直接在 async def 中调用会导致整个事件循环卡死。

解法：在线程池中运行图任务 + 线程安全地回传事件。

def worker(): for event in runner.run(query, sink=sink): loop.call_soon_threadsafe(queue.put_nowait, event)

asyncio.get_running_loop().run_in_executor(None, worker)

sink 通过 loop.call_soon_threadsafe 将事件从工作线程推回主线程的 asyncio.Queue。这样多个用户同时提问时，也不会相互阻塞。

---

下一篇：04 · RAG 全流程：从切分到引用溯源

一、为什么选 FastAPI

三个理由：

1. 原生 async + SSE ：Agent 要流式输出，FastAPI 的 StreamingResponse 最好用
2. 自带 OpenAPI ：/docs 直接能调试，面试演示很方便
3. Pydantic 校验 ：请求体自动校验，省一堆 if

二、目录即架构

app/
├── config.py               
├── llm/
│   ├── providers.py        
│   └── prompts.py          
├── rag/
│   ├── loader.py           
│   ├── embedding.py        
│   ├── vectorstore.py      
│   └── retriever.py        
├── agent/
│   ├── state.py            
│   ├── graph.py            
│   ├── tools.py            
│   ├── guardrails.py       
│   └── memory.py           
├── observability/tracer.py 
├── api/routes.py
└── main.py

分层原则 ：llm / rag / agent 三层彼此不知道 HTTP 的存在， api 层只做协议转换。好处是——换成 Java 实现时，业务逻辑是一模一样的 （见第 09 篇）。

三、配置中心：一处读环境变量

@dataclass
class Settings:
    llm_provider: str = "openai"
    llm_base_url: str = "https://api.deepseek.com/v1"
    ...
    @classmethod
    def from_env(cls) -> "Settings": ...

settings = Settings.from_env()

三个细节：

1. 自动加载.env：

from dotenv import load_dotenv
_BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
load_dotenv(os.path.join(_BASE_DIR, ".env"), override=False)

2. 自动降级 ：没配 Key 时 llm_provider 自动从 openai 变 mock， embed_provider 自动从 openai 变 hashing——保证永远不会因为缺配置而启动失败 。

3. 全环境变量可覆盖 ：Docker / K8s 里直接注入 env，不用改代码。

四、LLM 适配层：屏蔽厂商差异

class BaseLLM:
    def chat(self, messages, *, temperature=None, json_mode=False, trace_id="", span_name="llm"): ...
    def stream(self, messages, *, temperature=None, trace_id=""): ...

OpenAICompatibleLLM 走 OpenAI SDK，DeepSeek / Qwen / 通义 / vLLM / Ollama 全兼容 （它们都实现了 /v1/chat/completions）。

重试与主备降级

candidates = [self.model] + self.fallback_models   
for model in candidates:
    for attempt in range(self.max_retries + 1):    
        try:
            return self._client.chat.completions.create(**kwargs)
        except Exception as e:
            if attempt < self.max_retries:
                time.sleep(0.6 * (2 ** attempt))

为什么要主备降级 ：线上遇到过主模型限流/故障，整个 Agent 就不可用了。 配一个 LLM_FALLBACK_MODELS=qwen-plus,deepseek-chat 就能自动切。

Token 统计

每次调用把 usage 上报 Tracer：

tracer.record_llm(trace_id, u["prompt"], u["completion"], model=resp.model)

前端右上角实时显示累计 token——成本可见，才谈得上治理 。

五、Prompt 模板：五段式 + 集中管理

PLANNER_TMPL = """{system_role}

【可用工具】
{tools}

【历史对话摘要】
{summary}

【用户问题】
{question}

【任务】
把问题拆解为 1~3 个可执行步骤……严格输出 JSON……”
"""

我把所有模板放在 llm/prompts.py，理由：

• 可版本化 ：改 prompt 不改业务代码，能灰度、能 A/B
• 可测试 ：把模板抽出来后，能对同一模板做回归
• 可复用 ：Python 侧和 Java 侧用的是同一套模板文本

五段式结构 ：角色 → 目标 → 约束 → 输出格式 → 少样本示例。 规划节点强制输出 JSON，并配了容错解析器（去 markdown 代码块、截取首个 JSON 对象）：

def parse_json_safe(text):
    m = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)  
    ...
    start, end = raw.find("{"), raw.rfind("}")              

模型偶尔会输出 json ... ，这个解析器救过很多次。

六、API 设计

接口	方法	说明
/api/chat	POST	SSE 流式对话
/api/chat/sync	POST	非流式（curl / 定时任务 / 单测）
/api/knowledge/ingest	POST	知识库重建索引
/api/knowledge/search	POST	检索调试
/api/traces、/api/traces/{id}	GET	Trace 查询
/api/tools	GET	工具清单（暴露给前端展示）
/api/sessions/{id}	GET/DELETE	会话记忆

SSE 事件类型：

trace_start / plan / tool_start / tool_end / reflect / hitl
token(增量文本) / llm_usage / final / error / stream_end

七、一个踩坑：同步图 + async 服务会阻塞事件循环

LangGraph 的 invoke 是同步 的，直接在 async def 里调用会把整个事件循环卡死。

解法：线程池跑图 + 线程安全回传事件

def worker():
    for event in runner.run(query, sink=sink):
        loop.call_soon_threadsafe(queue.put_nowait, event)

asyncio.get_running_loop().run_in_executor(None, worker)

sink 用 loop.call_soon_threadsafe 把事件从工作线程推回主线程的 asyncio.Queue。 这样多个用户同时提问也不会互相阻塞。

---

下一篇：04 · RAG 全流程：从切分到引用溯源