给「启明英语」做的小 E前九篇攒下的代码全都在demo.py里跑得漂漂亮亮。直到同事来找我说想把批改接到小程序里给学员用。我第一反应是很简单把 demo 改改就行。结果一动手才发现本地能跑和能用中间隔着一整座桥小程序怎么调我的 Python两个学生同时提交会话历史会不会串模型返回慢了用户盯着白屏吗这一篇是这个系列的收官我们把前九篇的能力装进一个 FastAPI 服务——结构化输出、上下文管理、稳定性兜底全部收拢成三个 HTTP 接口真正推到线上。一、先扫盲脚本和服务差在哪1. 脚本是你跑一下服务是谁都能调demo.py是你双击运行输出打到终端。产品需要的是任何客户端小程序、网页、群机器人通过 HTTP 请求拿到结构化响应。这一步的载体就是 Web 服务框架——这里用 FastAPI轻、类型友好、自带交互式接口文档。2. 并发和隔离是脚本里不存在、服务里必须处理的问题脚本一次只服务你一个人。服务要同时面对成百上千个学生会话隔离A 的问题不能出现在 B 的上下文里。无状态每次请求是独立的记忆要由服务端自己管。兜底任何异常都不能返回一堆看不懂的栈给前端。这三件事脚本里一样都没有所以改改 demo 就能上线是个幻觉。3. 依赖注入同一份业务代码测试和生产的区别只在喂哪个模型这是服务化最值钱的一个决定。把模型当参数传进来defcreate_app(llm)-FastAPI:...测试传假模型生产传真模型业务逻辑一份代码。没有它你每次想跑测试都得连真 API、花真钱。二、工程设计把九篇的能力收拢10-服务化上线/ ├── src/ │ ├── settings.py # 配置复用第 1 篇 │ ├── llm.py # 模型工厂复用第 1 篇 │ ├── schema.py # 请求/响应的数据结构 │ ├── corrector.py # 批改核心逻辑结构化输出 │ ├── limits.py # 输入校验与上限 │ ├── sessions.py # 会话仓库隔离 LRU 淘汰 │ ├── streaming.py # SSE 流式输出 │ ├── app.py # FastAPI 应用 │ └── server.py # uvicorn 启动入口 ├── tests/ # 27 个用例离线全绿 └── demo.py一张表看清前几篇的能力都去哪了前篇能力落点结构化输出第 4 篇correct()返回Correction不返回散文上下文管理第 3 篇sessions.py会话隔离 滑窗历史稳定性兜底第 8 篇失败永远返回可解析的ok:false三、先定合同数据结构schema.py服务上线后请求体和响应体就是前后端之间的合同。用 pydantic 把它写成代码classCorrection(BaseModel):error_types:list[ErrorType]Field(default_factorylist)needs_rewrite:boolFalsecorrected_sentence:strcomment:strclassCorrectResponse(BaseModel):ok:boolcorrection:Correction|NoneNoneerror:str|NoneNone注意CorrectResponse外面包了一层ok/error而不是直接把Correction当响应体。这么设计有一个很实际的理由前端永远先看ok再看correction永远不用 try 解析。失败和成功走同一个结构只是ok翻一下——这是服务端失败兜底的第一课。四、批改逻辑两个通路corrector.py把第 4 篇的结构化输出收拢成一个函数。保留两个通路SYSTEM你是「启明英语」的 AI 助教小 E负责批改学生写的英语句子。 要求 1. 逐句找出错误一个错误一条不要漏 2. 句子本身没错误时明确说这句没问题 3. 只输出 JSON不要输出其它任何文字 PROMPTChatPromptTemplate.from_messages([(system,SYSTEM),(human,{format_instructions}\n\n学生写的句子\n{essay}),])defcorrect_structured(llm,essay)-Correction:通路一with_structured_output走 function calling最稳。structuredllm.with_structured_output(Correction)returnstructured.invoke(PROMPT.format(essayessay))defcorrect_parse(llm,essay)-Correction:通路二PydanticOutputParser 自己 parse适合模型不支持工具调用的场合。parserPydanticOutputParser(pydantic_objectCorrection)promptPROMPT.partial(format_instructionsparser.get_format_instructions())responsellm.invoke(prompt.format(essayessay))returnparser.parse(getattr(response,content,))服务里默认用correct_parse——它不依赖模型的工具调用能力而且 parse 过程可控可测模型没按格式来时我们能拿到原始文本而不是一个莫名奇妙的异常。format_instructions会把 schema 的字段说明、类型、示例自动塞进提示词这是第 2 篇把提示词做成对象的又一次兑现。五、输入防线服务端不信任前端limits.py上线后请求来自谁、内容多长都不归你控制。所以服务端必须自己设限MAX_ESSAY_CHARS2000defvalidate_essay(essay:str)-str:text(essayor).strip()ifnottext:raiseInputError(essay 不能为空)iflen(text)MAX_ESSAY_CHARS:raiseInputError(fessay 过长最多{MAX_ESSAY_CHARS}字符)returntext批改接口本来吃的是一句话超长的一般是误传或恶意请求。设 2000 字符超过直接 422。别信任前端已经校验过了这句话。六、会话隔离并发安全 会清理sessions.py第 3 篇讲过会话隔离这里补上服务端常驻进程带来的两个新问题——并发安全和内存无限涨dataclassclassConversation:session_id:strhistory:dequefield(default_factorylambda:deque(maxlenMAX_TURNS*2))deque(maxlen...)会在超长时自动丢掉最老的天然实现滑窗。仓库本身带锁和 LRU 淘汰classSessionStore:defget(self,session_id:str)-Conversation:withself._lock:ifsession_idnotinself._sessions:self._sessions[session_id]Conversation(session_idsession_id)iflen(self._sessions)self._max:# 超上限淘汰最久的self._sessions.popitem(lastFalse)else:self._sessions.move_to_end(session_id)returnself._sessions[session_id]内存版会话仓库够单进程小规模用多进程部署必须换成 Redis——这是上线清单里的一条硬项后面会讲。七、流式输出字是一个个蹦出来的streaming.py学生提交后看着空白页等 3 秒和看着字一个个蹦出来体感天差地别。流式不是炫技它直接决定这个产品快不快的主观感受。SSEServer-Sent Events的格式就两条一行data: json再空一行循环往复defsse_event(payload:dict)-str:returnfdata:{json.dumps(payload,ensure_asciiFalse)}\n\ndefsse_events(llm,messages):fortokeninstream_tokens(llm,messages):yieldsse_event({delta:token})yieldsse_event({done:True})# 结束标志绝不能少最后那条{done: true}是整段代码的命门不发它前端就不知道这轮说完了会一直等或把超时误判成结束。这里有个常见疑问为什么不直接用 WebSocket因为对话是一问一答的单向流SSE 更简单——它走普通 HTTP不需要专门的连接管理、不需要处理心跳还天然能被负载均衡和日志系统看懂。只有当你需要双向实时、客户端也能主动发比如多人协作白板时才上 WebSocket。能省则省这是工程里最划算的克制。八、组装三个接口app.pycreate_app(llm)工厂函数把上面所有模块组装成三个接口方法路径作用GET/health探活给监控/负载均衡用POST/correct同步批改返回结构化结果POST/chat/stream流式对话SSE 逐字返回批改要稳定、要结构对话要快、要流式——所以拆成两个接口而不是硬塞进一个。看流式接口的组装app.post(/chat/stream)defchat_stream(req:ChatRequest)-StreamingResponse:messagevalidate_message(req.message)session_idreq.session_idordefaultconversationstore.get(session_id)conversation.append_user(message)messagesconversation.to_messages(store.system)defgenerate():parts[]fortokeninstream_tokens(llm,messages):parts.append(token)yieldsse_event({delta:token})conversation.append_ai(.join(parts))# 回答也记进历史yieldsse_event({done:True})returnStreamingResponse(generate(),media_typetext/event-stream)两处细节流式过程边吐边把 token 累积起来结束后写回历史——这样下一轮才记得这次聊了什么失败时抛HTTPException(422)前端拿到的是一个明确的错误而不是 500 加一坨栈。九、启动uvicorn 跑起来server.pysettingsload_settings()llmbuild_llm(settings,temperature0.3)# 批改要稳定温度压到 0.3appcreate_app(llm)importuvicorn uvicorn.run(app,hostargs.host,portargs.port)python src/server.py起服务浏览器打开http://127.0.0.1:8000/docs就是 FastAPI 自带的交互式接口文档——不用装 Postman直接在网页上试接口。十、跑起来三个接口走一遍python demo.py全程离线不起真实服务器用 TestClient 直接打接口一、健康检查 GET /health 200 {status: ok, model: FakeListChatModel} 二、同步批改 POST /correct 状态码200 ok True 错误类型[时态] 需重写 True 改写 I went to the park yesterday. 解释 yesterday 是过去时间go 要改成 went。 边界空 essay 会被拦下返回 422 而不是 500 状态码422 三、流式对话 POST /chat/streamSSE content-type: text/event-stream; charsetutf-8 data: {delta: 这} data: {delta: 句} data: {delta: 时} data: {delta: 态} ... data: {done: true}看第三段流式接口吐出来的是一串delta最后以done收尾——这就是前端逐字渲染的原料。测试结果27 个用例全绿全程离线tests/test_corrector.py 4 个 结构化批改、两个通路、无错句 tests/test_limits.py 6 个 空/超长输入被拒、边界长度 tests/test_streaming.py 4 个 SSE 格式、逐字流、done 收尾 tests/test_sessions.py 6 个 隔离、滑窗、LRU 淘汰 tests/test_app.py 7 个 健康检查、批改、流式、历史记忆 27 passed in 1.40s这组测试里我最在意test_stream_keeps_history_across_turns它断言流式完成后回答真的进了下一轮请求——上下文记忆这种 bug 最阴险因为它不报错只是回答悄悄变蠢。十一、上线清单从能跑到能用还差这些代码跑通了不代表能上线。这张清单是我踩坑踩出来的每一条都对应一次真实事故.env配好 Key别提交 GitKey 进仓库等于把钱包密码发群里。.gitignore里已经排除了.env养成习惯。/health接进监控否则服务挂了没人知道等家长投诉才发现——那时已经晚了。加限流limits.py只限长度不限频率恶意请求能刷爆额度要再接一层频率限制比如按 IP 或 token 限次。会话仓库换 Redissessions.py是内存版多进程部署会各存各的用户换个请求就失忆必须换外部存储。接日志和告警别用print。出错时你需要的是带时间戳、能检索的日志以及一个能半夜叫醒人的告警。结构化输出加缓存同一句批改结果可以缓存重复请求不重复花钱——这和第 3 篇的上下文缓存是同一个省钱思路。这些看起来是运维的活但上线从来不只是写代码。产品跑在线上你就要对它的每一次调用负责。十二、踩坑清单“改改 demo 就能上线”脚本没有并发、隔离、兜底直接上线必翻车。响应体不做ok/error分层前端被迫 try 解析失败和成功混在一起。流式不发done前端永远不知道这轮结束把超时当结束。服务端信了前端校验不设输入上限恶意请求刷爆额度。会话放内存就跑多进程各进程各存各的用户换个请求就失忆。Key 提交进 Git泄露一次账单惊魂一次。十三、系列收尾十篇一条从零到上线的路回头把十篇串起来其实是一条完整的路第 1-2 篇环境配好提示词写对——先把话说清楚。第 3 篇管住上下文省下真金白银——先别烧冤枉钱。第 4 篇输出结构化——让程序能读懂模型的话。第 5 篇RAG 检索——让它开卷考别瞎编。第 6 篇LangGraph 状态机——把复杂流程拆成可回退的步骤。第 7 篇工具调用——给它接上真正能动手的手。第 8 篇稳定性——重试、超时、降级三条保险丝。第 9 篇评测——让每一次改动都能被打分。第 10 篇服务化——把这一切装进一个能上线的服务。这就是用 Python 做一个智能体产品全过程的骨架。你不需要一次掌握全部——从第 1 篇开始写一个真正能跑的 demo比读完十篇更有用。需要哪块能力就回来翻对应的那一篇。最后说几句这十篇反复出现、也最想留给你的三句话先把话说清楚提示词再别烧冤枉钱上下文然后让程序读懂它结构化输出。这三件事做扎实了后面的检索、编排、工具、稳定性、评测、上线都是顺水推舟。系列写到这里就收官了。但启明英语的小 E这个故事还没结束——它只是你手上下一个项目的一个影子。最后一句能跑的代码是你的本事能上线的服务才是别人的依赖——差的那段路叫工程。