1. 从告警短信到Token流:一个运维人的转型起点
2022年之前,我的日常基本被三样东西填满:Zabbix告警、Kubernetes的Pod重启记录、以及凌晨三点被叫起来处理磁盘写满的工单。那时候我对“大模型”这三个字的理解,大概停留在“能写诗的那个东西”的层面。直到有一次,业务方提了一个需求——把内部知识库做成问答接口,我才第一次认真去跑通一个本地推理服务。结果发现,部署一个模型比部署一套微服务还折腾:CUDA版本对不上、显存不够、推理速度慢得让人怀疑人生。但正是这次折腾,让我意识到一件事:运维的底层能力,和大模型工程化之间,其实只隔着一层窗户纸。
这篇文章不是成功学,也不是劝退帖。我想把这两年从系统运维转向大模型全栈开发的完整路径拆开来讲——包括我踩过的坑、做过的技术选型、以及那些“当时要是有人告诉我就好了”的关键节点。如果你也是运维出身,正在观望要不要往大模型方向走,或者你已经在大模型领域但工程化能力偏弱,那这篇内容应该能给你一些可以直接抄作业的东西。
先说结论:运维转大模型全栈,最大的优势不是算法,而是对“系统稳定性”和“资源边界”的直觉。你知道一个服务在什么情况下会崩,你知道日志该往哪里打,你知道怎么用最少的机器扛住最多的请求。这些东西,很多纯算法出身的人反而需要花很长时间补。而你需要补的,是Python工程化能力、模型推理的基本原理、以及一套能把模型变成产品的后端框架。
我选的技术栈很明确:Python + FastAPI + Ollama + LangChain。为什么是这套?后面会详细拆。先说说我最初犯的一个错误——试图用Flask去扛大模型的流式输出。结果就是,每次请求都像在等一个世纪,前端体验极差。后来换成FastAPI,原生支持异步和SSE,才算是把这条路走通了。
2. 为什么我最终选了FastAPI而不是Flask
2.1 从运维视角看框架选型:并发模型决定一切
运维出身的人对“并发”这个词是有肌肉记忆的。我们习惯看QPS、看连接数、看线程池大小。所以在选后端框架的时候,我第一反应就是去看它的并发模型。Flask默认是同步阻塞的,一个请求进来,线程就被占住,直到处理完才释放。这在传统CRUD场景下没问题,但大模型推理动辄几秒到几十秒,如果还用同步模型,你的服务并发能力基本等于线程数。
FastAPI基于Starlette,原生支持async/await。这意味着当一个请求在等模型推理结果的时候,事件循环可以去处理其他请求。实测下来,同样的机器配置,FastAPI的并发吞吐量大概是Flask的3到5倍。这个差距在大模型场景下会被进一步放大,因为推理时间越长,同步模型浪费的线程资源就越多。
还有一个细节:FastAPI自动生成OpenAPI文档。这个功能在前后端联调的时候太省事了。你定义好Pydantic模型,Swagger UI直接就能用,前端同事不用再追着你问接口字段是什么。对于运维转开发的人来说,这种“自带文档”的特性极大降低了沟通成本。
2.2 FastAPI项目目录结构:我踩过的三个坑
刚开始写FastAPI的时候,我按照Flask的习惯,把所有路由都塞在main.py里。结果不到一周,文件就膨胀到八百多行,改一个接口要翻半天。后来参考了一些开源项目的结构,才慢慢调整成现在这套:
project/ ├── app/ │ ├── main.py │ ├── api/ │ │ ├── __init__.py │ │ ├── routes/ │ │ │ ├── chat.py │ │ │ ├── model.py │ │ │ └── health.py │ │ └── dependencies.py │ ├── core/ │ │ ├── config.py │ │ └── logging.py │ ├── models/ │ │ └── schemas.py │ ├── services/ │ │ ├── llm_service.py │ │ └── vector_service.py │ └── utils/ │ └── helpers.py ├── tests/ ├── requirements.txt └── Dockerfile第一个坑是循环导入。routes里引用了services,services又引用了models,models反过来引用routes里的某个类型。解决方法是把Pydantic的Schema单独放在models/schemas.py里,所有层都只依赖它,不依赖具体路由。
第二个坑是配置管理混乱。开发环境和生产环境的Ollama地址不一样,我一开始硬编码在代码里,部署的时候改来改去。后来用pydantic-settings统一管理,通过环境变量注入,才算清爽了。
第三个坑是日志丢失。这个问题后面会单独讲,因为它是FastAPI + Uvicorn组合下最容易被忽略的坑之一。
2.3 异步接口的写法:别在async函数里干同步的事
FastAPI支持async def和普通def两种路由写法。我一开始图省事,全写成async def,然后在里面调用同步的requests库去请求Ollama。结果发现,事件循环被阻塞了,并发能力反而比Flask还差。
正确的做法是:如果调用的库是同步的,就用普通def写路由,FastAPI会自动把它放到线程池里执行。或者用httpx这种支持异步的HTTP客户端,配合async def使用。我现在的做法是统一用httpx.AsyncClient,所有外部调用都是异步的,这样事件循环才能真正发挥作用。
import httpx from fastapi import APIRouter router = APIRouter() @router.post("/chat") async def chat(prompt: str): async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post( "http://localhost:11434/api/generate", json={"model": "qwen2:7b", "prompt": prompt} ) return response.json()这段代码看起来简单,但背后有一个关键点:超时时间必须设置。大模型推理有时候会卡住,如果不设超时,请求会一直挂着,最终拖垮整个服务。我一般设60秒,超过就返回一个友好的错误提示。
3. 大模型本地部署:Ollama是我试过最省心的方案
3.1 为什么不用直接跑Transformers
刚开始的时候,我按照网上的教程,用transformers库直接加载模型。下载权重、配置CUDA、处理显存分配,一套流程走下来,半天就没了。而且每次换模型都要重新折腾一遍。更麻烦的是,transformers的推理服务需要自己写API封装,要考虑批处理、并发、显存回收,工程量不小。
后来试了Ollama,发现它把这些问题都解决了。Ollama的核心价值在于:它把模型下载、量化、推理服务、API暴露全部打包成了一个命令行工具。你只需要ollama run qwen2:7b,它就会自动下载模型并启动一个本地推理服务,默认监听11434端口。对于运维出身的人来说,这种“一条命令搞定”的体验太熟悉了,就像用docker run启动一个容器一样自然。
3.2 Ollama部署大模型的硬件门槛与量化选择
这里必须说清楚一个现实问题:大模型对显存的要求是有硬门槛的。我一开始用一台16GB显存的机器跑7B模型,勉强能跑,但上下文长度一开大就OOM。后来换成量化版本(Q4_K_M),显存占用降到6GB左右,才算是稳定了。
| 模型规模 | FP16显存需求 | Q4量化显存需求 | 推荐场景 |
|---|---|---|---|
| 7B | 约14GB | 约4-6GB | 个人开发、小规模问答 |
| 13B | 约26GB | 约8-10GB | 中等规模知识库 |
| 70B | 约140GB | 约35-40GB | 企业级私有化部署 |
量化本质上是用精度换空间。Q4量化会把模型权重从16位浮点数压缩到4位整数,显存占用降到原来的四分之一左右,但推理质量会有一定下降。我的经验是:7B模型用Q4量化,日常问答基本够用;如果要做代码生成或者复杂推理,建议至少上13B,并且用Q5或Q8量化。
还有一个容易被忽略的点:上下文长度直接影响显存占用。Ollama默认的上下文长度是2048,如果你调到8192,显存占用会显著增加。我一般根据实际需求设置,不要盲目开大。
3.3 FastAPI调用Ollama的完整封装
把Ollama的API封装成FastAPI接口,是我日常用得最多的功能。这里分享一个我实际在用的封装,包含了流式输出和错误处理:
import json import httpx from fastapi import APIRouter, HTTPException from fastapi.responses import StreamingResponse router = APIRouter() OLLAMA_BASE = "http://localhost:11434" async def stream_generator(prompt: str, model: str): async with httpx.AsyncClient(timeout=120.0) as client: async with client.stream( "POST", f"{OLLAMA_BASE}/api/generate", json={"model": model, "prompt": prompt, "stream": True} ) as response: async for line in response.aiter_lines(): if line: data = json.loads(line) yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n" @router.post("/chat/stream") async def chat_stream(prompt: str, model: str = "qwen2:7b"): return StreamingResponse( stream_generator(prompt, model), media_type="text/event-stream" )这段代码的关键在于StreamingResponse和text/event-stream。前端用EventSource接收,就能实现打字机效果。注意ensure_ascii=False,不然中文会被转义成Unicode编码,前端显示会出问题。
4. Uvicorn日志丢失:一个让我排查了整整两天的问题
4.1 问题现象:日志为什么凭空消失了
那是一个周五的下午,我在调试一个FastAPI接口,发现print出来的日志在终端里能看到,但用logging模块打的日志却不见了。更诡异的是,Uvicorn的访问日志正常输出,唯独我自己定义的logger没有任何输出。我检查了日志级别,是DEBUG;检查了handler,配置没问题;检查了logger名称,也对得上。但日志就是不出来。
这个问题困扰了我整整两天。期间我试过重启服务、换Python版本、甚至重装了Uvicorn。最后在一个技术社区的角落里,看到有人提到Uvicorn的日志配置会覆盖根logger的handler,才算是找到了方向。
4.2 根因定位:Uvicorn的日志配置机制
Uvicorn在启动的时候,会调用logging.config.dictConfig来配置自己的日志系统。这个配置会清空根logger上已有的handler,然后重新添加Uvicorn自己的handler。如果你的日志配置是在Uvicorn启动之前加载的,那它就会被覆盖掉。
具体来说,Uvicorn的默认日志配置里有一个disable_existing_loggers选项,默认值是False,但在某些版本或者特定配置下,它会变成True,导致你之前定义的所有logger都被禁用。
4.3 三种修复方案与我的最终选择
方案一:在Uvicorn启动之后再配置日志。把日志配置放到main.py的if __name__ == "__main__"块里,确保它在Uvicorn初始化之后执行。这个方案最简单,但如果你用uvicorn命令行启动,就不太好控制执行顺序。
方案二:自定义Uvicorn的日志配置。通过--log-config参数传入一个YAML文件,在里面同时定义Uvicorn和你自己的logger。这个方案最规范,但配置起来比较繁琐。
方案三:使用logging.getLogger("uvicorn")的父logger。把自定义logger挂到uvicorn.error下面,这样Uvicorn的配置就不会影响到它。这个方案最取巧,但不够优雅。
我最终选了方案二,因为项目要部署到生产环境,日志格式需要统一。下面是我用的log_config.yaml:
version: 1 disable_existing_loggers: false formatters: default: format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s" handlers: console: class: logging.StreamHandler formatter: default stream: ext://sys.stdout loggers: uvicorn: handlers: [console] level: INFO propagate: false app: handlers: [console] level: DEBUG propagate: false root: handlers: [console] level: INFO关键就是**disable_existing_loggers: false**这一行。加上它之后,我自定义的applogger就能正常输出了。
提示:如果你用的是
uvicorn.run()启动,可以直接传log_config参数;如果用命令行,就加--log-config log_config.yaml。
5. 从单次对话到AI Agent:LangChain和LangGraph的引入时机
5.1 什么时候该上LangChain
我一开始对LangChain是抗拒的。觉得它封装太厚,出了问题不好排查。但当我需要做“根据用户问题自动选择工具”的时候,手写路由逻辑变得越来越复杂。比如用户问“今天天气怎么样”,我需要判断这是天气查询,然后调用天气API;用户问“帮我总结这份文档”,我需要判断这是文档处理,然后调用文件读取和摘要工具。这些判断逻辑如果全用if-else写,代码会变得不可维护。
LangChain的价值在于它提供了一套标准化的工具调用和链式编排接口。你定义一个Tool,LangChain会自动处理参数解析和调用。你定义一个Chain,它会把多个步骤串起来。虽然底层还是那些东西,但代码组织上清晰很多。
5.2 LangGraph:当你的Agent需要“记住”上下文
LangChain的Chain是线性的,A步骤完了走B步骤。但实际场景中,Agent往往需要根据中间结果决定下一步走哪里。比如用户问了一个问题,Agent先检索知识库,如果检索结果相关,就直接回答;如果不相关,就调用搜索引擎。这种带分支和循环的逻辑,用LangGraph表达更自然。
LangGraph把Agent的状态定义成一个图,节点是处理步骤,边是跳转条件。你可以定义“如果检索置信度低于阈值,就跳到搜索节点”,这种逻辑用代码写就是几个if,但用图来表达,可读性和可维护性都好很多。
我现在的做法是:简单的单轮问答直接用FastAPI + Ollama,不引入LangChain;需要多工具协作或者多轮推理的场景,才上LangGraph。不要为了用框架而用框架,这是我从运维转开发最大的体会。
5.3 一个最小可用的Agent示例
下面这个例子展示了如何用LangGraph定义一个“先检索再回答”的Agent:
from langgraph.graph import StateGraph, END from typing import TypedDict class AgentState(TypedDict): question: str context: str answer: str def retrieve(state: AgentState): # 模拟检索 return {"context": "检索到的相关内容"} def generate(state: AgentState): # 模拟生成 return {"answer": f"基于{state['context']}的回答"} workflow = StateGraph(AgentState) workflow.add_node("retrieve", retrieve) workflow.add_node("generate", generate) workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "generate") workflow.add_edge("generate", END) app = workflow.compile()这个例子很简单,但结构清晰。你可以往里面加条件边、加循环、加人工审核节点。对于运维出身的人来说,这种“状态机”的思维方式其实很熟悉,和Prometheus的告警状态流转是一个道理。
6. 企业私有化部署:从“能跑”到“能用”的距离
6.1 私有化部署的真实需求是什么
很多人以为企业私有化部署就是把模型放到内网服务器上,然后开个API就完事了。实际远不止这些。我接触过的需求里,排在前面的几个是:数据不能出内网、响应速度要稳定、并发要能扛住、模型要能换、日志要能审计。这几个需求叠加起来,就不是一个ollama run能解决的了。
数据不出内网意味着你不能调用任何外部API,所有推理必须在本地完成。响应速度稳定意味着你不能让模型和别的服务抢资源,最好有独立的GPU节点。并发要能扛住意味着你需要做请求队列和限流。模型要能换意味着你的架构不能和某个特定模型绑定。日志要能审计意味着每一次问答都要有记录,包括谁问的、问了什么、模型答了什么。
6.2 我的部署架构与踩坑记录
我最终的部署架构是这样的:一台GPU服务器跑Ollama,一台CPU服务器跑FastAPI,中间用内网HTTP通信。FastAPI层做鉴权、限流、日志记录,Ollama层只负责推理。这样做的好处是,GPU资源可以独立扩展,FastAPI层可以水平扩容。
踩过的坑有几个。第一个是GPU服务器的显存碎片化。Ollama长时间运行后,显存会出现碎片,导致新请求无法分配显存。解决方法是定期重启Ollama服务,或者用ollama ps查看模型加载状态,手动卸载不用的模型。
第二个是FastAPI的请求队列。当并发请求超过Ollama的处理能力时,请求会堆积。我一开始没做队列,结果就是所有请求都超时。后来加了一个简单的信号量限流,超过阈值的请求直接返回“服务繁忙”,用户体验反而更好。
第三个是日志审计的存储。每次问答都记日志,一天下来就是几个GB。我一开始用文件存储,后来发现查询太慢,换成了SQLite。再后来数据量大了,又换成了PostgreSQL。这个演进过程说明,日志存储方案要跟着数据量走,不要一开始就上重型方案。
6.3 模型微调:什么时候需要,什么时候不需要
关于大模型微调,我的观点可能和很多人不一样:大部分企业场景不需要微调,用RAG就够了。微调的成本很高,需要标注数据、需要GPU资源、需要反复调参。而且微调后的模型会“遗忘”一些通用能力,在非特定任务上表现可能下降。
RAG(检索增强生成)的思路是:不改模型,而是在提问的时候,先从知识库里检索相关内容,然后把检索结果和问题一起送给模型。这样模型就能基于最新的、私有的知识来回答。RAG的优点是灵活,知识库更新了,回答就更新了,不需要重新训练。
当然,有些场景确实需要微调。比如你要让模型学会一种特定的输出格式,或者你要让模型掌握某个垂直领域的专业术语。这种情况下,微调是有效的。但我的建议是:先试RAG,RAG解决不了再考虑微调。
7. 给运维转大模型开发者的几条实在建议
7.1 Python工程化能力比算法知识更紧迫
我见过很多运维同事,Linux玩得很溜,但写Python的时候还是脚本思维——所有逻辑堆在一个文件里,没有类型注解,没有异常处理,没有单元测试。这种代码在个人项目里能跑,但一旦要协作或者上生产,就会出问题。
我的建议是,先把Python的工程化能力补上来。具体来说:学会用pydantic做数据校验,学会用pytest写测试,学会用black和ruff做代码格式化,学会用poetry或pip-tools管理依赖。这些东西不难,但能让你写出来的代码从“能跑”变成“能用”。
7.2 不要试图成为算法专家,但要懂推理原理
你不需要会推导Transformer的注意力公式,但你需要知道:上下文长度为什么影响显存、量化为什么影响质量、温度参数为什么影响输出的随机性。这些知识不需要很深,但能帮你在选型和调参的时候做出正确判断。
比如,当业务方说“回答太死板了”,你知道可以把温度调高一点。当业务方说“回答太发散了”,你知道可以把温度调低。当业务方说“响应太慢了”,你知道可以换更小的模型或者降低量化精度。这些判断不需要算法背景,但需要你对推理过程有基本的理解。
7.3 运维经验是你的差异化优势
最后说一点:不要觉得运维转开发是“降级”或者“从零开始”。你对系统稳定性的理解、对资源边界的敏感、对故障排查的直觉,这些都是纯开发人员需要花时间积累的。在大模型工程化这个领域,能把模型跑起来的人很多,但能让模型稳定跑下去的人很少。后者需要的,恰恰是运维的核心能力。
我现在的日常工作,大概30%在写业务代码,30%在调优推理性能,40%在处理各种工程化问题——部署、监控、日志、限流、容灾。后面这40%,几乎全是运维的老本行。所以如果你正在考虑转型,我的建议是:不要丢掉你的运维底子,把它当成你的核心竞争力,然后用Python和FastAPI把大模型的能力包装成产品。这条路我走通了,你也可以。