企业官网建设流程全解析

基于 instructor 实现带精确引用的结构化事实抽取：FastAPI + SSE 流式服务实战

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor

本指南围绕仓库中的 citation_with_extraction 示例 展开，讲解如何用 instructor 构建一个 FastAPI 服务：让 LLM 基于给定上下文回答问题，并把答案拆解为携带**原文精确引用（citation）**的结构化事实，通过 Server-Sent Events（SSE）实时流式返回。读完本文，你将掌握 ResponseSchema 数据模型设计、流式多任务解析（MultiTaskBase / IterableBase）、模糊子串定位等关键技术，可直接复刻出一套可验证、可追溯的 RAG 引用抽取管线。

示例要解决的核心问题

在 RAG（检索增强生成）场景中，LLM 生成的答案往往"看似合理却无法溯源"。本示例给出一种工程化解法：让模型在回答每个事实（Fact）的同时，直接从原上下文中引用一段子串（substring quote）作为证据，服务端再通过精确的字符串定位把引用还原为原文中的起止位置（span），最终以 SSE 事件流返回{body, spans, citation}三元组。

整个服务的入口是 examples/citation_with_extraction/main.py：它定义一个 FastAPI 应用，暴露POST /extract端点，接收 JSON 格式的context（待回答的上下文文本）与query（问题），返回结构化、带精确引用的事实流。

数据模型：让"引用"成为模型输出的一等公民

Fact：一个事实 + 一组证据子串

在 main.py 中，每个事实被建模为Fact：

class Fact(BaseModel): fact: str = Field( ..., description="Body of the sentences, as part of a response, it should read like a sentence that answers the question", ) substring_quotes: list[str] = Field( ..., description="Each source should be a direct quote from the context, as a substring of the original content", )

• fact：一句读起来像答案的自然语言陈述；
• substring_quotes：支撑该事实的直接引用列表，要求是原上下文的真实子串。

字段描述（description）会被 instructor 写入 OpenAI 函数调用的 JSON Schema，是引导模型输出质量的关键——在这里它明确告诉模型"引用必须是原文子串"。

Fact还实现了两个定位方法：

def _get_span(self, quote, context): import regex minor, major = quote, context errs_ = 0 s = regex.search(f"({minor}){{e<={errs_}}}", major) while s is None and errs_ <= len(context) * 0.05: errs_ += 1 s = regex.search(f"({minor}){{e<={errs_}}}", major) if s is not None: yield from s.spans() def get_spans(self, context): if self.substring_quotes: for quote in self.substring_quotes: yield from self._get_span(quote, context)

这段逻辑值得细读：它使用第三方regex库的模糊匹配能力（{e<=N}语法，允许 N 个编辑错误），从容错为 0 开始逐步放宽，直到命中或错误上限超过上下文长度的 5%。这意味着即使模型引用的子串与原文存在轻微差异（如大小写、标点、空格），服务端依然能把它"锚定"回原文位置——这是保证spans和citation准确性的核心容错机制。

QuestionAnswer：多任务的流式输出容器

class QuestionAnswer(ResponseSchema, MultiTaskBase): question: str = Field(..., description="Question that was asked") tasks: list[Fact] = Field( ..., description="Body of the answer, each fact should be its separate object with a body and a list of sources", ) QuestionAnswer.task_type = Fact

QuestionAnswer同时继承两个基类：

• ResponseSchema：instructor 提供的响应模型基类（定义于 instructor/v2/core/function_calls.py），提供openai_schema类方法，把 Pydantic 模型序列化为 OpenAI 函数调用所需的 JSON Schema；
• MultiTaskBase（旧版 DSL 的多任务基类）：配合task_type = Fact声明"流式输出一批Fact子任务"。在 instructor v2 中，这一能力收敛为 instructor/v2/dsl/iterable.py 的IterableBase，其from_streaming_response类方法接收流式 completion，通过stream_extractor抽取 JSON 增量块，再用task_parser把每个任务解析为独立的 BaseModel 实例并逐条产出。

依赖关系如图 schema.png 所示（由 diagram.py 调用 erdantic 自动生成）：QuestionAnswer持有question: str与answer: List[Fact]，每个Fact又包含fact: str与substring_quote: List[str]，一对多的结构一目了然。

流式提取：从函数调用流到逐条事实

核心提取逻辑在 main.py 的stream_extract：

