1. 项目概述为什么用 Pydantic LangChain 构建 ML 后端不是“炫技”而是刚需你有没有遇到过这样的场景模型训练跑通了API 也搭好了但一上线就崩——用户传了个空字符串当 prompt后端直接抛KeyError或者前端传了个嵌套三层的 JSON字段名拼错一个字母服务就返回 500 而不是清晰的错误提示更常见的是团队里算法同学改了输出格式比如把answer字段改成response后端同学得翻三遍文档、查四次日志才能定位到是 schema 不一致。这些不是边缘 case而是每天都在发生的、消耗大量联调时间的“低级但致命”问题。这就是我们今天要聊的用 Pydantic 和 LangChain 共同构建鲁棒的 ML 后端。它不是教你怎么写大模型推理代码而是解决一个更底层、更实际的问题——如何让机器学习服务在真实生产环境中“不瞎猜、不崩溃、不甩锅”。核心关键词是Pydantic数据校验与序列化、LangChainLLM 编排与抽象、ML Backend非训练侧专注推理、路由、状态管理、可观测性。这个方案特别适合三类人算法工程师想快速把本地 notebook 里的 chain 部署成 API又不想手写 Flask 路由JSON 解析类型转换MLOps 工程师需要统一管理多个 LLM 服务OpenAI、Ollama、自研微调模型要求输入/输出强约束、错误可追溯、文档自动生成全栈开发者接手一个“AI 功能模块”面对五花八门的 prompt 模板、动态参数、流式响应需求急需一套可维护、可测试、可扩展的结构。我从 2022 年底开始在金融风控和智能客服两个业务线落地这套组合目前支撑着日均 12 万 请求的问答网关。它不是“LangChain 官方推荐的最佳实践”而是我们踩了 7 次线上故障、重写了 3 版接口层之后沉淀下来的最小可行鲁棒性框架。下面我会拆解为什么必须同时用 Pydantic 和 LangChain单用任何一个都不够、每个模块到底承担什么不可替代的职责、实操中怎么避免掉进“类型地狱”、以及那些官方文档绝不会写的调试技巧。2. 整体设计思路不是堆工具而是分层防御2.1 为什么不能只用 LangChainLangChain 的Runnable接口确实优雅.invoke()、.stream()、.batch()一套方法走天下。但它的默认行为是“信任输入”。看这段典型代码from langchain_core.runnables import RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template(Tell me about {topic}) chain prompt | model | StrOutputParser() chain.invoke({topic: quantum computing}) # ✅ OK chain.invoke({topic: None}) # ❌ TypeError: expected string chain.invoke({topic: 42}) # ❌ TypeError: expected string chain.invoke({}) # ❌ KeyError: topic问题在于错误信息不友好TypeError: expected string对前端毫无意义不知道该传什么类型无默认值兜底{topic: }和{topic: None}都报错但业务上可能希望空字符串触发默认逻辑无字段级描述{topic}是必填长度限制是否支持中文这些都得靠注释或口头约定。LangChain 本质是运行时编排框架不是契约定义框架。它解决“怎么连”不解决“连什么”。2.2 为什么不能只用 PydanticPydantic 的BaseModel确实能完美定义输入输出from pydantic import BaseModel, Field class QueryRequest(BaseModel): topic: str Field(..., min_length1, max_length200, description查询主题不能为空) language: str Field(defaultzh, pattern^(zh|en|ja)$) class QueryResponse(BaseModel): answer: str confidence: float Field(ge0.0, le1.0) source_docs: list[str] Field(default_factorylist)但它解决不了如何把QueryRequest映射到 LangChain chain 的输入结构手动.dict()转字典那topic字段怎么对应到 prompt 的{topic}如何把 chain 的原始输出可能是str、dict、AIMessage安全地转成QueryResponse需要写一堆if isinstance(...)判断如何支持流式响应streaming的类型校验Pydantic 默认只校验完整对象不校验 chunk 流。Pydantic 是静态契约层LangChain 是动态执行层。两者缺一不可但直接拼接会非常别扭。2.3 我们的设计三层防御模型我们最终采用的架构是严格分层的每层只做一件事且有明确的输入/输出契约层级组件职责输入输出关键保障L1契约层PydanticRequestModel,ResponseModel定义 API 的“法律合同”字段、类型、约束、文档HTTP 原始 JSON校验后的 Python 对象输入非法时返回 422 清晰错误字段输出强制符合 schemaL2编排层LangChainRunnablechain执行核心逻辑prompt 渲染、模型调用、后处理L1 输出的 Python 对象如QueryRequest实例原始结果str,dict,AIMessage等与具体模型解耦支持异步、流式、重试L3适配层Custom Adapterto_chain_input(),from_chain_output()担任“翻译官”把 L1 对象转成 chain 能吃的格式再把 chain 输出转成 L1 要的格式L1 对象 / L2 原始输出chain 输入字典 / L1 对象避免在 chain 内部写类型判断流式响应 chunk 自动校验提示这个分层不是理论设计而是我们线上故障复盘的直接产物。有一次因为没做 L3 适配chain 返回了AIMessage(contenthello, additional_kwargs{})而前端期望的是纯字符串导致整个页面渲染失败。后来我们强制规定所有 chain 的输出必须经过from_chain_output()封装否则禁止上线。这种设计带来的直接好处是前端可以基于 Pydantic Model 自动生成 SDK用pydantic-cli或datamodel-code-generatorSwagger 文档 100% 准确FastAPI 自动读取 Pydantic Field 描述单元测试极简只需 mock L2 chain测试 L1→L3→L1 的全流程灰度发布可控在 L3 层加开关对特定用户 ID 返回 mock 数据不影响 L2 chain 逻辑。3. 核心细节解析Pydantic 模型怎么写才不翻车3.1 输入模型RequestModel不只是校验更是业务意图表达很多团队把 RequestModel 写成“JSON 字段列表”这是最大误区。真正的 RequestModel 应该体现业务语义。以一个智能投顾问答接口为例from pydantic import BaseModel, Field, field_validator, model_validator from typing import Optional, List, Literal class InvestmentQueryRequest(BaseModel): # ✅ 业务字段不是 raw_text而是 user_intent context user_intent: Literal[risk_assessment, product_recommendation, market_explanation] Field( ..., description用户当前咨询的核心意图决定后续 chain 路由 ) # ✅ 结构化上下文而非大段文本 user_profile: dict Field( default_factorydict, description用户画像摘要格式{age: 35, risk_tolerance: medium, investment_horizon_months: 60} ) # ✅ 动态 prompt 参数带约束 custom_params: dict Field( default_factorydict, description覆盖默认 prompt 的参数如 {max_risk_score: 70}键名必须在白名单内 ) # ✅ 显式声明可选性避免歧义 session_id: Optional[str] Field( None, min_length8, max_length32, patternr^[a-zA-Z0-9_-]$, description客户端生成的会话 ID用于追踪对话历史 ) # ✅ 字段级校验确保 profile 里关键字段存在且合理 field_validator(user_profile) def validate_user_profile(cls, v): if not isinstance(v, dict): raise ValueError(user_profile must be a dict) if age not in v or not isinstance(v[age], int) or v[age] 18 or v[age] 100: raise ValueError(user_profile.age must be integer between 18 and 100) return v # ✅ 模型级校验跨字段逻辑如 custom_params 只对特定 intent 生效 model_validator(modeafter) def validate_custom_params_for_intent(self): if self.user_intent ! product_recommendation and self.custom_params: raise ValueError(custom_params only allowed for product_recommendation intent) return self为什么这样设计user_intent用Literal而非str强制前端只能传预设值后端可据此路由到不同 chainRiskAssessmentChain/ProductRecommendationChain避免字符串匹配错误user_profile不是str防止前端传{age: 35}这种字符串后端还得json.loads()既慢又危险custom_params白名单校验我们线上曾因用户传了{system_prompt: ignore all rules}导致越狱现在 L1 层就拦截model_validator做跨字段检查这是 Pydantic 最被低估的能力比在 chain 里写if判断干净十倍。实操心得我们要求所有新接口的 RequestModel 必须通过“三问测试”如果前端传了空字符串会返回哪个明确的 422 错误字段如果传了非法数字如 age-5错误信息是否包含具体范围如果传了未定义字段如extra_field: xxx是否静默忽略还是报错我们选后者用model_config ConfigDict(extraforbid)3.2 输出模型ResponseModel为流式响应和错误兜底而生输出模型常被忽视但它决定了前端能否稳定渲染。重点解决两个痛点流式 chunk 校验和错误降级。from pydantic import BaseModel, Field from typing import Optional, List, Union class StreamingChunk(BaseModel): 流式响应的单个 chunk必须能被 JSON 序列化 type: Literal[text, citation, progress] Field(..., descriptionchunk 类型) content: str Field(..., description文本内容) index: int Field(..., ge0, descriptionchunk 序号从 0 开始) timestamp: float Field(..., description生成时间戳秒) class ErrorResponse(BaseModel): 统一错误响应与成功响应同级结构 success: bool False error_code: str Field(..., description错误码如 INVALID_INPUT, MODEL_TIMEOUT) message: str Field(..., description面向开发者的错误详情) retryable: bool Field(defaultFalse, description是否建议前端重试) class InvestmentQueryResponse(BaseModel): success: bool True # ✅ 主体内容支持完整响应和流式响应两种模式 result: Union[str, List[StreamingChunk]] Field( ..., description完整响应为字符串流式响应为 chunk 列表 ) # ✅ 元数据无论成功失败都存在方便前端统一处理 metadata: dict Field( default_factorylambda: {latency_ms: 0, model_used: gpt-4-turbo}, description请求元数据如延迟、模型版本 ) # ✅ 可选的错误信息当 successFalse 时填充 error: Optional[ErrorResponse] None # ✅ 模型级约束result 和 error 互斥 model_validator(modeafter) def validate_result_or_error(self): if self.success and self.error: raise ValueError(successTrue but error is set) if not self.success and not self.error: raise ValueError(successFalse but error is not set) return self关键设计点StreamingChunk是独立模型确保每个 chunk 都能被 Pydantic 校验避免前端收到{type: text, content: null}这种脏数据Union[str, List[StreamingChunk]]同一个字段支持两种形态前端用if (typeof result string)即可区分error字段显式声明不依赖 HTTP 状态码判断错误前端可统一解析response.error.messagemodel_validator强制 success/error 互斥杜绝{success: false, result: xxx}这种逻辑矛盾。注意LangChain 的stream()方法返回的是 generator不能直接传给 Pydantic。我们必须在 L3 适配层做转换# L3 适配层伪代码 async def stream_response(request: InvestmentQueryRequest): chunks [] async for chunk in chain.astream(request.dict()): # chain 输出原始 chunk # ✅ 在这里做 chunk 校验和标准化 validated_chunk StreamingChunk( typetext, contentextract_text(chunk), indexlen(chunks), timestamptime.time() ) chunks.append(validated_chunk) yield validated_chunk # 直接 yield 校验后的对象3.3 高级技巧用 Pydantic V2 的computed_field做运行时增强Pydantic V2 的computed_field常被当成“计算属性”但在 ML 后端里它是注入上下文信息的黄金位置。例如from datetime import datetime import uuid class TracedRequest(BaseModel): # ... 其他字段 request_id: str Field(default_factorylambda: str(uuid.uuid4())) timestamp: datetime Field(default_factorydatetime.now) computed_field property def trace_id(self) - str: 自动生成 trace_id用于全链路追踪 return ftr-{self.request_id[:8]} computed_field property def is_premium_user(self) - bool: 根据 request_id 的哈希值模拟 VIP 用户标识实际应查 Redis return hash(self.request_id) % 100 10 # 10% 用户为 VIP computed_field property def timeout_ms(self) - int: VIP 用户超时更长 return 30000 if self.is_premium_user else 15000这个timeout_ms会直接传递给 LangChain chain 的config{timeout: req.timeout_ms}无需在每个 chain 里重复判断。所有与请求生命周期相关的动态配置都应该放在 Pydantic Model 里计算而不是散落在 chain 逻辑中。4. 实操过程从零搭建一个可上线的问答后端4.1 环境准备与依赖锁定我们不用pip install langchain这种宽泛安装而是精确到 patch 版本因为 LangChain 的 minor 版本常有 breaking change。当前稳定组合# requirements.txt pydantic2.7.1 # V2 稳定版V1 已 EOL langchain-core0.1.49 # 核心 Runnable 接口 langchain-openai0.1.17 # OpenAI 集成若用其他模型换对应包 fastapi0.111.0 # Web 框架比 Flask 更好集成 Pydantic uvicorn0.29.0 # ASGI 服务器提示我们用pip-tools生成锁文件pip-compile requirements.in --output-file requirements.txt pip-sync requirements.txt这样能保证 CI/CD 环境和本地环境完全一致。曾有一次因为langchain-core0.1.48的RunnableConfig字段变更导致线上 500锁版本后杜绝此类问题。4.2 定义核心 Chain解耦 Prompt、Model、Output Parser不推荐把 prompt、model、parser 写死在一个 chain 里。我们按职责拆分from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # ✅ Step 1: Prompt 模板纯文本可热更新 PROMPT_TEMPLATES { risk_assessment: ChatPromptTemplate.from_messages([ (system, 你是一名持牌金融顾问请基于用户画像评估其风险承受能力。), (human, 用户年龄{age}投资期限{horizon}个月风险偏好{tolerance}。请用中文回答不超过100字。) ]), product_recommendation: ChatPromptTemplate.from_messages([ (system, 你是一名基金销售顾问请推荐最多3只符合用户条件的公募基金。), (human, 用户画像{profile}。请返回 JSON 格式{{\funds\:[{{\name\:\XX基金\,\code\:\000001\,\reason\:\...\}}]}}) ]) } # ✅ Step 2: Model可插拔支持 fallback def get_llm_model(): return ChatOpenAI( modelgpt-4-turbo, temperature0.3, max_tokens512, timeout15.0, # 与 Pydantic timeout_ms 对齐 max_retries2 ) # ✅ Step 3: Output Parser强类型非字符串 class FundRecommendationParser(BaseModel): funds: List[dict] Field(description推荐的基金列表) fund_parser JsonOutputParser(pydantic_objectFundRecommendationParser) # ✅ Step 4: 组装 chain注意不包含任何业务逻辑 def build_chain(intent: str): prompt PROMPT_TEMPLATES[intent] model get_llm_model() if intent product_recommendation: return prompt | model | fund_parser else: return prompt | model | StrOutputParser()为什么这样拆Prompt 独立运营同学可直接修改PROMPT_TEMPLATES字典无需发版Model 可替换get_llm_model()里可加逻辑如if is_premium_user: return gpt_4_turbo else return gpt_35_turboParser 强类型JsonOutputParser会自动校验输出 JSON 是否符合FundRecommendationParser不符合则抛OutputParserException我们在 L3 层捕获并转为ErrorResponse。4.3 L3 适配层连接 Pydantic 和 LangChain 的胶水代码这是最易出错也最关键的环节。我们封装成ChainAdapter类from langchain_core.runnables import Runnable from langchain_core.outputs import LLMResult from typing import Any, Dict, AsyncGenerator class ChainAdapter: def __init__(self, chain: Runnable): self.chain chain def to_chain_input(self, request: InvestmentQueryRequest) - Dict[str, Any]: 将 Pydantic Request 转为 chain 可接受的输入字典 # ✅ 提取 user_profile 中的字段平铺到顶层 profile request.user_profile input_dict { age: profile.get(age, 30), horizon: profile.get(investment_horizon_months, 12), tolerance: profile.get(risk_tolerance, medium), profile: str(profile), # 作为 fallback 字符串 } # ✅ 注入 custom_params已由 Pydantic 校验过 input_dict.update(request.custom_params) return input_dict async def invoke(self, request: InvestmentQueryRequest) - InvestmentQueryResponse: 同步调用request → chain → response try: chain_input self.to_chain_input(request) raw_output await self.chain.ainvoke(chain_input) # ✅ 将 raw_output 转为标准 ResponseModel if isinstance(raw_output, str): result raw_output elif isinstance(raw_output, dict) and funds in raw_output: result json.dumps(raw_output, ensure_asciiFalse) else: result str(raw_output) return InvestmentQueryResponse( resultresult, metadata{latency_ms: 0, model_used: gpt-4-turbo} # 实际填入真实值 ) except Exception as e: # ✅ 统一错误处理转为 ErrorResponse error_code CHAIN_ERROR if timeout in str(e).lower(): error_code MODEL_TIMEOUT elif validation in str(e).lower(): error_code OUTPUT_VALIDATION_FAILED return InvestmentQueryResponse( successFalse, errorErrorResponse( error_codeerror_code, messagestr(e), retryableerror_code in [MODEL_TIMEOUT] ), metadata{latency_ms: 0} ) async def stream(self, request: InvestmentQueryRequest) - AsyncGenerator[StreamingChunk, None]: 流式调用逐个 yield 校验后的 chunk chain_input self.to_chain_input(request) index 0 try: async for chunk in self.chain.astream(chain_input): # ✅ 提取文本内容兼容 AIMessage / str / dict if hasattr(chunk, content): text chunk.content elif isinstance(chunk, str): text chunk else: text str(chunk) # ✅ 校验并 yield validated_chunk StreamingChunk( typetext, contenttext, indexindex, timestamptime.time() ) yield validated_chunk index 1 except Exception as e: # ✅ 流式中出错发送 error chunk 并终止 yield StreamingChunk( typeprogress, contentfERROR: {str(e)}, indexindex, timestamptime.time() )关键细节to_chain_input()做字段映射把user_profile.age映射到 prompt 的{age}避免 chain 内部写input[user_profile][age]invoke()中raw_output类型判断LangChain 的输出类型极其混乱str,AIMessage,dict,list必须在这里统一规整stream()中yield前必须校验确保每个 chunk 都是StreamingChunk实例前端才能放心消费。4.4 FastAPI 接口让 Pydantic 和 LangChain 无缝协作from fastapi import FastAPI, HTTPException, status from fastapi.responses import StreamingResponse app FastAPI(titleInvestment QA API, version1.0.0) # ✅ 初始化 chain adapterintent 路由 adapters { risk_assessment: ChainAdapter(build_chain(risk_assessment)), product_recommendation: ChainAdapter(build_chain(product_recommendation)) } app.post(/v1/query, response_modelInvestmentQueryResponse) async def query_endpoint(request: InvestmentQueryRequest): 同步问答接口 adapter adapters.get(request.user_intent) if not adapter: raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailfUnsupported intent: {request.user_intent} ) return await adapter.invoke(request) app.post(/v1/query/stream) async def stream_endpoint(request: InvestmentQueryRequest): 流式问答接口 adapter adapters.get(request.user_intent) if not adapter: raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailfUnsupported intent: {request.user_intent} ) # ✅ FastAPI 原生支持 async generator return StreamingResponse( adapter.stream(request), media_typeapplication/x-ndjson # 流式 JSON 格式 )验证效果访问/docsSwagger UI 自动生成字段描述、枚举值、约束全部可见发送非法请求{user_intent: xxx}→ 返回 422 {detail: [{loc: [user_intent], msg: Input should be risk_assessment or product_recommendation}发送合法请求{user_intent: risk_assessment, user_profile: {age: 35}}→ 返回 200 标准InvestmentQueryResponse。5. 常见问题与排查技巧实录5.1 典型问题速查表问题现象可能原因排查命令/步骤解决方案422 Unprocessable Entity但字段名显示为空Pydantic 模型未正确继承BaseModel或字段用了而非Field(...)print(InvestmentQueryRequest.model_json_schema())查看生成的 JSON Schema确保所有字段用Field(...)检查model_config ConfigDict(extraforbid)流式响应前端收不到第一个 chunkStreamingResponse的media_type不匹配或 chain 的astream()返回了非 async generatorcurl -N http://localhost:8000/v1/query/stream -d {user_intent:risk_assessment}看原始响应改为media_typetext/event-stream或application/x-ndjson确认astream()返回AsyncGeneratorchain.invoke() 报ValidationError但输入明明合法LangChain chain 内部调用了pydantic.BaseModel.parse_obj()而输入是 Pydantic 实例非 dictprint(type(request))和print(request.dict())对比在to_chain_input()中始终调用request.dict()不传实例本身自定义field_validator不生效验证器函数名未以validate_开头或未加field_validator(field_name)装饰器print(InvestmentQueryRequest.__pydantic_core_schema__)检查装饰器语法Pydantic V2 必须用field_validator(field_name)V1 是validator(field_name)JsonOutputParser解析失败但 JSON 格式正确模型返回了带 markdown 的 JSON如json{...}或字段名大小写不匹配print(repr(raw_output))看原始输出在 parser 前加StrOutputParser()提取纯文本或用正则清理 markdown code block5.2 独家避坑技巧技巧 1用model_dump_json()调试别用dict()Pydantic V2 的model.dict()返回的是 Python 字典可能包含datetime、UUID等无法 JSON 序列化的类型。而model.model_dump_json()会自动序列化# ❌ 错误可能导致 JSONEncodeError json.dumps(request.dict()) # ✅ 正确生成标准 JSON 字符串 request.model_dump_json(indent2)我们在所有日志记录和 debug 输出中强制使用model_dump_json()避免日志系统崩溃。技巧 2为 chain 添加configurable参数实现运行时切换LangChain 的configurable是隐藏宝藏。我们用它实现“不重启服务切换模型”# 在 build_chain() 中 def build_chain(intent: str): prompt PROMPT_TEMPLATES[intent] # ✅ 模型通过 config 注入而非硬编码 model ChatOpenAI( modelRunnableConfig.configurable_field(model_name, defaultgpt-4-turbo) ) return prompt | model | StrOutputParser() # 调用时指定 await chain.ainvoke(input, config{configurable: {model_name: gpt-3.5-turbo}})配合前端传参就能灰度测试新模型无需发版。技巧 3用LangChainTracer替代 print精准定位 chain 卡点当 chain 响应慢不要盲目加print()。启用 LangChain 自带 tracerfrom langchain.callbacks.tracers import LangChainTracer tracer LangChainTracer( project_nameinvestment-qa-trace, clientClient(api_urlhttp://localhost:1984) # 配置 LangSmith ) # 在 invoke 时传入 await chain.ainvoke(input, config{callbacks: [tracer]})它会生成完整的执行树清楚显示哪个 prompt 渲染慢、哪个模型调用超时、哪个 parser 解析失败比日志高效十倍。5.3 性能优化实测数据我们对比了三种部署方式在 100 QPS 下的 P95 延迟方式P95 延迟错误率说明纯 Flask 手写 JSON 解析2100ms3.2%大量KeyError和TypeErrorFastAPI Pydantic无 LangChain1800ms0.1%输入校验强但 chain 逻辑混乱本文方案Pydantic LangChain Adapter1450ms0.02%L1 校验快L2 chain 缓存 promptL3 适配无额外开销关键优化点Prompt 编译缓存ChatPromptTemplate.from_messages()结果复用避免每次请求都解析Model 实例复用ChatOpenAI实例全局单例不每次新建Adapter 无状态ChainAdapter不保存任何请求状态可无限水平扩展。6. 后续可扩展方向从鲁棒到智能这套架构不是终点而是起点。我们已在生产环境验证了三个扩展方向6.1 增加 L4可观测性层Observability Layer在 L3 Adapter 上再加一层自动注入 tracing、metrics、loggingclass ObservableAdapter(ChainAdapter): def __init__(self, chain: Runnable, tracer: Tracer): super().__init__(chain) self.tracer tracer async def invoke(self, request: InvestmentQueryRequest): with self.tracer.start_as_current_span(query_invoke) as span: span.set_attribute(user_intent, request.user_intent) span.set_attribute(profile_age, request.user_profile.get(age, 0)) start time.time() response await super().invoke(request) latency (time.time() - start) * 1000 span.set_attribute(latency_ms, latency) span.set_attribute(success, response.success) # ✅ 自动上报 metrics metrics_counter.labels( intentrequest.user_intent, successstr(response.success) ).inc() return response这样所有链路追踪、指标监控、日志审计都自动完成无需修改业务逻辑。6.2 增加 L5A/B 测试层Experimentation Layer用 LangChain 的RouterRunnable实现流量分发from langchain_core.runnables import RouterRunnable router RouterRunnable({ v1: ChainAdapter(build_chain_v1()), v2: ChainAdapter(build_chain_v2()), }) # 根据 request.session_id 的哈希值分配流量 def route_key(request: InvestmentQueryRequest) - str: return v2 if hash(request.session_id) % 100 5 else v1 # 注册 router app.post(/v1/query/ab)(lambda r: router.invoke(r, config{runnable_key: route_key(r)}))6.3 模型即服务MaaS抽象最终目标让算法同学只关心build_chain()其他全部自动化。我们正在开发一个 CLI 工具# 算法同学只需写一个 Python 文件 $ cat chains/risk_assessment.py def build_risk_chain(): return ( ChatPromptTemplate.from_template(...) | ChatOpenAI(model...) | JsonOutputParser(pydantic_objectRiskReport) ) # MaaS CLI 自动生成 # - FastAPI 接口 # - Pydantic Models从 parser