用 FastAPI 提供 DSPy 问答推理接口-未完纪

来源与署名：DSPy 官方教程维护方（原页未署个人作者）；未完纪翻译整理。原文：Tutorial: Deploying your DSPy program。本文为中文翻译整理；技术核对日期为 2026 年 10 月 5 日。

DSPy FastAPI 非流式请求经过输入验证和 asyncify 线程池调用模型，生产入口还需认证、额度、超时及脱敏策略（原创示意图，不是实测截图。）

DSPy 程序可以作为 Python 对象运行，也可以封装成服务，让其他应用通过 HTTP 调用。原教程展示两条路线：直接用 FastAPI 提供轻量 REST 接口，或用 MLflow 记录程序、环境与版本，再部署模型服务。本文以完整的 FastAPI 非流式主线作为实施范围；流式与 MLflow 部分完整解释其机制和原例问题，作为延伸阅读，不把附加片段拼成可直接投产的应用。

源页为滚动的 current 文档，核对日期 2026-10-05。它同时含 DSPy 2.6.0+ 与 MLflow 2.22.0 时点说明，不能据此推断所有最新版 API 都完全相同。本文没有安装依赖、调用收费模型、提交 API 密钥或启动服务。代码和响应均为来源说明或标明的编辑修改，不是运行报告。

先明确要服务的 DSPy 程序

源文使用最简单的问答模块作为例子：给出 question，生成 answer。可以把它换成更复杂、已经验证的 DSPy 模块。

import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
dspy_program = dspy.ChainOfThought("question -> answer")

模型标识来自原教程，不表示本文推荐读者必须使用该模型，也不保证该标识在未来仍可访问。模型调用可能收费，问题会被发送到模型服务方；不要在演示问题里提交秘密或未经许可的个人信息。

FastAPI：请求结构、异步封装与线程池

源文在已有 DSPy 环境中安装 FastAPI 与 Uvicorn，然后设置 OPENAI_API_KEY。下列安装行是准备说明，尚未执行；正式复现时应在隔离环境中固定 DSPy、FastAPI、Uvicorn 以及其依赖版本。密钥使用部署环境的秘密注入机制提供，不应写进源文件、笔记、命令历史或版本库。

python -m pip install fastapi uvicorn

下面保留原文非流式应用的完整逻辑，文件名可以是 fastapi_dspy.py：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import dspy

app = FastAPI(
    title="DSPy Program API",
    description="A simple API serving a DSPy Chain of Thought program",
    version="1.0.0"
)

class Question(BaseModel):
    text: str

lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm, async_max_workers=4)  # 原文说明默认是 8

dspy_program = dspy.ChainOfThought("question -> answer")
dspy_program = dspy.asyncify(dspy_program)

@app.post("/predict")
async def predict(question: Question):
    try:
        result = await dspy_program(question=question.text)
        return {"status": "success", "data": result.toDict()}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Pydantic 的 Question 定义 JSON 输入结构，FastAPI 据此验证请求并生成接口文档。dspy.asyncify 把原本同步的程序封装为可等待的异步调用。按源文说明，内部是在独立线程中运行 DSPy 程序，协程等待结果；默认最多八个并行工作线程，本例改为四个。超出容量的调用需要等待已有调用完成。

线程池容量不是模型供应商的速率限制，也不是用户额度、全局并发上限或超时策略。多个 Uvicorn 进程各自持有应用状态；若不控制入口，仍可能累积等待、消耗资源或超过模型配额。

对原文主线的两处最小修订

原例的 text 无长度上限，且把 str(e) 原样发给客户端。异常消息可能泄露供应商细节、路径、请求内容或其他敏感信息。下面是编辑修订，替换对应导入、请求模型和端点；DSPy 配置保持原例。修改仅增加输入长度约束与通用错误消息，正常响应仍返回原来的 result.toDict()。

from pydantic import BaseModel, Field

class Question(BaseModel):
    text: str = Field(min_length=1, max_length=8000)

