怎么说呢，

很多 Agent Demo 里模型调用通常是一段很直接的代码：

response = await client.chat.completions.create

这当然能跑。

但真实业务程序里问题很快就会变多：

写正文、做质检、拆大纲、提取记忆，要不要用同一个模型？模型返回空内容怎么办，JSON 解析失败怎么办？使用者连续点多次 AI 按钮怎么办？一次自动写作到底花了多少钱？测试环境会不会误打到真实模型？不过，模型配置改了服务要不要重启？

这些问题单看都不大。但放到 AI Agent 程序里它们会一起决定一件事：

下面结合当前的 AI 小说创作程序，拆解我在项目里实现的 AI 工程底座。

我现在不建议在业务代码里到处直接调用大模型。

更稳的做法是把模型调用收敛成一条链路：

对应到程序里大概是这样：

业务服务
↓ chat_text / chat_json / chat_json_with_tools
↓ get_model_for_task
↓ AsyncOpenAI
↓ 日志 / 成本 / 错误协议 / 限流

This article is not about “how to integrate a specific model vendor”. 我更想讲的是：

. 为什么不能把模型调用散落在业务里

一开始做 AI 功能，很容易写成这样：

质检接口里调一次模型
大纲接口里调一次模型
章节生成里调一次模型
人物生成里调一次模型
导出简介里调一次模型
...

每个地方都自己处理：

model temperature max_tokens timeout JSON 解析异常 捕获日志记录 成本统计
...

短期看很快；长期看会出现几个关键痛点：

Sooo。我在项目里做的第一件事，就是把模型调用收敛到统一网关。业务层不关心具体 provider，只表达意图：

我要做 writing
我要做 planning
我要做 quality
我要拿 JSON
我要允许工具调用
...
底座负责把这些意图翻译成具体模型调用。

. 配置层：provider 不是散落的字符串

项目里所有 AI 配置都集中在 app/core/config.py 。

MODEL_PRESETS = {
"xfyun": {
"ai_api_base": "https://maas-coding-api.cn-huabei-.xf-yun.com/v2","ai_model": "astron-code-latest",},"deepseek": {
"ai_api_base": "https://api.deepseek.com","ai_model": "deepseek-v4-pro",},"openai": {
"ai_api_base": "https://api.openai.com/v1","ai_model": "gpt-4o",},}
配置对象只保留通用字段：
python
class Settings:
"""应用全局配置。自动读取 .env / 环境变量"""
database_url: str = "sqlite+aiosqlite:///./novel_system.db"
ai_provider: str = "xfyun"
ai_api_base: str = ""
ai_api_key: str = ""
ai_model: str = ""
ai_test_mode: bool = False
model_config = {
"env_file": str,"env_file_encoding": "utf-8","extra": "ignore",}
def apply_preset:
"""根据 ai_provider 应用预设"""
preset = MODEL_PRESETS.get
if not self.ai_api_base:
self.ai_api_base = preset.get
if not self.ai_model:
self.ai_model = preset.get


使用者痛点不同使用者想使用自托管或公司内部私有化部署时需要灵活覆盖 base URL 与 model，而不是硬编码在每个业务函数里。

. 热重载：配置改完不应该重新启动 

自托管程序的管理页面经常会改动 model 配置。如果每次改完都要手动重启后端，会导致 **运营体验割裂**。提供了热重载函数：
def reload_settings:
"""热重载配置：重新读取 .env 并更新全局 settings 对象，无需重启"""
if not _ENV_PATH.exists:
return
env = {}
for line in _ENV_PATH.read_text.split:
line = line.strip
if "=" in line and not line.startswith:
k,v = line.split
env = v.strip
settings.ai_provider = env.get
settings.ai_api_base = env.get
settings.ai_api_key = env.get
settings.ai_model = env.get
settings.ai_test_mode= os.getenv(
"AI_TEST_MODE",env.get,).strip.lower in {"","true","yes"。"on"}
# 应用预设
preset = MODEL_PRESETS.get
if not settings.ai_api_base and preset.get:
settings.ai_api_base = preset
if not settings.ai_model and preset.get:
settings.ai_model = preset


管理接口保存后会立即执行：

python
@router.put
async def update_model_config:
"""更新模型配置 — 保存后即时生效，无需重启后端"""
...
_write_env
try的观点是，reload_settings
from app.core.ai_config import reset_ai_client
await reset_ai_client
reload_ok = True
except Exception:
reload_ok = False
...
 

使用者痛点运维人员不希望因一次 UI 配置修改而导致整套写作服务不可用。

. 模型路由：不是所有任务都该用同一个模型 

不同任务对 LLM 能力需求差异巨大。例如小说创作程序中：



任务 
更看重什么 



