一、LLM 应用的 API 设计挑战

LLM 应用的 API 与传统的 CRUD API 有本质区别：

使用者痛点：响应时间不确定、模型初始化耗时、上下文丢失、幻觉导致答案不可用，这些都是在生产环境中最常被投诉的问题。不过，

这篇文章基于 LexRAG 和 HeritageMind 两个项目的 FastAPI 实践。程序讲解 LLM 应用的 API 工程化设计。

二、项目结构约定

两个项目都遵循相同的模块组织方式：

project/
├── api.py # FastAPI 应用入口
├── config.py # Pydantic Settings 配置
├── main.py # Streamlit 前端
│
├── src/
│ ├── agent/ # Agent 与工作流
│ ├── retrieval/ # 检索模块
│ ├── graph/ # 知识图谱
│ └── ...
├── data/ # 数据文件
└── tests/ # 测试

主要原则：api.py 只做路由和请求响应处理。业务逻辑全部在. 这保持了的简洁，方便维护和测试。

三、全局组件管理与 Lifespan

至于问题，模型不能每个请求都加载一次（Pain Point：内存爆炸 & 响应慢)

# ❌ 致命错误：每个请求重新加载模型
@app.post
async def query:
embeddings = BGEEmbeddings # 加载 .3GB 模型！retriever = LawRetriever # 加载向量索引
workflow = LegalRAGWorkflow
return workflow.query
# 每个请求耗时 30s+，内存爆炸

再看方法，Lifespan + 全局单例

from contextlib import asynccontextmanager
from fastapi import FastAPI
# ── 全局组件单例 ──
components: dict = {}
@asynccontextmanager
async def lifespan:
"""应用生命周期管理：
- yield 前：启动时初始化
- yield 后：关闭时清理"""
logger.info
说到try，# . 加载配置
settings = Settings
# . 初始化嵌入模型
embeddings = EmbeddingManager
# . 初始化检索器
law_indexer = LawIndexer
law_retriever = LawRetriever
bm25_retriever = BM25Retriever if settings.bm25_enabled else None
reranker = CrossEncoderReranker if settings.reranker_enabled else None
# . 初始化知识图谱
law_graph = LawGraph.from_json
# . 初始化工作流
workflow = LegalRAGWorkflow(
llm=ChatOpenAI,retriever=law_retriever,bm25=bm25_retriever,reranker=reranker。graph=law_graph,)
# . 存入全局容器
components = settings
...
except Exception as e:
...
yield # ← 应用在这里运行
app = FastAPI(
title="LexRAG API",description="法律 RAG 知识问答程序",version="0.1.0",lifespan=lifespan,)

关键设计决策：

• # ✅ 正确：通过 lifespan 初始化，存入全局字典 components = LegalRAGWorkflow
• # ❌ 错误：使用模块级变量 workflow = None  # 可能在 lifespan 之前被访问
• # ❌ 错误：使用 Depends 注入 async def get_workflow: return LegalRAGWorkflow

延迟加载模式

对于不需要在启动时就用的组件，使用延迟加载减少启动时间：

class EmbeddingManager:
def __init__:
...
@property
def model:
...

Pain Point Highlight:


"API 容器可以快速启动并响应健康检查，模型仅在实际查询到来时才加载，避免了冷启动导致的超时错误。"


四、端点设计 

完整的端点地图

LexRAG 为主要功能提供了以下七类端点：

@app.get
async def root:
"""程序信息"""
return {
"message": "LexRAG - 法律知识提高检索程序","version": "0.1.0","docs": "/docs"。}
@app.get
async def health_check:
"""健康检查 —— Docker & LB 使用"""
...
@app.post
async def query:
"""法律问答 —— 主要端点"""
...
@app.post
async def upload_document:
"""上传法律文档并索引"""
...
@app.get
async def get_graph_stats:
...
@app.post
async def query_graph:
...
@app.get
async def get_verification:
...
@app.get
async def get_intent_examples:
...

RESTful 命名原则 & 痛点对应表格化展示 
`

`


原则
示例
说明 & 痛点缓解说明 

GET 用于读取
/graph
读取统计信息，无副作用；避免误触发计算导致延迟，

POST 用于有副作用或计算密集型操作
/graph/query
查询图谱方法会触发计算。需要 POST 明确意图，防止浏览器预取造成不必要负载。

资源名使用名词
/query
统一口径，让前端 SDK 能自动生成类型安全代码。

子资源层级嵌套
/verify/{query_id}
清晰表达查询历史关联，便于日志追踪和审计。

批量/特化操作标记
/graph vs /graph/query
返回全量 vs 部分结果，降低前端分页实现复杂度。



五、请求与响应模型 

Pydantic 模型分层 & 使用者痛点——字段校验缺失导致异常 
 from pydantic import BaseModel。Field
from typing import Optional,List,Dict
## 请求模型
class QueryRequest:
"""查询请求"""
query : str = Field(...,min_length=1,max_length=1024,description='法律问题文本',example='合同在什么情况下可以撤销？')
session_id : Optional = Field(default=None,description='会话 ID，用于多轮对话追溯')
config : Optional = Field(default=None。description='查询配置覆盖')
 class Config:
json_schema_extra ={ 'example':{'query':'合同在什么情况下可以撤销？','session_id':'sess_abc123'} }


class QueryConfig:
topk : Optional = Field
temperature : Optional = Field
enableverification : bool = True