@app.post("/predict")
async def predict(question: Question):
    try:
        result = await dspy_program(question=question.text)
        return {"status": "success", "data": result.toDict()}
    except Exception:
        raise HTTPException(
            status_code=500,
            detail="推理暂时失败，请稍后重试。"
        ) from None

不要同时保留两个 /predict 路由；修订片段是替换，而非追加。8000 是本稿选取的演示输入上限，不是模型的 token 上限或自动防注入机制；空白字符串也仍可能通过最小长度检查。原例返回的 Prediction 可能包含 reasoning 等字段，若产品只需答案，应定义输出模型并明确允许返回的字段。不要把模型输出当可执行指令、可信 HTML 或权限决策。

此修改不构成生产级部署：身份认证、每用户和全局额度、供应商限流、超时、取消策略、日志脱敏及监控仍需按服务设计实现。尤其线程中的同步工作在外层等待超时后可能继续运行，不能把简单协程超时误当成取消模型调用。

仅在本地启动并检查接口

原文使用开发热重载启动：

uvicorn fastapi_dspy:app --reload

它通常绑定 http://127.0.0.1:8000/。–reload 用于本地开发，不应作为生产部署选项。为了明确监听范围，可以按本地部署需要显式指定 –host 127.0.0.1。以下客户端是在读者已授权使用模型、设置好环境并启动服务之后的检查方式；本次没有执行。

import requests

response = requests.post(
    "http://127.0.0.1:8000/predict",
    json={"text": "What is the capital of France?"},
    timeout=60,  # 编辑补充：避免客户端无限等待
)
response.raise_for_status()  # 编辑补充：显式检查 HTTP 失败
print(response.json())

原文响应示例的 status 为 success，data 内含 reasoning 与 answer，答案表达“法国首都是巴黎”。模型输出并非固定字符串，不能把原文样例当作本文已验证的响应。客户端设置 timeout 只约束等待时间，不能保证服务端线程或收费请求随之停止。

延伸一：流式输出片段如何理解

原文称 DSPy 2.6.0+ 支持流式输出，并用 dspy.streamify 包装程序。它在最终 Prediction 前传回中间片段，底层沿用 asyncify 的执行语义。原始手写 SSE 片段如下：

dspy_program = dspy.asyncify(dspy.ChainOfThought("question -> answer"))
streaming_dspy_program = dspy.streamify(dspy_program)

@app.post("/predict/stream")
async def stream(question: Question):
    async def generate():
        async for value in streaming_dspy_program(question=question.text):
            if isinstance(value, dspy.Prediction):
                data = {"prediction": value.labels().toDict()}
            elif isinstance(value, litellm.ModelResponse):
                data = {"chunk": value.json()}
            yield f"data: {orjson.dumps(data).decode()}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

SSE 通过 data: 行与空行分隔事件，最后发送 [DONE] 标志。静态核对：片段没有导入 litellm、orjson、StreamingResponse；若迭代器返回未覆盖类型，data 可能未定义或沿用上一轮。当前返回类型还须与固定 DSPy 版本核对，不能把 ModelResponse 分支当通用约定。

原文另给出 streaming_response 帮助函数，以替代手工序列化：

from dspy.utils.streaming import streaming_response

@app.post("/predict/stream")
async def stream(question: Question):
    stream = streaming_dspy_program(question=question.text)
    return StreamingResponse(
        streaming_response(stream), media_type="text/event-stream"
    )

两段是互斥的实现方式，原页面以同一路径和同名函数展示，不能整体复制并注册两次。帮助函数例子仍需正确导入 FastAPI 响应类，并另行设计断连处理、流式错误、反向代理缓冲和敏感中间输出策略。本文没有扩展为完整流式服务，也没有把中间推理内容视为必须向用户公开的数据。

延伸二：MLflow 记录与服务程序

当需要将 DSPy 程序和依赖环境一起打包、记录版本和部署时，原文介绍 MLflow。它给出的最低安装版本为 2.18.0。下列把版本约束加引号，是为避免某些 shell 把大于号解析成重定向；不表示已经选择了完整锁定环境。

python -m pip install "mlflow>=2.18.0"
mlflow ui

