文章目录【64.PythonAI】生产级AI API服务LangChain FastAPI 从开发到Docker部署导入语1 ~ 最小可用API1.1 从最简单的开始1.2 这个能上线吗2 ~ 请求队列防止并发打爆 LLM2.1 问题分析2.2 用 asyncio.Queue 实现排队3 ~ Redis 缓存相同问题不重复花钱3.1 LLM 缓存的价值3.2 实现语义缓存4 ~ 速率限制令牌桶算法4.1 为什么需要速率限制4.2 令牌桶实现5 ~ 整体架构5.1 请求处理全流程5.2 完整的 FastAPI 代码6 ~ Prometheus 监控 Docker 部署6.1 监控指标6.2 Dockerfile6.3 docker-compose.yml7 ~ 生产 Checklist思考 总结结尾【64.PythonAI】生产级AI API服务LangChain FastAPI 从开发到Docker部署文章简介本文手把手教你用LangChain FastAPI搭建一个可以直接上线的AI API服务。文章覆盖从单机脚本到生产级服务的完整演进过程请求队列防止并发打爆LLM、Redis缓存LLM响应节省Token成本、基于令牌桶的速率限制保护后端、Prometheus指标暴露实现Grafana可视化监控以及Docker多阶段构建和docker-compose一键部署。配有Mermaid架构图展示请求处理流程适合已经能用LangChain写单次调用脚本、想要把AI能力包装成稳定可用API的开发者。 个人主页源码骑士❄专栏传送门《Android开发基础》《python基础课程》⭐️热衷从源码视角拆解技术底层原理将复杂架构讲得通俗易懂 源码骑士的简介5年Android Framework系统开发经验曾主导多项系统级性能优化专项技术栈覆盖Android系统全链路Binder/Handler/AMS/WMS/启动流程及Java后端全家桶Spring MyBatis Redis Oracle累计产出原创技术文章100篇文章以流程图为特色被读者评价为看一篇胜过啃一周源码导入语你写了一个很牛的LangChain Chain——能检索知识库、能调用工具、能多轮对话。但问题是它在你的笔记本里只有你能用。要把这个Chain变成别人也能调用的API不是把代码往Flask里一扔就完事了。LLM API有自己的速率限制、每次调用要花钱、高并发下会把你的Token配额打爆。生产级的AI API服务需要队列、缓存、限流、监控——少一个都不靠谱。这篇文章就把这一整套东西串起来。一个完整的、可Docker一键部署的LangChain FastAPI AI API服务。1 ~ 最小可用API1.1 从最简单的开始fromfastapiimportFastAPIfrompydanticimportBaseModelfromlangchain_openaiimportChatOpenAIfromlangchain.chainsimportRetrievalQA appFastAPI(titleAI API Service)llmChatOpenAI(modelgpt-4,temperature0)qa_chainRetrievalQA.from_chain_type(llmllm,retrieverretriever)classQueryRequest(BaseModel):question:strclassQueryResponse(BaseModel):answer:strmodel:strapp.post(/qa,response_modelQueryResponse)asyncdefqa(request:QueryRequest):resultqa_chain.invoke({query:request.question})returnQueryResponse(answerresult[result],modelgpt-4)1.2 这个能上线吗缺失项风险没有并发控制100个人同时请求 → Token额度秒空没有速率限制恶意刷接口无感知没有缓存重复问题每次都花Token没有监控出问题全靠用户反馈没有错误重试LLM API偶尔抽风 → 直接返回5002 ~ 请求队列防止并发打爆 LLM2.1 问题分析LLM API尤其是GPT-4对并发有硬性限制——RPM每分钟请求数和TPM每分钟Token数。直接用FastAPI的async处理瞬时涌入100个请求会全部丢给LLM然后收到一堆429。2.2 用 asyncio.Queue 实现排队importasyncioclassLLMRequestQueue:LLM请求队列控制并发调用数量def__init__(self,max_concurrent:int5):self.semaphoreasyncio.Semaphore(max_concurrent)asyncdefexecute(self,chain,query:str):asyncwithself.semaphore:returnawaitchain.ainvoke({query:query})queueLLMRequestQueue(max_concurrent5)app.post(/qa)asyncdefqa(request:QueryRequest):resultawaitqueue.execute(qa_chain,request.question)returnQueryResponse(answerresult[result],modelgpt-4)Semaphore(5)保证同一时间最多只有5个请求在处理LLM调用超出数量的请求会排队等待。3 ~ Redis 缓存相同问题不重复花钱3.1 LLM 缓存的价值很多场景下用户问的问题是高度重复的比如客服系统的Top 20高频问题。缓存这些问题的LLM回答能省下大量Token费用。3.2 实现语义缓存importredisimporthashlibimportjson redis_clientredis.Redis(hostlocalhost,port6379,db0)CACHE_TTL3600# 缓存1小时defget_cache_key(question:str)-str:生成缓存Key问题文本的SHA256摘要returnfllm:cache:{hashlib.sha256(question.encode()).hexdigest()}asyncdefcached_qa(question:str)-dict:cache_keyget_cache_key(question)# 先查缓存cachedredis_client.get(cache_key)ifcached:print(f✅ 命中缓存{question[:30]}...)returnjson.loads(cached)# 缓存未命中调用LLMresultawaitqueue.execute(qa_chain,question)# 写入缓存redis_client.setex(cache_key,CACHE_TTL,json.dumps(result))returnresult4 ~ 速率限制令牌桶算法4.1 为什么需要速率限制你已经控制了LLM的并发但用户请求的速率本身也需要限制——防止单个IP刷接口把你的服务打挂。4.2 令牌桶实现importtimeclassTokenBucket:令牌桶速率限制器def__init__(self,rate:int,capacity:int):self.raterate# 每秒生成令牌数self.capacitycapacity# 桶最大容量self.tokenscapacity# 当前令牌数self.last_refilltime.monotonic()defconsume(self,tokens:int1)-bool:消费令牌返回是否成功nowtime.monotonic()elapsednow-self.last_refill# 补充令牌self.tokensmin(self.capacity,self.tokenselapsed*self.rate)self.last_refillnowifself.tokenstokens:self.tokens-tokensreturnTruereturnFalse# 全局桶每秒10个请求最大突发20个bucketTokenBucket(rate10,capacity20)fromfastapiimportHTTPExceptionapp.post(/qa)asyncdefqa(request:QueryRequest):ifnotbucket.consume():raiseHTTPException(status_code429,detail请求过于频繁请稍后再试)# ...继续处理5 ~ 整体架构5.1 请求处理全流程通过限流命中未命中客户端请求令牌桶限流Redis缓存检查返回429返回缓存结果请求队列Semaphore控制并发LangChain处理LLM调用写入Redis缓存返回结果5.2 完整的 FastAPI 代码fromfastapiimportFastAPI,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefromcontextlibimportasynccontextmanagerasynccontextmanagerasyncdeflifespan(app:FastAPI):应用生命周期启动时初始化资源关闭时清理# 启动时初始化print(✅ AI API Service Started)yield# 关闭时清理redis_client.close()print( AI API Service Stopped)appFastAPI(titleAI API Service,lifespanlifespan)app.add_middleware(CORSMiddleware,allow_origins[*])app.post(/qa)asyncdefqa(request:QueryRequest):# 1. 速率限制ifnotbucket.consume():raiseHTTPException(status_code429,detailToo Many Requests)# 2. 缓存 LLM调用resultawaitcached_qa(request.question)returnQueryResponse(answerresult[result],modelgpt-4)app.get(/health)asyncdefhealth():return{status:ok,queue_size:5-queue.semaphore._value}6 ~ Prometheus 监控 Docker 部署6.1 监控指标fromprometheus_fastapi_instrumentatorimportInstrumentator instrumentatorInstrumentator()instrumentator.instrument(app).expose(app)# 自定义指标fromprometheus_clientimportCounter,Histogram llm_requestsCounter(llm_requests_total,Total LLM requests)llm_latencyHistogram(llm_latency_seconds,LLM call latency)6.2 DockerfileFROM python:3.11-slim AS builder WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt FROM python:3.11-slim WORKDIR /app COPY --frombuilder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]6.3 docker-compose.ymlversion:3.9services:api:build:.ports:-8000:8000environment:-OPENAI_API_KEY${OPENAI_API_KEY}-REDIS_URLredis://redis:6379depends_on:-redisredis:image:redis:7-alpineports:-6379:6379prometheus:image:prom/prometheusports:-9090:9090volumes:-./prometheus.yml:/etc/prometheus/prometheus.ymlgrafana:image:grafana/grafanaports:-3000:30007 ~ 生产 Checklist上线前逐项确认☐ 限流生效了吗压测100QPS 确认429正确返回 ☐ 缓存命中率是多少Redis 监控 key 命中率 ☐ LLM 并发上限是多少Semaphore 值是否匹配 API 限制 ☐ 健康检查接口可用吗/health 返回200 ☐ 日志级别对了吗生产用 INFO不要用 DEBUG ☐ Prometheus 指标正常暴露了吗/metrics 可访问 ☐ Docker 镜像大小合理吗500MB 为佳 ☐ 敏感信息API Key走环境变量了吗思考 总结不要把 Jupyter 里的代码直接拿去上线一个能跑通的 API 和能上线的 API中间隔着队列、缓存、限流、监控四座大山。优先级排序限流 缓存 队列 监控。先保证不被刷爆和Token浪费再优化吞吐和可观测性。Semaphore 是最简单的并发控制方式不需要引入 Celery 这种重型任务队列。99% 的 LLM API 场景下asyncio.Semaphore就够了。Redis 缓存 ROI 最高GPT-4 每次调用几毛钱高频问题每天可能被问几百次。缓存一开Token 开销直接砍半。Docker Compose 让你一键启动全家桶API Redis Prometheus Grafana一条docker-compose up全搞定。生产级的 AI API 服务代码只占 30%基础设施缓存/限流/监控/部署占 70%。结尾各位小伙伴本文的内容到这里就全部结束了源码骑士在这里再次感谢您的阅读源码骑士 — Android Framework 全栈开发关注跟博主一起从源码视角深耕底层原理见证每一次成长❤️点赞让优质内容被更多人看见让知识传递更有力量⭐收藏把核心知识点存好在需要时随时查、随时用评论分享你的经验或疑问评论区一起交流避坑一键四连不要忘记给博主一键四连哦️寄语技术之路难免有困惑但同行的人会让前进更有方向结语从脚本到服务差的不是代码量是那层生产意识。本文的代码模板可以直接用到你的项目中——把业务 Chain 替换进去队列、缓存、限流、监控一个不少。不要忘记给博主一键四连哦
64-LangChain-FastAPI-AI-API服务-生产级部署-并发缓存监控
文章目录【64.PythonAI】生产级AI API服务LangChain FastAPI 从开发到Docker部署导入语1 ~ 最小可用API1.1 从最简单的开始1.2 这个能上线吗2 ~ 请求队列防止并发打爆 LLM2.1 问题分析2.2 用 asyncio.Queue 实现排队3 ~ Redis 缓存相同问题不重复花钱3.1 LLM 缓存的价值3.2 实现语义缓存4 ~ 速率限制令牌桶算法4.1 为什么需要速率限制4.2 令牌桶实现5 ~ 整体架构5.1 请求处理全流程5.2 完整的 FastAPI 代码6 ~ Prometheus 监控 Docker 部署6.1 监控指标6.2 Dockerfile6.3 docker-compose.yml7 ~ 生产 Checklist思考 总结结尾【64.PythonAI】生产级AI API服务LangChain FastAPI 从开发到Docker部署文章简介本文手把手教你用LangChain FastAPI搭建一个可以直接上线的AI API服务。文章覆盖从单机脚本到生产级服务的完整演进过程请求队列防止并发打爆LLM、Redis缓存LLM响应节省Token成本、基于令牌桶的速率限制保护后端、Prometheus指标暴露实现Grafana可视化监控以及Docker多阶段构建和docker-compose一键部署。配有Mermaid架构图展示请求处理流程适合已经能用LangChain写单次调用脚本、想要把AI能力包装成稳定可用API的开发者。 个人主页源码骑士❄专栏传送门《Android开发基础》《python基础课程》⭐️热衷从源码视角拆解技术底层原理将复杂架构讲得通俗易懂 源码骑士的简介5年Android Framework系统开发经验曾主导多项系统级性能优化专项技术栈覆盖Android系统全链路Binder/Handler/AMS/WMS/启动流程及Java后端全家桶Spring MyBatis Redis Oracle累计产出原创技术文章100篇文章以流程图为特色被读者评价为看一篇胜过啃一周源码导入语你写了一个很牛的LangChain Chain——能检索知识库、能调用工具、能多轮对话。但问题是它在你的笔记本里只有你能用。要把这个Chain变成别人也能调用的API不是把代码往Flask里一扔就完事了。LLM API有自己的速率限制、每次调用要花钱、高并发下会把你的Token配额打爆。生产级的AI API服务需要队列、缓存、限流、监控——少一个都不靠谱。这篇文章就把这一整套东西串起来。一个完整的、可Docker一键部署的LangChain FastAPI AI API服务。1 ~ 最小可用API1.1 从最简单的开始fromfastapiimportFastAPIfrompydanticimportBaseModelfromlangchain_openaiimportChatOpenAIfromlangchain.chainsimportRetrievalQA appFastAPI(titleAI API Service)llmChatOpenAI(modelgpt-4,temperature0)qa_chainRetrievalQA.from_chain_type(llmllm,retrieverretriever)classQueryRequest(BaseModel):question:strclassQueryResponse(BaseModel):answer:strmodel:strapp.post(/qa,response_modelQueryResponse)asyncdefqa(request:QueryRequest):resultqa_chain.invoke({query:request.question})returnQueryResponse(answerresult[result],modelgpt-4)1.2 这个能上线吗缺失项风险没有并发控制100个人同时请求 → Token额度秒空没有速率限制恶意刷接口无感知没有缓存重复问题每次都花Token没有监控出问题全靠用户反馈没有错误重试LLM API偶尔抽风 → 直接返回5002 ~ 请求队列防止并发打爆 LLM2.1 问题分析LLM API尤其是GPT-4对并发有硬性限制——RPM每分钟请求数和TPM每分钟Token数。直接用FastAPI的async处理瞬时涌入100个请求会全部丢给LLM然后收到一堆429。2.2 用 asyncio.Queue 实现排队importasyncioclassLLMRequestQueue:LLM请求队列控制并发调用数量def__init__(self,max_concurrent:int5):self.semaphoreasyncio.Semaphore(max_concurrent)asyncdefexecute(self,chain,query:str):asyncwithself.semaphore:returnawaitchain.ainvoke({query:query})queueLLMRequestQueue(max_concurrent5)app.post(/qa)asyncdefqa(request:QueryRequest):resultawaitqueue.execute(qa_chain,request.question)returnQueryResponse(answerresult[result],modelgpt-4)Semaphore(5)保证同一时间最多只有5个请求在处理LLM调用超出数量的请求会排队等待。3 ~ Redis 缓存相同问题不重复花钱3.1 LLM 缓存的价值很多场景下用户问的问题是高度重复的比如客服系统的Top 20高频问题。缓存这些问题的LLM回答能省下大量Token费用。3.2 实现语义缓存importredisimporthashlibimportjson redis_clientredis.Redis(hostlocalhost,port6379,db0)CACHE_TTL3600# 缓存1小时defget_cache_key(question:str)-str:生成缓存Key问题文本的SHA256摘要returnfllm:cache:{hashlib.sha256(question.encode()).hexdigest()}asyncdefcached_qa(question:str)-dict:cache_keyget_cache_key(question)# 先查缓存cachedredis_client.get(cache_key)ifcached:print(f✅ 命中缓存{question[:30]}...)returnjson.loads(cached)# 缓存未命中调用LLMresultawaitqueue.execute(qa_chain,question)# 写入缓存redis_client.setex(cache_key,CACHE_TTL,json.dumps(result))returnresult4 ~ 速率限制令牌桶算法4.1 为什么需要速率限制你已经控制了LLM的并发但用户请求的速率本身也需要限制——防止单个IP刷接口把你的服务打挂。4.2 令牌桶实现importtimeclassTokenBucket:令牌桶速率限制器def__init__(self,rate:int,capacity:int):self.raterate# 每秒生成令牌数self.capacitycapacity# 桶最大容量self.tokenscapacity# 当前令牌数self.last_refilltime.monotonic()defconsume(self,tokens:int1)-bool:消费令牌返回是否成功nowtime.monotonic()elapsednow-self.last_refill# 补充令牌self.tokensmin(self.capacity,self.tokenselapsed*self.rate)self.last_refillnowifself.tokenstokens:self.tokens-tokensreturnTruereturnFalse# 全局桶每秒10个请求最大突发20个bucketTokenBucket(rate10,capacity20)fromfastapiimportHTTPExceptionapp.post(/qa)asyncdefqa(request:QueryRequest):ifnotbucket.consume():raiseHTTPException(status_code429,detail请求过于频繁请稍后再试)# ...继续处理5 ~ 整体架构5.1 请求处理全流程通过限流命中未命中客户端请求令牌桶限流Redis缓存检查返回429返回缓存结果请求队列Semaphore控制并发LangChain处理LLM调用写入Redis缓存返回结果5.2 完整的 FastAPI 代码fromfastapiimportFastAPI,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefromcontextlibimportasynccontextmanagerasynccontextmanagerasyncdeflifespan(app:FastAPI):应用生命周期启动时初始化资源关闭时清理# 启动时初始化print(✅ AI API Service Started)yield# 关闭时清理redis_client.close()print( AI API Service Stopped)appFastAPI(titleAI API Service,lifespanlifespan)app.add_middleware(CORSMiddleware,allow_origins[*])app.post(/qa)asyncdefqa(request:QueryRequest):# 1. 速率限制ifnotbucket.consume():raiseHTTPException(status_code429,detailToo Many Requests)# 2. 缓存 LLM调用resultawaitcached_qa(request.question)returnQueryResponse(answerresult[result],modelgpt-4)app.get(/health)asyncdefhealth():return{status:ok,queue_size:5-queue.semaphore._value}6 ~ Prometheus 监控 Docker 部署6.1 监控指标fromprometheus_fastapi_instrumentatorimportInstrumentator instrumentatorInstrumentator()instrumentator.instrument(app).expose(app)# 自定义指标fromprometheus_clientimportCounter,Histogram llm_requestsCounter(llm_requests_total,Total LLM requests)llm_latencyHistogram(llm_latency_seconds,LLM call latency)6.2 DockerfileFROM python:3.11-slim AS builder WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt FROM python:3.11-slim WORKDIR /app COPY --frombuilder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]6.3 docker-compose.ymlversion:3.9services:api:build:.ports:-8000:8000environment:-OPENAI_API_KEY${OPENAI_API_KEY}-REDIS_URLredis://redis:6379depends_on:-redisredis:image:redis:7-alpineports:-6379:6379prometheus:image:prom/prometheusports:-9090:9090volumes:-./prometheus.yml:/etc/prometheus/prometheus.ymlgrafana:image:grafana/grafanaports:-3000:30007 ~ 生产 Checklist上线前逐项确认☐ 限流生效了吗压测100QPS 确认429正确返回 ☐ 缓存命中率是多少Redis 监控 key 命中率 ☐ LLM 并发上限是多少Semaphore 值是否匹配 API 限制 ☐ 健康检查接口可用吗/health 返回200 ☐ 日志级别对了吗生产用 INFO不要用 DEBUG ☐ Prometheus 指标正常暴露了吗/metrics 可访问 ☐ Docker 镜像大小合理吗500MB 为佳 ☐ 敏感信息API Key走环境变量了吗思考 总结不要把 Jupyter 里的代码直接拿去上线一个能跑通的 API 和能上线的 API中间隔着队列、缓存、限流、监控四座大山。优先级排序限流 缓存 队列 监控。先保证不被刷爆和Token浪费再优化吞吐和可观测性。Semaphore 是最简单的并发控制方式不需要引入 Celery 这种重型任务队列。99% 的 LLM API 场景下asyncio.Semaphore就够了。Redis 缓存 ROI 最高GPT-4 每次调用几毛钱高频问题每天可能被问几百次。缓存一开Token 开销直接砍半。Docker Compose 让你一键启动全家桶API Redis Prometheus Grafana一条docker-compose up全搞定。生产级的 AI API 服务代码只占 30%基础设施缓存/限流/监控/部署占 70%。结尾各位小伙伴本文的内容到这里就全部结束了源码骑士在这里再次感谢您的阅读源码骑士 — Android Framework 全栈开发关注跟博主一起从源码视角深耕底层原理见证每一次成长❤️点赞让优质内容被更多人看见让知识传递更有力量⭐收藏把核心知识点存好在需要时随时查、随时用评论分享你的经验或疑问评论区一起交流避坑一键四连不要忘记给博主一键四连哦️寄语技术之路难免有困惑但同行的人会让前进更有方向结语从脚本到服务差的不是代码量是那层生产意识。本文的代码模板可以直接用到你的项目中——把业务 Chain 替换进去队列、缓存、限流、监控一个不少。不要忘记给博主一键四连哦