def stream_extract(question: Question) -> Iterable[Fact]: completion = client.chat.completions.create( model="gpt-4o-mini", temperature=0, stream=True, functions=[QuestionAnswer.openai_schema], function_call={"name": QuestionAnswer.openai_schema["name"]}, messages=[ {"role": "system", "content": "You are a world class algorithm to answer questions with correct and exact citations. "}, {"role": "user", "content": "Answer question using the following context"}, {"role": "user", "content": f"{question.context}"}, {"role": "user", "content": f"Question: {question.query}"}, {"role": "user", "content": "Tips: Make sure to cite your sources, and use the exact words from the context."}, ], max_tokens=2000, ) return QuestionAnswer.from_streaming_response(completion)

关键点：

• 模型实际使用gpt-4o-mini而非 README 标题中提到的 GPT-4（仓库当前代码以gpt-4o-mini为准，README 的描述已略滞后于代码）；
• temperature=0：让输出尽量确定，保证多次回答一致性，也便于引用定位；
• 以**函数调用（function calling）**方式驱动结构化输出：functions=[QuestionAnswer.openai_schema]把 Pydantic 模型注入工具定义，function_call={"name": ...}强制模型调用该函数；
• Prompt 末尾的 "Tips" 反复强调"引用原文、使用上下文的精确措辞"，与 Schema 中的字段描述互为呼应；
• 最后调用QuestionAnswer.from_streaming_response(completion)，把增量 token 流实时解析成一条条Fact生成器。

SSE 端点：把事实流推给客户端

/extract端点定义在 main.py：

@app.post("/extract", response_class=StreamingResponse) async def extract(question: Question, openai_key: str = Depends(get_api_key)): ... facts = stream_extract(question) async def generate(): for fact in facts: logger.info(f"Fact: {fact}") spans = list(fact.get_spans(question.context)) resp = { "body": fact.fact, "spans": spans, "citation": [question.context[a:b] for (a, b) in spans], } resp_json = json.dumps(resp) yield f"data: {resp_json}" yield "data: [DONE]" return StreamingResponse(generate(), media_type="text/event-stream")

它把"结构化抽取"与"精确引用"在响应层合二为一：

• 对每一条Fact，调用get_spans(question.context)把substring_quotes定位回上下文的具体字符区间（span）；
• 用 span 反切原文得到最终citation片段——这一步让客户端拿到的引用必然逐字命中原文；
• 以text/event-stream逐条推送data: {json}，最后发送data: [DONE]标记流结束。

请求体模型Question（main.py）也很简单：context与query两个必填字符串。

需要留意一个仓库现状：当前extract函数体首行raise Exception(...)（main.py），提示 "The 'openai.api_key' option isn't read in the client API"，即直接照搬 README 的 curl 示例运行会遇到该中断，正确做法是仿照提示，把Depends(get_api_key)取到的 key 显式传入OpenAI(api_key=openai_key)后创建客户端。这说明 README 中的端到端演示与最新代码之间存在待同步的偏差，复刻时请以源码中的异常提示为准。此外get_api_key（main.py）的取值顺序是：优先读环境变量OPENAI_API_KEY，否则解析请求头Authorization: Bearer <key>，缺失时返回 401。

动手实践：安装、运行与调用

安装依赖

仓库提供了 requirements.txt：

pip install -r requirements.txt

内容包含fastapi、uvicorn、openai>=1.0.0、pydantic、instructor、regex，其中regex是模糊子串定位的底层依赖，缺它_get_span无法工作。

启动服务

uvicorn main:app --reload

服务默认监听http://localhost:8000，/extract端点随之就绪；--reload便于开发期热更新。

用 curl 发起请求

curl -X POST -H "Content-Type: application/json" -d '{ "context": "My name is Jason Liu, and I grew up in Toronto Canada but I was born in China.I went to an arts highschool but in university I studied Computational Mathematics and physics. As part of coop I worked at many companies including Stitchfix, Facebook. I also started the Data Science club at the University of Waterloo and I was the president of the club for 2 years.", "query": "What did the author do in school?" }' -N http://localhost:8000/extract

-N（即--no-buffer）对 SSE 至关重要——它禁用 curl 的缓冲，让流式事件能逐条实时显示。

解析响应

README 给出的典型输出如下：

data: {'body': 'In school, the author went to an arts high school.', 'spans': [(91, 106)], 'citation': ['arts highschool']} data: {'body': 'In university, the author studied Computational Mathematics and physics.', 'spans': [(135, 172)], 'citation': ['Computational Mathematics and physics']}