按原文，UI/跟踪服务位于 http://127.0.0.1:5000/。这里的 log_model 是把程序及环境信息记录为 artifact，不只是写一条文本日志。教程指出，截至 MLflow 2.22.0，服务接口需要位置参数，而 DSPy 的 Predict/ChainOfThought 等预置模块不接受这样的调用方式，因此用自定义 Module.forward 包一层：

import dspy
import mlflow

mlflow.set_tracking_uri("http://127.0.0.1:5000/")
mlflow.set_experiment("deploy_dspy_program")
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

class MyProgram(dspy.Module):
    def __init__(self):
        super().__init__()
        self.cot = dspy.ChainOfThought("question -> answer")

    def forward(self, messages):
        return self.cot(question=messages[0]["content"])

dspy_program = MyProgram()
with mlflow.start_run():
    mlflow.dspy.log_model(
        dspy_program,
        "dspy_program",
        input_example={"messages": [
            {"role": "user", "content": "What is LLM agent?"}
        ]},
        task="llm/v1/chat",
    )

task=”llm/v1/chat” 用于让输入输出采用常见聊天接口的结构。此 wrapper 只读取 messages[0][“content”]；空列表、缺字段、非预期角色或复杂多轮消息都需要进一步校验。原文建议将其写到 mlflow_dspy.py 并执行，然后在 deploy_dspy_program 实验的 run → Artifacts 中查看记录。本文没有执行或产生 run id，也没有复制一张声称属于本次的 UI 截图。

路径纠错：上述 log_model 的 artifact_path 为 dspy_program，但源文服务命令使用 runs:/{run_id}/model。两者不一致。以下采用与示例记录名一致的路径；{run_id} 必须换为实际 run id，并核对该 MLflow 版本及实际 artifact 布局：

mlflow models serve -m "runs:/{run_id}/dspy_program" -p 6000

原文客户端调用 /invocations，以 messages 数组发送问题：

curl http://127.0.0.1:6000/invocations   -H "Content-Type:application/json"   --data '{"messages": [{"content": "what is 2 + 2?", "role": "user"}]}'

源文输出示例包含 choices 数组，message.role 为 assistant，content 是含 reasoning 与 answer 的 JSON 字符串，finish_reason 为 stop。它用“2 + 2 = 4”说明结构；本文没有收到该服务响应，也不承诺不同版本会返回完全相同封装。

MLflow 的环境、版本与部署边界

环境管理：用 conda.yaml 或 requirements.txt 记录 Python 依赖，并进一步固定复现环境。
版本管理：为程序版本添加有意义的标签与说明。
输入验证：定义清晰 schema 和 input_example，不能依赖示例消息永远存在。
监控：建立经过脱敏的日志、错误统计与性能监控；避免记录完整秘密或个人输入。

原文进一步展示容器打包，仍写了 /model 路径。下面只修正为与本页记录名一致的 dspy_program，并将示例发布端口绑定回环地址；这是编辑修改，未构建或运行：

mlflow models build-docker -m "runs:/{run_id}/dspy_program" -n "dspy-program"
docker run -p 127.0.0.1:6000:8080 dspy-program

原命令 -p 6000:8080 可能在所有宿主接口开放服务，而示例没有认证。容器本身也不会自动解决密钥注入、入口认证、限流、TLS 和日志策略。更完整的行为及版本差异应读 MLflow 官方文档，并在实际选定版本上核对 DSPy 集成接口。

来源、许可与核对范围

本次仅进行静态核对，实际发现了无鉴权/限流/超时、异常信息直出、输入未限长、流式导入缺失与重复路径、MLflow artifact 路径不一致等问题。主线修订、客户端错误检查、安装约束引号与回环端口绑定均明确标出。没有发现不等于不存在漏洞；静态核对不表示服务已经投产或通过安全测试。

原始版权与许可

DSPy 原始许可来源：官方 LICENSE。本篇为中文翻译与编辑整理。

MIT License

Copyright (c) 2023 Stanford Future Data Systems

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

文章版权归作者所有，未经允许请勿转载。

THE END