AI应用开发：用统一端点整合模型路由、工具调用与记忆管理

2026/8/29 14:18:07 拓冰建站浏览量

很多做 AI 应用的同学应该都有同感模型渠道越来越多工具函数越写越多会话历史散落在各个业务系统里每次新接一个模型或新加一个能力都要在调用方改一遍代码。最近我在做一个智能助手项目时正好把这个过程里反复踩的坑集中整理了一遍核心思路可以浓缩成一句话——在 AI 与它的连接、记忆、技能之间加一个统一端点。这篇文章会根据我实际落地的代码拆解这个统一端点的设计思路与实现方式。内容包括为什么需要统一端点、它和普通 API 网关有什么区别、如何用 FastAPI 搭建一个可运行的 AI 端点、如何处理工具调用和会话记忆以及常见报错和工程化建议。无论你是刚开始接触 AI 应用开发还是已经在做 agent 工程化都能从中找到可以直接复用的内容。1. 背景为什么 AI 应用需要“统一端点”1.1 一个端点解决什么问题你可以把统一端点理解为 AI 应用的“门面层”。客户端不需要关心后面接的是 DeepSeek、OpenAI、Ollama 还是 vLLM也不需要关心记忆存在 SQLite、Redis 还是向量数据库里更不需要关心工具体现在哪个函数、哪个外部服务上。它只需要向同一个地址发请求剩下的事情由端点来编排。具体来说常见的混乱场景是这样的模型调用直接散落在业务代码里换模型要改多处。对话历史由各业务系统自己存用户换个入口上下文就断了。工具函数散落在不同服务里agent 想调用却不知道从哪里发现、如何鉴权。同一个应用接入了多个模型但缺少统一的参数校验、错误处理和链路追踪。这些问题会让 AI 应用越做越重尤其是在智能客服、Copilot、自动化助手这类需要长期迭代的项目里影响会非常明显。统一端点的价值就在于把“模型路由、记忆存取、工具执行”收敛成一个边界清晰的层让上层业务只关心消息输入和输出。1.2 统一端点与普通 API 网关的区别有同学看到这里会问这和 Spring Cloud Gateway、Kong 这类 API 网关有什么区别API 网关解决的是“请求路由、限流、鉴权、灰度”等问题它工作在 HTTP 层负责把外部请求转发到不同的后端服务。而 AI 统一端点更像一个“AI 中间件”它在 API 网关的基础之上多做了一层语义编排能力普通 API 网关AI 统一端点请求转发支持支持模型路由一般不支持支持可按 provider、model 分发工具发现与执行不支持支持agent 可自动调用记忆持久化不支持支持保存会话上下文协议转换一般只做 HTTP 转发可做 OpenAI 兼容协议转换成本与配额控制基础限流可按模型维度精细化控制所以更准确地说AI 统一端点往往是架设在 API 网关下游的一层“智能编排层”。如果服务规模较小也可以直接暴露出来由业务侧统一调用。1.3 适用场景与目标读者这个模式适合以下场景需要接入多个大模型想统一切换和降级。正在开发 AI Agent需要让模型自动调用工具。需要保存多轮对话但不希望业务系统各自维护一份历史。想沉淀一套可复用的 AI 调用层供多个项目使用。本文的示例代码使用 Python 和 FastAPI 实现适合熟悉 Python 的后端开发者、AI 应用开发者以及做 agent 工程化落地的团队。2. 环境准备与核心概念拆解2.1 运行环境与依赖本文示例以以下环境为例版本可以根据你的项目实际情况调整重点看设计思路Python 3.10 及以上。FastAPI 0.115 及以上。Uvicorn 0.32 以上。OpenAI Python SDK 1.x用于调用 OpenAI 兼容接口。PyYAML 用于加载配置文件。SQLite 作为记忆存储。依赖文件如下文件路径requirements.txt。fastapi0.115.6 uvicorn[standard]0.32.1 openai1.58.1 pyyaml6.0.2 pydantic2.10.4 httpx0.28.1安装命令pip install -r requirements.txt如果你的网络环境比较特殊可以使用国内镜像源安装例如pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple2.2 核心概念Endpoint、模型路由、工具调用、记忆在开始写代码前先弄清楚四个关键概念。Endpoint端点指的是客户端访问服务的具体 URL。本文中的统一端点指/v1/chat这个入口所有对话请求都发到这里。模型路由根据配置或用户参数决定这次请求使用哪个模型供应商和哪个模型。实现上可以简单理解为“名称到客户端实例”的映射。工具调用也就是 Function Calling。模型在生成回复时如果需要查天气、算公式、查数据库会先返回一个“要调用哪个函数、参数是什么”的结构化结果而不是直接生成最终文本。AI 端点收到这个结果后执行对应函数再把函数结果回传给模型让模型结合真实结果生成答案。记忆对话历史。没有记忆的 AI 每次都是“陌生人”有了记忆才能形成连续的多轮对话。记忆可以分为短期会话记忆和长期用户记忆本文先实现短期会话记忆。2.3 请求链路设计一次完整的请求链路如下客户端应用 │ POST /v1/chat { session_id, messages } ▼ ┌─────────────────────────────────────────────┐ │ 统一 AI Endpoint │ │ ┌──────────┐ ┌──────────┐ ┌───────────┐ │ │ │ 记忆模块 │ │ 工具注册 │ │ 模型路由 │ │ │ │ SQLite │ │ 技能函数 │ │ 多 Provider│ │ │ └──────────┘ └──────────┘ └───────────┘ │ └─────────────────────────────────────────────┘ │ │ │ 历史上下文函数调用 OpenAI 兼容请求 ▼ DeepSeek / OpenAI / Ollama / vLLM图中的箭头代表了数据流向先读记忆再带上当前消息和工具描述发给模型如果模型触发工具调用则执行工具并把结果追加到上下文再次调用模型直到模型给出最终答案最后把整个会话摘要写回记忆。3. 架构设计与数据模型3.1 总体分层为了让代码可维护我按职责拆成几层models.py请求和响应的 Pydantic 模型。memory.py记忆存储层当前用 SQLite 实现。tools.py工具注册与执行层。clients.py模型客户端封装层。gateway.py统一编排层把上面几层串起来。main.pyFastAPI 入口。这样做的好处是每一层都可以单独替换。比如记忆从 SQLite 换到 Redis只需要改memory.py新增模型供应商只需要在clients.py和配置文件中添加。3.2 配置文件设计配置文件使用 YAML 格式文件路径config.yaml。llm: default_provider: deepseek providers: deepseek: base_url: https://api.deepseek.com/v1 api_key_env: DEEPSEEK_API_KEY model: deepseek-chat timeout: 60 openai: base_url: https://api.openai.com/v1 api_key_env: OPENAI_API_KEY model: gpt-4o-mini timeout: 60 ollama: base_url: http://127.0.0.1:11434/v1 api_key_env: OLLAMA_API_KEY model: qwen2.5:7b timeout: 120 memory: type: sqlite sqlite_path: ./data/memory.db max_history: 20 tools: enabled: true max_iterations: 5这里我把api_key_env配置成环境变量名而不是直接写密钥避免密钥泄露到代码仓库中。ollama这种本地模型不需要真正有效的 API Key所以读取不到时可以用占位符。3.3 请求与响应协议统一端点内部使用 OpenAI 兼容的消息格式因为目前 DeepSeek、Ollama、vLLM、LM Studio 等绝大多数模型服务都兼容这个协议。请求体的核心包括session_id会话 ID为空时自动创建新会话。messages消息数组角色包括 system、user、assistant、tool。provider可选指定模型供应商。model可选指定模型名。stream是否流式返回。响应体包含最终回复内容、本轮工具调用记录、token 用量以及实际使用的 provider 和 model。这样客户端可以拿到完整信息方便展示和计费。4. 完整实战用 FastAPI 搭建统一 AI 端点4.1 项目结构ai-endpoint/ ├── requirements.txt ├── config.yaml ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── gateway.py # 统一端点核心编排 │ ├── models.py # Pydantic 数据模型 │ ├── tools.py # 工具注册与执行 │ ├── memory.py # 记忆存储 │ └── clients.py # 模型客户端封装 └── data/ # SQLite 数据目录先创建目录mkdir -p ai-endpoint/app ai-endpoint/data4.2 实现数据模型文件路径app/models.py。from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: str Field(description消息角色system/user/assistant/tool) content: Optional[str] None tool_calls: Optional[List[Dict[str, Any]]] None tool_call_id: Optional[str] None class ChatRequest(BaseModel): messages: List[ChatMessage] session_id: Optional[str] None provider: Optional[str] None model: Optional[str] None temperature: Optional[float] 0.7 stream: Optional[bool] False class ChatResponse(BaseModel): session_id: str reply: str tool_calls: Optional[List[Dict[str, Any]]] None usage: Optional[Dict[str, Any]] None provider: str model: strChatMessage中的tool_calls是模型返回的“要调用哪个函数”的结构化信息tool_call_id用于把工具执行结果关联回对应调用。这里都先定义为可选字段因为普通文本消息不会携带这些内容。4.3 实现记忆存储文件路径app/memory.py。import json import sqlite3 from datetime import datetime from typing import Any, Dict, List class SQLiteMemory: 基于 SQLite 的短期会话记忆存储。 def __init__(self, db_path: str ./data/memory.db, max_history: int 20): self.db_path db_path self.max_history max_history self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, messages TEXT NOT NULL, updated_at TEXT NOT NULL ) ) def get_messages(self, session_id: str) - List[Dict[str, Any]]: with sqlite3.connect(self.db_path) as conn: row conn.execute( SELECT messages FROM sessions WHERE session_id ?, (session_id,), ).fetchone() if not row: return [] return json.loads(row[0]) def save_messages(self, session_id: str, messages: List[Dict[str, Any]]): with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO sessions (session_id, messages, updated_at) VALUES (?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET messages excluded.messages, updated_at excluded.updated_at , ( session_id, json.dumps(messages, ensure_asciiFalse), datetime.now().isoformat(), ), )这个实现有几个值得注意的点使用ON CONFLICT DO UPDATE同一个session_id直接覆盖历史避免重复插入。ensure_asciiFalse保证中文按原文存储不变成\uXXXX转义。max_history在写入时控制最多保留多少条消息防止上下文无限膨胀。4.4 实现工具注册与执行文件路径app/tools.py。工具层采用注册表模式函数通过装饰器注册模型需要调用时按名字查找。这里实现三个示例工具获取当前时间、安全计算数学表达式、获取网页标题。import ast import inspect import json import operator from datetime import datetime from typing import Any, Callable, Dict, List from zoneinfo import ZoneInfo class ToolRegistry: 工具/技能注册中心。 def __init__(self): self._tools: Dict[str, Callable] {} def register(self, tool: Callable): self._tools[tool.__name__] tool return tool def get(self, name: str) - Callable: if name not in self._tools: raise KeyError(ftool not found: {name}) return self._tools[name] def list_tools(self) - List[Dict[str, Any]]: 返回 OpenAI 兼容的 tools 描述。 tools [] for name, func in self._tools.items(): sig inspect.signature(func) parameters {} for pname, param in sig.parameters.items(): parameters[pname] { type: _type_to_json_schema(param.annotation), description: , } tools.append( { type: function, function: { name: name, description: inspect.getdoc(func) or , parameters: { type: object, properties: parameters, required: list(parameters.keys()), }, }, } ) return tools def call(self, name: str, arguments: Dict[str, Any]) - str: func self.get(name) try: result func(**arguments) return json.dumps(result, ensure_asciiFalse) except Exception as exc: return json.dumps({error: str(exc)}, ensure_asciiFalse) def _type_to_json_schema(annotation): if annotation is str: return string if annotation is int: return integer if annotation is float: return number if annotation is bool: return boolean return string registry ToolRegistry() registry.register def get_current_time(timezone: str Asia/Shanghai): 获取指定时区的当前时间。 return {time: datetime.now(ZoneInfo(timezone)).isoformat()} # 安全计算用 AST 解析禁止 eval 任意表达式 _OPERATORS { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod, ast.Pow: operator.pow, ast.USub: operator.neg, ast.UAdd: operator.pos, } def _safe_eval(node): if isinstance(node, ast.Expression): return _safe_eval(node.body) if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return node.value if isinstance(node, ast.BinOp) and type(node.op) in _OPERATORS: return _OPERATORS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right)) if isinstance(node, ast.UnaryOp) and type(node.op) in _OPERATORS: return _OPERATORS[type(node.op)](_safe_eval(node.operand)) raise ValueError(unsupported expression) registry.register def calculator(expression: str): 安全计算简单的数学表达式。 tree ast.parse(expression, modeeval) result _safe_eval(tree.body) return {result: result} registry.register def fetch_url_title(url: str): 获取网页标题作为连接外部系统的示例。仅允许 https。 import httpx from urllib.parse import urlparse parsed urlparse(url) if parsed.scheme ! https: raise ValueError(only https is allowed) # 生产环境还应增加 SSRF 防护禁止内网 IP、禁止云元数据地址 resp httpx.get(url, timeout5, follow_redirectsTrue) resp.raise_for_status() text resp.text if title in text: title text.split(title)[1].split(/title)[0] return {title: title[:100]} return {title: }这里有一个容易被忽视的安全点工具函数如果直接eval用户传入的表达式等于把执行权限交给了调用方。所以我用 Python 的ast模块解析表达式只允许加减乘除、取模、幂和正负号从根本上避免任意代码执行。4.5 实现模型客户端封装文件路径app/clients.py。import os from typing import Any, Dict, Optional from openai import OpenAI class LLMClient: OpenAI 兼容模型客户端支持多供应商路由。 def __init__(self, config: Dict[str, Any]): self.config config self._clients {} def _get_client(self, provider_name: str) - OpenAI: if provider_name in self._clients: return self._clients[provider_name] provider self.config[providers][provider_name] api_key os.getenv(provider[api_key_env], not-needed) client OpenAI( base_urlprovider[base_url], api_keyapi_key, timeoutprovider.get(timeout, 60), ) self._clients[provider_name] client return client def chat_completion( self, provider_name: str, model: str, messages: list, tools: Optional[list] None, temperature: float 0.7, ): provider self.config[providers][provider_name] client self._get_client(provider_name) kwargs { model: model or provider[model], messages: messages, temperature: temperature, } if tools: kwargs[tools] tools kwargs[tool_choice] auto return client.chat.completions.create(**kwargs)OpenAI客户端通过base_url切换不同供应商这是 OpenAI 兼容协议的优势。只要供应商暴露了兼容接口这段代码就完全不用改。本地使用 Ollama 时base_url填http://127.0.0.1:11434/v1即可。4.6 实现核心编排层文件路径app/gateway.py。编排层是整个统一端点的核心它负责从记忆模块加载历史消息。拼接当前请求消息。带上工具描述调用模型。如果模型返回工具调用就执行工具并回传结果。循环调用直到模型给出最终答案。把本轮对话压缩后写回记忆。import json import uuid from typing import Any, Dict, List from .clients import LLMClient from .memory import SQLiteMemory from .models import ChatRequest, ChatResponse from .tools import registry class AIEndpoint: 统一端点串联记忆、工具技能与模型路由。 def __init__(self, config: Dict[str, Any]): self.config config self.llm LLMClient(config[llm]) self.memory SQLiteMemory( db_pathconfig[memory].get(sqlite_path, ./data/memory.db), max_historyconfig[memory].get(max_history, 20), ) self.max_iterations config[tools].get(max_iterations, 5) def chat(self, req: ChatRequest) - ChatResponse: session_id req.session_id or str(uuid.uuid4()) # 1. 加载历史拼接新消息 history self.memory.get_messages(session_id) new_messages [item.model_dump() for item in req.messages] messages history new_messages provider req.provider or self.config[llm][default_provider] model req.model tools registry.list_tools() if self.config[tools].get(enabled, True) else None last_content usage {} last_response None # 2. 工具调用循环 for _ in range(self.max_iterations): clean_messages [_clean_message(m) for m in messages] last_response self.llm.chat_completion( provider_nameprovider, modelmodel, messagesclean_messages, toolstools, temperaturereq.temperature, ) choice last_response.choices[0] assistant_msg choice.message tool_calls assistant_msg.tool_calls or [] messages.append( { role: assistant, content: assistant_msg.content or , tool_calls: [tc.model_dump() for tc in tool_calls] if tool_calls else None, } ) # 3. 没有工具调用说明已生成最终答案 if not tool_calls: last_content assistant_msg.content or break # 4. 执行工具并回传结果 for tc in tool_calls: fn tc.function arguments fn.arguments if isinstance(arguments, str): arguments json.loads(arguments or {}) tool_result registry.call(fn.name, arguments) messages.append( { role: tool, tool_call_id: tc.id, content: tool_result, } ) # 5. 写入记忆 self._save_messages(session_id, messages) if last_response is not None and last_response.usage: usage last_response.usage.model_dump() return ChatResponse( session_idsession_id, replylast_content, tool_callsself._extract_tool_calls(messages), usageusage, providerprovider, modelmodel or self.config[llm][providers][provider].get(model, ), ) def _save_messages(self, session_id: str, messages: List[Dict[str, Any]]): 把对用户有价值的问答压缩后写入记忆。这里故意丢弃 tool_calls 和 tool 中间结果否则多轮工具调用会让历史消息快速膨胀。 compact [] for msg in messages: role msg.get(role) if role user: compact.append({role: user, content: msg.get(content, )}) elif role assistant and not msg.get(tool_calls): compact.append({role: assistant, content: msg.get(content, )}) compact compact[-self.memory.max_history:] if compact: self.memory.save_messages(session_id, compact) staticmethod def _extract_tool_calls(messages): calls [] for msg in messages: if msg.get(tool_calls): calls.extend(msg[tool_calls]) return calls def _clean_message(msg: Dict[str, Any]) - Dict[str, Any]: 去掉值为 None 的字段避免部分模型供应商拒绝请求。 return {k: v for k, v in msg.items() if v is not None}这里最需要注意的是_clean_message。OpenAI 官方 SDK 允许tool_calls为空数组但有些兼容供应商对tool_calls: null会直接返回 400。在发请求前统一清洗掉空字段能避开不少兼容性问题。4.7 实现 FastAPI 入口文件路径app/main.py。import yaml from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from .gateway import AIEndpoint from .models import ChatRequest, ChatResponse with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) app FastAPI(titleAI Unified Endpoint, version0.1.0) endpoint AIEndpoint(config) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) app.get(/health) def health(): return {status: ok} app.post(/v1/chat, response_modelChatResponse) def chat(req: ChatRequest): try: return endpoint.chat(req) except Exception as exc: # 生产环境建议把异常堆栈记录到日志返回给客户端的错误信息要脱敏 raise HTTPException(status_code500, detailstr(exc))4.8 运行与验证先配置环境变量。以 DeepSeek 为例export DEEPSEEK_API_KEY你的密钥启动服务cd ai-endpoint uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload看到以下日志说明启动成功INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Application startup complete.健康检查curl http://127.0.0.1:8000/health预期返回{status:ok}接下来发送一个会触发工具调用的请求curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d { session_id: demo-001, messages: [ {role: user, content: 现在几点了另外帮我计算 (1234)*5 的结果} ] }这个请求包含两个意图查时间、算数学。模型大概率会依次触发get_current_time和calculator两个工具最后返回汇总结果。返回的 JSON 结构类似{ session_id: demo-001, reply: 现在是北京时间 2025-01-18T10:30:0008:00(1234)*5 的结果是 230。, tool_calls: [...], usage: { prompt_tokens: 189, completion_tokens: 68, total_tokens: 257 }, provider: deepseek, model: deepseek-chat }4.9 测试本地模型如果你暂时没有云端模型的 API Key可以用 Ollama 跑本地模型测试只需要在配置里把default_provider改成ollama然后启动 Ollama 并拉取

建站实战干货

AI应用开发：用统一端点整合模型路由、工具调用与记忆管理