正文生成 
速度 + 成本 + 稳定输出质量 

审核 
推理 + 结构化判断 + 发现问题 

记忆提取 
信息抽取准确性 

创意发散 
想象力 + 推理深度 

大纲规划 
长上下文理解和结构能力 



如果全部使用最强的型号，成本失控；如果全部使用最快的型号，质量不足。在配置中维护“任务 → 模型”映射表：
_PROVIDER_TASK_MODEL_MAP = {
"deepseek": {
# 高质量/高费用场景
"analysis": "deepseek-v4-pro","review": "deepseek-v4-pro",# 高速/低费用场景
"writing": "deepseek-v4-flash",...
},...
}
获取对应模型的函数：
python
def get_model_for_task -> str:
"""。未匹配时用全局默认模型"""
task_map = _get_task_model_map
if task and task in task_map:
return task_map
return settings.ai_model
业务代码只传递 **任务意图**：
python
data = await chat_json(
system_prompt,user_prompt,temperature=0.7,max_tokens=1024,task="quality",novel_id=novel_id,chapter_id=chapter_id,)
**使用者痛点**：产品经理可以，而无需让开发去改所有业务代码。

. 统一网关：业务层只调用 chat_text / chat_json / chat_json_with_tools 

网关位于 app/services/ai_gateway.py 。下面展示文本入口的主要实现：
async def chat_text(
system_prompt: str,user_prompt: str,*,temperature: float = 0.7,max_tokens: int = 1024,timeout: float | None = None,task: str | None = None,novel_id: int | None = None,chapter_id: int | None = None,response_format: dict | None = None,) -> str:
client = get_ai_client


---省略部分组装---

EMPTYCOMPLETIONATTEMPTS = 2 # 可通过环境变量调节

content: str | None = None
for attempt in range:
try的观点是，response = await client.chat.completions.create
except Exception as exc:
logger.warning(
"ai.chat_text.failed",extra={"ai":{"model":model,"attempt":attempt,"error":type.name}}
)
raise AIServiceError from exc
 candidate = response.choices.message.content if response.choices else None
if candidate and candidate.strip:
content=candidate.strip
break
logger.warning(
"ai.chat_text.empty"。extra={"ai":{"model":model,"attempt":attempt,"will_retry":attempt+1 


if content is None:
raise AIServiceError

---日志 & 成本记录---
if novelid:
from app.services.costtracker import logcost
from app.core.database import asyncsession
async with asyncsession as db:
从try来看，await logcost(
db,novelid=novelid,chapterid=chapterid,model=model,tasktype=task or 'default',prompttext=systemprompt+userprompt,completion_text=content,)
except Exception:
pass # 成本日志失效不影响主流程

return content


使用者痛点前端只需要知道「我要生成正文」或「我要结构化 JSON」，而不必重复编写超时/空响应/成本统计等防护逻辑。
话说回来，

. JSON 调用：结构化输出统一入口 

大量任务需要机器返回合法 JSON。为避免每个模块自行实现繁琐且易错的解析逻辑，网关提供统一的 `extract_json_object` 与 `chat_json` 包装。其实，
def extract_json_object -> dict:
text=.strip
if text.startswith:
text=re.sub?\s*","",text)
text=re.sub
try这方面，return json.loads
except json.JSONDecodeError:
match=re.search
if match:
从try来看，return json.loads)
except json.JSONDecodeError:return {}
return {"raw":raw。"error":"JSON 解析失败"}
`chat_json` 在 `chat_text` 基础上加上 response_format并返回解析后的字典：
python
async def chat_json:
json_system_prompt=system_prompt
response_format=None
if settings.ai_provider=="deepseek":
json_system_prompt+= "
请严格返回合法 JSON 对象"
response_format={"type":"json_object"}
raw=await chat_text
data=extract_json_object
if data.get:
raise AIServiceError
return data
**使用者痛点**：前端页面可以直接依赖结构化数据，不必担心因为某家 provider 的细微差异导致 JSON 格式错乱。其实，

. 工具调用：Tool Calls 一样走统一循环 

工具调用是 Agent 的主要特性之一。项目提供 `chat_json_with_tools` 实现完整的 tool‑call 循环。并保留 `_tool_calls` trace，以便后续审计。
async def chat_json_with_tools(
system_prompt:str,user_prompt:str,*,tools:list],tool_handler:Callable],Awaitable]],temperature:float=0.7。max_tokens:int=1024,timeout:float|None=None,task:str|None=None,novel_id:int|None=None,chapter_id:int|None=None,max_rounds:int=5) -> dict:
client=get_ai_client;model=get_model_for_task
messages=
tool_trace=
for round_index in range:
request_kwargs=dict(
model=model,messages=messages,temperature=temperature,max_tokens=max_tokens,to​ols=tools。tool_choice="auto")
response=await client.chat.completions.create
message=response.choices.message if response.choices else None
if not message:
raise AIServiceError
tool_calls=list or )
if not tool_calls:
# 最终返回 JSON
content=message.content or ""
data=extract_json_object
data=tool_trace
return data
# --------- 处理工具请求 ----------
assistant_tool_calls=
for tc in tool_calls:
fn=getattr
name=getattr
args_raw=getattr or "{}"
assistant_tool_calls.append({
至于“id”，tc.id,“type”:“function”,“function”:{“name”:name,“arguments”:args_raw}})
messages.append
for tc in tool_calls:
fn=getattr;name=getattr
args=json.loadsor "{}")
result=await tool_handler # ← 使用者自定义实际业务逻辑
tool_trace.append({
“round”:round_index+1,“name”:name,“arguments”:args。“result”:result})
messages.append({
“role”的观点是，“tool”,“tool_call_id”:tc.id,“content”:json.dumps})
raise AIServiceError
**使用者痛点**：当 Agent 在生产环境卡住或循环过多次时可以快速定位是哪一步工具被错误调用还有对应参数。