每条事件包含三部分：body（事实陈述）、spans（引用在 context 中的起止字符位置，左闭右开）、citation（从 context 中按 span 切出的原文片段）。客户端拿到这些数据后，即可在原文上高亮标注每个事实的证据来源，实现"每个结论都指向原文"的可信输出。若服务部署在其他主机/端口，把http://localhost:8000替换为实际地址即可。

进阶变体：用 Pydantic 校验器兜底引用质量

同目录的 citation_fuzzy_match.py 提供了一个非流式 + 服务端校验的变体，适合对引用质量要求更高的离线场景，其核心差异有三点：

1. 模型校验器做引用落地（citation_fuzzy_match.py）：Fact上挂@model_validator(mode="after")，利用 instructor 传入的validation_context（代码中为{"text_chunk": context}）把每个substring_phrase定位回上下文，并把字段替换为真实的原文切片；若某条引用定位不到，则直接丢弃。
2. 回答级二次过滤（citation_fuzzy_match.py）：QuestionAnswer的校验器会把"没有任何证据子串"的事实从answer列表中剔除，保证每条输出都有据可依。
3. 通过response_model走非流式结构化输出（citation_fuzzy_match.py）：使用client.chat.completions.create(..., response_model=QuestionAnswer, validation_context={"text_chunk": context})，由 instructor 完成解析 + 校验闭环，运行该脚本（文件末尾自带示例问题与上下文）即可在日志中看到校验过程与最终 JSON。

部署选项：Docker 与 Modal

除了本地uvicorn，仓库提供两条部署路径：

• Docker（Dockerfile）：基于python:3.10-slim-bullseye，安装requirements.txt后执行uvicorn main:app --host 0.0.0.0 --port 8080，适合自托管；
• Modal 无服务器（modal_main.py）：直接复用main的app，以@stub.function+@modal.asgi_app()包装成 ASGI 应用，镜像仅需安装fastapi、instructor>=0.2.1、regex。README 中给出的https://jxnl--rag-citation-fastapi-app.modal.run/extract即为作者公开的 Modal 实例：带上你自己的Authorization: Bearer <OPENAI_API_KEY>即可试用，作者声明该代码公开且不存储你的 key。

curl -X 'POST' \ 'https://jxnl--rag-citation-fastapi-app.modal.run/extract' \ -H 'accept: */*' \ -H 'Content-Type: application/json' \ -H 'Authorization: Bearer <OPENAI_API_KEY>' \ -d '{ "context": "My name is Jason Liu, and I grew up in Toronto Canada but I was born in China.I went to an arts highschool but in university I studied Computational Mathematics and physics. As part of coop I worked at many companies including Stitchfix, Facebook. I also started the Data Science club at the University of Waterloo and I was the president of the club for 2 years.", "query": "What did the author do in school?" }'

注意事项与适用边界

• API Key 与用量：运行前需准备有效的 OpenAI API Key；本示例每次请求都会调用模型，请留意 OpenAI API 的使用限制与计费策略，合理控制请求频率与max_tokens（示例中设为 2000）。
• 引用质量依赖 Prompt 与 Schema 描述：substring_quotes是否真的逐字取自上下文，取决于模型遵循指令的能力；服务端模糊匹配只做"定位与兜底"，并不改写模型产出。
• 模糊匹配的代价：_get_span允许最多约 5% 上下文长度的编辑误差，容错放宽到极端时可能匹配到非预期位置，生产环境建议对 span 数量与位置做进一步约束（如限制为 1 处、校验长度比）。
• 版本偏差：README 描述的服务行为基于较早版本代码，当前 main.py 已改用gpt-4o-mini且/extract存在显式抛错提示，复刻时以源码为准；MultiTaskBase属于旧版 DSL，v2 中对应能力已统一到 instructor/v2/dsl/iterable.py 的IterableBase。

小结

本示例把 instructor 的"结构化输出"能力与"精确引用"工程需求结合，形成了一条完整链路：ResponseSchema建模 → 函数调用驱动流式输出 → 模糊子串定位 → SSE 逐条推送。它既是 RAG 应用中"可信引用"的参考实现，也是理解 instructor 流式多任务解析（MultiTaskBase/IterableBase与from_streaming_response）与validation_context校验机制的极佳范例，可直接迁移到文档问答、合规审计、知识图谱构建等需要"每句话都有出处"的场景。相关示例遵循 MIT 协议，仓库根目录的 LICENSE 说明了使用与分发条款。

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor