基于 instructor 实现带精确引用的结构化事实抽取FastAPI SSE 流式服务实战【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor本指南围绕仓库中的 citation_with_extraction 示例 展开讲解如何用 instructor 构建一个 FastAPI 服务让 LLM 基于给定上下文回答问题并把答案拆解为携带**原文精确引用citation**的结构化事实通过 Server-Sent EventsSSE实时流式返回。读完本文你将掌握 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 中每个事实被建模为Factclass Fact(BaseModel): fact: str Field( ..., descriptionBody of the sentences, as part of a response, it should read like a sentence that answers the question, ) substring_quotes: list[str] Field( ..., descriptionEach 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库的模糊匹配能力{eN}语法允许 N 个编辑错误从容错为 0 开始逐步放宽直到命中或错误上限超过上下文长度的 5%。这意味着即使模型引用的子串与原文存在轻微差异如大小写、标点、空格服务端依然能把它锚定回原文位置——这是保证spans和citation准确性的核心容错机制。QuestionAnswer多任务的流式输出容器class QuestionAnswer(ResponseSchema, MultiTaskBase): question: str Field(..., descriptionQuestion that was asked) tasks: list[Fact] Field( ..., descriptionBody of the answer, each fact should be its separate object with a body and a list of sources, ) QuestionAnswer.task_type FactQuestionAnswer同时继承两个基类ResponseSchemainstructor 提供的响应模型基类定义于 instructor/v2/core/function_calls.py提供openai_schema类方法把 Pydantic 模型序列化为 OpenAI 函数调用所需的 JSON SchemaMultiTaskBase旧版 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_extractdef stream_extract(question: Question) - Iterable[Fact]: completion client.chat.completions.create( modelgpt-4o-mini, temperature0, streamTrue, 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: fQuestion: {question.query}}, {role: user, content: Tips: Make sure to cite your sources, and use the exact words from the context.}, ], max_tokens2000, ) return QuestionAnswer.from_streaming_response(completion)关键点模型实际使用gpt-4o-mini而非 README 标题中提到的 GPT-4仓库当前代码以gpt-4o-mini为准README 的描述已略滞后于代码temperature0让输出尽量确定保证多次回答一致性也便于引用定位以**函数调用function calling**方式驱动结构化输出functions[QuestionAnswer.openai_schema]把 Pydantic 模型注入工具定义function_call{name: ...}强制模型调用该函数Prompt 末尾的 Tips 反复强调引用原文、使用上下文的精确措辞与 Schema 中的字段描述互为呼应最后调用QuestionAnswer.from_streaming_response(completion)把增量 token 流实时解析成一条条Fact生成器。SSE 端点把事实流推给客户端/extract端点定义在 main.pyapp.post(/extract, response_classStreamingResponse) 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(fFact: {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 fdata: {resp_json} yield data: [DONE] return StreamingResponse(generate(), media_typetext/event-stream)它把结构化抽取与精确引用在响应层合二为一对每一条Fact调用get_spans(question.context)把substring_quotes定位回上下文的具体字符区间span用 span 反切原文得到最终citation片段——这一步让客户端拿到的引用必然逐字命中原文以text/event-stream逐条推送data: {json}最后发送data: [DONE]标记流结束。请求体模型Questionmain.py也很简单context与query两个必填字符串。需要留意一个仓库现状当前extract函数体首行raise Exception(...)main.py提示 The openai.api_key option isnt read in the client API即直接照搬 README 的 curl 示例运行会遇到该中断正确做法是仿照提示把Depends(get_api_key)取到的 key 显式传入OpenAI(api_keyopenai_key)后创建客户端。这说明 README 中的端到端演示与最新代码之间存在待同步的偏差复刻时请以源码中的异常提示为准。此外get_api_keymain.py的取值顺序是优先读环境变量OPENAI_API_KEY否则解析请求头Authorization: Bearer key缺失时返回 401。动手实践安装、运行与调用安装依赖仓库提供了 requirements.txtpip install -r requirements.txt内容包含fastapi、uvicorn、openai1.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 提供了一个非流式 服务端校验的变体适合对引用质量要求更高的离线场景其核心差异有三点模型校验器做引用落地citation_fuzzy_match.pyFact上挂model_validator(modeafter)利用 instructor 传入的validation_context代码中为{text_chunk: context}把每个substring_phrase定位回上下文并把字段替换为真实的原文切片若某条引用定位不到则直接丢弃。回答级二次过滤citation_fuzzy_match.pyQuestionAnswer的校验器会把没有任何证据子串的事实从answer列表中剔除保证每条输出都有据可依。通过response_model走非流式结构化输出citation_fuzzy_match.py使用client.chat.completions.create(..., response_modelQuestionAnswer, validation_context{text_chunk: context})由 instructor 完成解析 校验闭环运行该脚本文件末尾自带示例问题与上下文即可在日志中看到校验过程与最终 JSON。部署选项Docker 与 Modal除了本地uvicorn仓库提供两条部署路径DockerDockerfile基于python:3.10-slim-bullseye安装requirements.txt后执行uvicorn main:app --host 0.0.0.0 --port 8080适合自托管Modal 无服务器modal_main.py直接复用main的app以stub.functionmodal.asgi_app()包装成 ASGI 应用镜像仅需安装fastapi、instructor0.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属于旧版 DSLv2 中对应能力已统一到 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考