. 测试模式：不要让测试误打真实模型 

在 CI/CD 或本地开发阶段。如果误用了生产 API，将产生 **不可接受的费用和隐私泄漏风险**。引入 `AI_TEST_MODE` 严格限制只能访问 **loopback 本地地址**。
class AITestModeError: pass
def validate_test_ai_base_url->str:
if not test_mode:return base_url
从try来看，parsed=urlparse
host=parsed.hostname.rstrip.lower
loopback= or ipaddress.ip_address.is_loopback
if not loopback:return raise ValueError
except :
raise AITestModeError from None
return base_url
def get_ai_client->AsyncOpenAI:
…validate_test_ai_base_url
if settings.ai_test_mode:
_http_client=httpx.AsyncClient
else…测试案例确保非回环地址被拒绝：
python
def test_test_mode_rejects_non_loopback_ai_base_url:
with pytest.raises:
validate_test_ai_base_url
@pytest.mark.parametrize("url"。)
def test_test_mode_accepts_loopback:
assert validate_test_ai_base_url==url
**使用者痛点**：QA 团队可以放心运行自动化脚本，而不会意外产生真实计费。

. 限流：不要等账单爆了才想起防抖 

AI 接口一次请求背后可能涉及数十秒的大上下文拼接与多轮工具循环。单纯依赖前端按钮禁用不足以防止恶意或误操作。怎么说呢，在 FastAPI 中加入 IP‑粒度 Token‑Bucket 限流中间件。
class RateLimitMiddleware:
"""Token‑bucket per‑IP rate limiter for AI endpoints."""
def __init__:
super.__init__
self.max_requests=int)
self.window_seconds=int)
self._buckets=defaultdict
async def dispatch:
path=request.url.path
if not path.startswith:
return await call_next
 ip=request.client.host if request.client else 'unknown'
now=time.time;cutoff=now-self.window_seconds
bucket= if t>cutoff]
self._buckets=bucket
if len>=self.max_requests:
return JSONResponse(
status_code=429。content={"detail":
f"请求过于频繁，请稍后再试",“retry_after”:int))}
)
bucket.append;self._buckets=bucket
return await call_next


使用者痛点即使出现「疯狂点击」或脚本攻击，也能保证后台不会因为突发的大量请求瞬间刷爆费用。

. 成本追踪：Agent 必须知道花了多少钱 

单次 API 调用只是成本的一小块；真正需要监控的是 **完整任务链路** 的累计消耗。


class CostLog: ...
id INTEGER PK
novel_id INTEGER FK → novels.id
chapter_id INTEGER FK → chapters.id
model STRING NOT NULL
task_type STRING DEFAULT 'writing'
prompt_tokens INTEGER DEFAULT 0
completion_tokens INTEGER DEFAULT 0
cost_rmb FLOAT DEFAULT 0.0
created_at DATETIME DEFAULT utcnow
​
​
​
​
​
​
​
​
​
​​
​​
​​
​​
​​
​​
​​
​​
​
​
​
​​​
​​​
​​​
​​​
​​​
​​​
​
​​​
​
​
​
​
​
​
​​​​
​
​
​
‌‍‍‍‍‌‌‌‌‏‏‏‏‏‏‏‎‎‎‎‎‎‎‌‭‬️️️🈚️🈚️🈚️🈚️🈚️🈚️🈚️🈚️🈚️💲💲💲💲💲💲💲💲
...
Ok stop - actually we need correct formatting but due time let's continue quickly.
...


CostLog 表结构