AI大模型实战-大模型API封装:自建大模型如何对外服务?

自建大模型怎么对外服务：用 FastAPI 把推理能力封装成 Web API

上一讲我们把企业知识库搭了起来，用的是向量库 + LangChain + ChatGLM3-6B 这套组合。回头看那个演示项目，其实接口封装那一步已经悄悄用上了——页面上敲的问题，是通过一层接口送到后端服务、再把结果带回来的。

但模型本身并不自带 Web API。想让它对外服务，就得自己在推理接口外面套一层 HTTP——企业要自建大模型，这一步躲不掉。

Python 这边承担这个角色的是 FastAPI，作用大致相当于 Java 世界里的 SpringBoot，注册接口非常省事。这一讲就用它把模型接口包起来。写个能跑的 Demo 其实不难，难的是真要放到工程里还得照顾一堆细节，所以下面我会把和 API 相关的关键点都摊开讲。这节内容配上前面的部署篇，你本地那套模型基本就能对外提供服务了。

一、先认识 FastAPI 和 Uvicorn

要提供 Web API，需要两样东西搭配着用。

• Uvicorn 负责当 Web 服务器，位置上对标 Tomcat，但轻得多。HTTP 请求它是异步处理的，底层建立在 httptools 和 uvloop 之上，性能很好，扛并发是强项。
• FastAPI 负责当 API 框架，角色对标 SpringBoot，同样轻很多——注意只是"角色像"，不是功能对等。

两者搭在一起：Uvicorn 把 FastAPI 应用跑起来，负责并发这一头；FastAPI 让接口写得又快又清爽，参数校验还是白送的。下面一步步来。

二、跑通第一个接口

装依赖

pip install fastapipip install uvicorn

最小可用的一个文件

新建 api.py，下面这些代码就足够定义出一个接口了：

import uvicornfrom fastapi import FastAPI # 创建 API 应用app = FastAPI() @app.get(”/”)async def root():    return {”message”: ”Hello World”} if __name__ == '__main__':    uvicorn.run(app, host='0.0.0.0', port=6006, log_level=”info”, workers=1)

启动它：

python api.py

三、请求参数怎么定义：Pydantic

真写业务的时候，一个接口往往要收好几个字段。Java 的做法是拿一个 Request 实体类把 HTTP 参数接住；Python 这边对应的是 Pydantic 模型——它是个做数据校验和配置管理的库，靠类型注解来验证数据，位置上相当于 Java 的 Validation。

先看一眼 Java 那边你熟悉的写法：

import javax.validation.constraints.Min;import javax.validation.constraints.NotNull;import javax.validation.constraints.Size; public class Product {     @NotNull    @Size(min = 2, max = 30)    private String name;     @NotNull    @Min(0)    private Float price;     // 构造器、getter 和 setter 省略}

换成 Python 就是这个样子：

import uvicornfrom fastapi import FastAPIfrom pydantic import BaseModel, Fieldfrom typing import List app = FastAPI() class Message(BaseModel):    role: str    content: str class ChatMessage(BaseModel):    history: List[Message]    prompt: str    max_tokens: int    temperature: float    top_p: float = Field(default=1.0) @app.post(”/v1/chat/completions”)async def create_chat_response(message: ChatMessage):    return {”message”: ”Hello World”} if __name__ == '__main__':    uvicorn.run(app, host='0.0.0.0', port=6006, log_level=”info”, workers=1)

这里出场的 BaseModel 有点像 Java 的 Object，但也只是"有点像"。Java 那边所有类都以 Object 为基类，因此默认就带着 equals()、hashCode()、toString() 这几个公共方法；BaseModel 的出发点则完全落在数据校验和管理上——一个类继承了它，就自动获得校验、序列化、反序列化这几项能力。

四、工程化：把代码分层

实际项目里，总不能把接口定义连同 Pydantic 类一股脑堆在最外层。Java 那套工程化经验是分层——service、controller、tool、model 各归各位；Python 想把代码管好，同样要分。一个典型结构长这样：

project_name/│├── app/ # 主应用目录│   ├── main.py # FastAPI 应用入口│   ├── controller/ # 接口层│   │   └── chat.py│   └── common/ # 通用组件│       └── errors.py # 错误处理与自定义异常│├── services/ # 服务层│   └── chat_service.py # 对话相关的业务逻辑│├── schemas/ # Pydantic 模型（请求 / 响应结构）│   └── chat_schema.py│├── database/ # 数据库连接与会话管理│   ├── session.py # 会话配置│   └── engine.py # 引擎配置│├── tools/ # 工具脚本│   └── data_migration.py # 数据迁移│├── tests/ # 测试│   ├── conftest.py # 测试配置与夹具│   ├── test_services/│   │   └── test_chat_service.py│   └── test_controller/│       └── test_chat_controller.py│├── requirements.txt # 依赖清单└── setup.py # 打包与分发配置

把散落的路由收拢到主应用上，靠的是 FastAPI 的 include_router。应用一旦做大，这个方法对组织和拆分代码帮助很明显。改造后的代码分成四个文件。

应用入口 main.py

import uvicornfrom fastapi import FastAPI from controller.chat_controller import chat_router app = FastAPI()app.include_router(chat_router, prefix=”/chat”, tags=[”chat”]) if __name__ == '__main__':    uvicorn.run(app, host='0.0.0.0', port=6006, log_level=”info”, workers=1)

接口层 chat_controller.py

from fastapi import APIRouter from service.chat_service import ChatServicefrom schema.chat_schema import ChatMessage chat_router = APIRouter()chat_service = ChatService() @chat_router.post(”/new/message/”)def post_message(message: ChatMessage):    return chat_service.post_message(message) @chat_router.get(”/get/messages/”)def get_messages():    return chat_service.get_messages()

服务层 chat_service.py

from schema.chat_schema import ChatMessage class ChatService:     def post_message(self, message: ChatMessage):        print(message.prompt)        return {”message”: ”post message”}     def get_messages(self):        return {”message”: ”get message”}

参数定义 chat_schema.py

from pydantic import BaseModel, Field class Message(BaseModel):    role: str    content: str class ChatMessage(BaseModel):    prompt: str    max_tokens: int    temperature: float = Field(default=1.0)    top_p: float = Field(default=1.0)

业务逻辑往 chat_service 里写就行，到这一步，写法上和 Java 已经没什么两样了。拿一段脚本自测一下：

import requests url = 'http://localhost:6006/chat/new/message/'data = {    'prompt': 'hello',    'max_tokens': 1000,}response = requests.post(url, json=data) # 用 json= 而不是 data=print(response.text) url2 = 'http://localhost:6006/chat/get/messages/'response = requests.get(url2)print(response.text)

输出：

{”message”:”post message”}{”message”:”get message”}

FastAPI 更细的用法可以去翻官方教程。骨架搭好，接下来把模型接进来。

五、把大模型接进来

各家模型的对话接口不太一样，下面按 ChatGLM3-6B 来写，封装动作放在 service 层：

from datetime import datetime import model_managerfrom schema.chat_schema import ChatMessage class ChatService:     def post_message(self, message: ChatMessage):        model = model_manager.ModelManager.get_model()        tokenizer = model_manager.ModelManager.get_tokenizer()         response, history = model.chat(            tokenizer,            message.prompt,            history=message.history,            max_length=message.max_tokens,            top_p=message.top_p,            temperature=message.temperature,        )         ts = datetime.now().strftime(”%Y-%m-%d %H:%M:%S”)        answer = {            ”response”: response,            ”history”: history,            ”status”: 200,            ”time”: ts,        }         print(f'[{ts}] prompt: ”{message.prompt}”, response: ”{repr(response)}”')        return answer     def get_messages(self):        return {”message”: ”get message”}

模型别每次请求都加载一遍，用一个 ModelManager 做懒加载：

from transformers import AutoTokenizer, AutoModelForCausalLM class ModelManager:    _model = None    _tokenizer = None     @classmethod    def get_model(cls):        if cls._model is None:            cls._model = AutoModelForCausalLM.from_pretrained(                ”chatglm3-6b”, trust_remote_code=True).half().cuda().eval()        return cls._model     @classmethod    def get_tokenizer(cls):        if cls._tokenizer is None:            cls._tokenizer = AutoTokenizer.from_pretrained(                ”chatglm3-6b”, trust_remote_code=True)        return cls._tokenizer

model.chat() 是 6B 对外暴露的对话入口，包一层就有了基本的对话接口。不过它是把整段回答一次性吐出来的——而我们平时用 ChatGPT、文心一言，看到的是一个字一个字往外蹦。那种叫流式输出。

六、流式输出

流式走的是另一个接口 model.stream_chat，具体怎么吐有几种做法。一种是每次只给新增的那一小段：

我是中国人

另一种是每次都把已经生成的内容整个重发一遍：

我我是我是中我是中国我是中国人

也有一次吐两个字的。到底怎么切，生产上按产品的交互设计来定就行。下面这段用一个 stream 变量控制走不走流式：

if stream:    async for token in callback.aiter():  # 用 server-sent events 把 token 逐个推给前端        yield json.dumps(            {”text”: token, ”message_id”: message_id},            ensure_ascii=False)else:    answer = ””    async for token in callback.aiter():        answer += token    yield json.dumps(        {”text”: answer, ”message_id”: message_id},        ensure_ascii=False) await task

输入一句"你好"，把 stream 打开，接口就是这样一段段往外推的：

data: {”text”: ”你”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”好”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”👋”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”！”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”我是”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”人工智能”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”助手”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ” Chat”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”GL”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”M”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”3”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”-”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”6”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}data: {”text”: ”B”, ”message_id”: ”80b2af55c5b7440eaca6b9d510677a75”}……（后面依次是”，很高兴见到你，欢迎问我任何问题。”，格式相同）

把 stream 关掉，就退回成一条完整响应：

data: {”text”: ”你好！我是人工智能助手，很高兴为您服务。请问有什么问题我可以帮您解答吗？”, ”message_id”: ”741a630ac3d64fd5b1832cc0bae6bb68”}

到这里，模型的 API 就算封装完了，接下来看怎么调。

七、上层怎么调用

工程上通常的分工是：AI 相关的逻辑连同模型 API 封装都留在 Python 应用里，面向用户的那一层交给别的语言，Java、C#、Go 都有人用。这里拿 Java 举个例子。非流式就是一次普通的 HTTP 请求，没什么可说的，重点看流式怎么接——两步，两步都是流式的。

第一步：Java 调 Python 接口

主力是 okhttp3，动作拆成三段：拼参数、发起流式请求、挂上事件监听。

@ApiOperation(value = ”流式发送对话消息”)@PostMapping(value = ”sendMessage”)public void sendMessage(@RequestBody ChatRequest request, HttpServletResponse response) {    try {        JSONObject body = new JSONObject();        body.put(”model”, request.getModel());        body.put(”stream”, true);         JSONArray messages = new JSONArray();        JSONObject query = new JSONObject();        query.put(”role”, ”user”);        query.put(”content”, request.getQuery());        messages.add(query);        body.put(”messages”, messages);         EsListener eventSourceListener = new EsListener(request, response);        RequestBody formBody = RequestBody.create(body, MediaType.parse(”application/json”));         Request.Builder requestBuilder = new Request.Builder();        Request streamRequest = requestBuilder.url(URL).post(formBody).build();         EventSource.Factory factory = EventSources.createFactory(OkHttpUtil.getInstance());        factory.newEventSource(streamRequest, eventSourceListener);         eventSourceListener.getCountDownLatch().await();    } catch (Exception e) {        log.error(”流式调用异常”, e);    }}

EsListener 继承自 EventSourceListener。请求进行的过程中，它的 onEvent 会被反复触发，我们在里面把数据写回前端：

@Overridepublic void onEvent(EventSource eventSource, String id, String type, String data) {    try {        output.append(data);         if (”finish”.equals(type)) {            // 收尾逻辑        }        if (”error”.equals(type)) {            // 异常逻辑        }         // 这里只做最基本的透传，实际业务逻辑自行扩展        if (response != null) {            response.getWriter().write(data);            response.getWriter().flush();        }    } catch (Exception e) {        log.error(”事件处理异常”, e);    }}

第二步：前端调 Java 接口

浏览器原生的 EventSource 就够用了：

let eventData = '';  const eventSource = new EventSource('http://localhost:8888/sendMessage');   eventSource.onmessage = function (event) {    // 把陆续到达的片段拼起来    eventData += event.data;  };

走完这两步，从封装到调用的整条链路就通了，建议自己串起来跑一遍看看效果。真做工程还会碰到别的问题——Java 调 Python 这一段的鉴权、跨域，以及限流（模型吞吐量摆在那儿，扛不住就得拦），这些放到后面的课再说。

八、照着敲会踩到的几个坑

课程里的示例是讲解用的片段，直接拷进项目跑不一定顺，下面几处值得注意（本文的代码已经按这几点调过）。

• ModelManager 的懒加载：如果写成 _model = AutoModelForCausalLM.from_pretrained(...)，赋的是函数内的局部变量，类属性 cls._model 始终是 None——第一次调用返回的是局部变量，第二次因为跳过了 if、局部变量没被赋值，直接 UnboundLocalError。必须写成 cls._model = ...，get_tokenizer 同理。
• 时间格式化：from datetime import datetime 之后，正确写法是 datetime.now()。再写 datetime.datetime.now() 会 AttributeError。
• 字段名要对得上：调 model.chat() 时传的是 history，别写成 histroy；另外分层之后的 ChatMessage 里已经没有 history 字段了，接模型之前记得把它加回去，否则取不到属性。
• 自测脚本发请求：requests.post(url, data=json.dumps(...)) 发出去的 Content-Type 是表单，FastAPI 会按 422 拒掉。要么改用 json=data，要么自己把 header 设成 application/json。

九、小结

这一节讲的是自建大模型对外服务绕不开的一环，整体不算难。真要说门槛，也就是得用 Python：FastAPI 里异步操作很多，和 Java 的处理习惯有些出入，写的时候留意一下。

学到这里，企业内部构建大模型这条线基本走完了，你自己搭的模型已经可以对外服务。不过要真上生产，降级预案一定得先备好——不确定的地方太多了：吞吐量（TPS）估得准不准、模型会不会冒出意料之外的输出，一旦出状况就得随时降级。

思考题：前面提过一句，模型这一层的接口用 Python 封装，真正对外服务时外头还会再叠一层 Java 服务。这么分工的理由是什么？欢迎在留言区聊聊你的想法。

十、留言区里的几个问题

挑几条有代表性的，按我的理解重新整理了一遍，答案要点来自作者的回复。

Q1（思考题）：为什么 Python 外面还要套一层 Java？

读者的答案是从各语言的长处来拆的：核心 API 交给 Python，是因为机器学习这块的生态几乎都长在它上面，TensorFlow、PyTorch 都在这儿，主流大模型也基本都配了 Python SDK；外头那层 Java，则是冲着它在 Web 服务端积累多年的家底去的——路由、限流、熔断、降级、可观测、服务鉴权，这些能力都能很快搭起来。还有读者从分层架构的角度补充：各司其职，Python 专心管模型，对上层保持透明，Java 按需调用、不必侵入模型实现；同时 Python 这层对外只暴露抽象接口，调用方就不限于 Java 了，别的语言照着接口实现一样能用。作者对这两种说法都点了头。

Q2：文中的代码能直接跑吗？Python 基础一般，感觉结构和实际案例对不上，controller 里 import service 就引不到。

作者的建议是自己稍微调一调，正好练练 Python——毕竟搞 AI 主要靠它，这关不过后面很难深入。具体要改哪些地方，可以对照上面第八节那几条。

Q3：模型被问到"你是谁"这类身份问题时，怎么让它按我指定的口径回答？

作者的答复很干脆：微调。这类身份设定属于要把答案"写进"模型里的需求，靠外部封装解决不了。

Q4：大模型的输出降级有哪些做法？被诱导说出不当内容怎么快速堵上？机器负载打满了又该怎么办？

作者给了三层：一是上线前把模型测充分；二是在请求前后各加一道内容检测，命中问题就直接拦掉；三是负载到顶就限流，这是通行做法，ChatGPT 也一样。

Q5：国内访问 HuggingFace 不方便，有什么替代？

作者提到可以用魔搭（ModelScope）作为替代来源。上一讲下载 Embedding 模型时我们走的就是这条路，国内速度确实更稳。

Q6：示例用的是哪个推理框架？能不能换成 vLLM？

示例基于 PyTorch，作者建议其他框架自己试。另外也有读者问课程代码有没有仓库，作者的回复是没有放仓库，得自己照着敲。