## 响应模型
class QueryResponse:
queryid : str
query : str
answer : str
intent : str
citations : List=
verificationstatus : str # PASS|FAIL|SKIPPED
issues : List=
sub_questions : List=
metadata : Dict= {}

class CitationInfo:
article_id : str # '民法典第157条'
...


Pain Point: 若不加限制。使用者可能提交空字符串或超过 token 限制，引发后端 OOM 或 LLM 报错。通过 Pydantic 的 `min_length`/`max_length` / `ge` / `le` 可以在网关层提前拦截。

统一响应格式 
 from fastapi.responses import JSONResponse
def success:
return JSONResponse(status_code=200。content={ 'code':0,'message':'OK','data':data })

This pattern让前端统一错误处理，提高开发效率，也解决了“不同接口返回结构不一致导致前端调试困难”的痛点。

六、错误处理与异常程序 

分层异常设计 
 ## 基础异常
class LexRAGException:
def __init__:
self.message=msg;self.detail=detail or {}
class RetrievalException: pass
class LLMException: pass
class VerificationException: pass
class DocumentParseException: pass

全局异常处理器 
 @app.exception_handler
async def lexrag_exception_handler:
return JSONResponse(status_code=400,content={'error':exc.message,'type':type.__name__,'detail':exc.detail})
@app.exception_handler
async def llm_exception_handler:
status_code = 502 if 'timeout' in str.lower else 500
return JSONResponse(status_code=status_code。content={'error':exc.message,'type':'LLMException','retryable':status_code==502})
@app.exception_handler
async def general_exception_handler:
logger.error
return JSONResponse(status_code=500,content={'error':'Internal server error','type':type.__name__})

Pain Point 缓解： 当 LLM 超时或配额耗尽时返回 `retryable:true`，前端可以自动重试，而不是直接崩溃。

业务层优雅降级 
 @app.post
async def query:
wf = components.get
if not wf:
raise HTTPException
再看try。result = wf.query
except LLMException as e:
logger.warning
return success({
'query_id':str),'query':request.query,'answer':'抱歉，AI 推理暂不可用。

This design directly addresses “LLM 服务偶尔宕机导致整个程序不可用”的使用者投诉。

七、请求生命周期追踪 

请求 ID 注入 & 耗时日志 
 import uuid。time
@app.middleware
async def request_tracking_middleware:
request_id=str)
start=time.time
request.state.request_id=request_id
logger.info
response=await call_next
duration=-start)*1000
response.headers=request_id
logger.info")
return response

查询结果内存缓存供后续验证追溯 
 query_results:Dict={}
@app.post
async def query:
result=components.query
query_results=result
return success
@app.get
async def get_verification:
res=query_results.get
if not res:
raise HTTPException
return success

*Pain Point*: 在客户要求“提供查询过程审计日志”时这种轻量级缓存足以满足需求，无需额外数据库投入。

八、CORS 与安全配置 

CORS 配置 
 from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware。allow_origins=,# 开发阶段宽松策略
allow_credentials=True,allow_methods=,allow_headers=,)

Pain Point: 生产环境如果忘记收紧 CORS，会导致跨站脚本攻击风险。下面给出推荐写法：
 app.add_middleware(
CORSMiddleware,allow_origins=,allow_credentials=True,allow_methods=,allow_headers=,)

IDKey 防护 – 环境变量安全存储 
 from pydantic import BaseSettings。Field
class Settings:
deepseek_api_key:str=Field
 class Config:
env_file=".env"
env_file_encoding="utf-8"


.gitignore 必须包含 `.env` 文件，以免把密钥提交到仓库。

九、自动化测试 

IDTestClient 基础测试示例 

 from fastapi.testclient import TestClient
from api import app 


client=TestClient 

def testroot:
resp=client.get
assert resp.statuscode==200 

def test_health:
...


IDMock LLM 避免配额消耗 

 @pytest.fixture
def mockllm:
with patch as mock:
mock.returnvalue=QueryResponse(
queryid="test-001",query="合同成立条件？",answer="根据民法典第XX条..."。verificationstatus="PASS")
yield mock



十、自动生成的 API 文档 

A big advantage of FastAPI is that docs are generated automatically.



- Swagger UI ： http://localhost:/docs .
.

- ReDoc ： http://localhost:/redoc。.
.

Pydantic 的 Field 描述和 example 会直接渲染到 UI 中，让使用者一眼看懂输入要求。
.

E.g.,QueryRequest 中已写好 example：“合同在什么情况下可以撤销？”，Swagger 会自动填充示例值，降低学习成本。
.



Pain Point 缓解： 客户经常抱怨“没有文档不知道怎么调用”。只要保持字段描述完整，就能省去手动撰写外部文档的工夫。不过，

十一、两个项目的端点对比 

LlexRag 与 HeritageMind 均遵循 “最少可用端点” 原则。每个 endpoint 都对应明确业务需求，没有冗余接口，从而降低运维成本和故障面。

十二、设计原则 



- Lifespan 管理重量级资源；切勿在每个 request 中重复加载模型。

- 全局组件字典比更好 Depends 注入；后者会产生多实例，对大模型极其不友好。

- 异常分层并实现降级策略；确保 LLM 故障不会导致整个服务不可用。

- 请求唯一 ID 与结果缓存，实现可追溯审计。老实说，

- Pydantic Field 必须设定约束&示例；它们是防止恶意输入及提高文档质量的关键手段。

- 利用 FastAPI 自动生成 OpenAPI 文档，把“文档缺失”转变为免费优势。<\/ul>