使用 LangSmith 与 Instructor 增强 OpenAI 客户端：多标签问题分类实战

使用 LangSmith 与 Instructor 增强 OpenAI 客户端多标签问题分类实战【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor本文以instructor仓库中的docs/examples/batch_classification_langsmith.md为蓝本讲解如何将 LangSmith 的wrap_openai与instructor的结构化输出能力叠加在同一个 OpenAI 异步客户端上实现带可观测性、带重试的多标签文本分类任务。读完本文你将掌握 LangSmith 与instructor的组合用法、Pydantic 约束分类输出的建模技巧以及用asyncio并发执行批量分类的标准套路。背景LangSmith 并非 LangChain 专属很多人误以为 LangChain 推出的 LangSmith 只能配合 LangChain 的模型链路使用。实际上LangSmith 是一个面向 LLM 应用研发全流程的统一 DevOps 平台覆盖开发、协作、测试、部署与监控等环节。它提供的wrap_openai包装器可以直接包裹任意 OpenAI 兼容客户端在不侵入业务代码的前提下记录 trace、统计 token 用量、观察每次调用的输入输出。因此LangSmith 完全可以与instructor一起工作LangSmith 负责看得见instructor负责结构化。本仓库给出了一个完整的可运行示例examples/batch-classification/run_langsmith.py下面将围绕它展开讲解。环境准备安装依赖并配置 API Key首先安装必要的 Python 包pip install -U langsmith pip install -U instructor使用 LangSmith 前需要先设置 API Key将下面的环境变量配置到你的 shell 或运行环境例如.env文件中export LANGCHAIN_API_KEYyour-api-key同时示例依赖 OpenAI 官方 SDK 与 Pydantic确保它们也已安装pip install openai pydantic三步组合wrap_openai → from_provider → 并发调度示例的核心只有三步代码非常紧凑import instructor import asyncio from langsmith import traceable from langsmith.wrappers import wrap_openai from openai import AsyncOpenAI from pydantic import BaseModel, Field, field_validator from typing import List from enum import Enum # 1. 用 LangSmith 包裹 OpenAI 客户端 client wrap_openai(AsyncOpenAI()) # 2. 用 instructor 将客户端升级为结构化输出客户端 client instructor.from_provider(openai/gpt-4o) # 3. 用一个信号量做简单的请求限流 sem asyncio.Semaphore(5)wrap_openai(AsyncOpenAI())这是 LangSmith SDK 提供的包装函数langsmith.wrappers它保持原有 API 不变但在背后为每次请求注入 trace 数据。需要注意的是wrap_openai仅对 OpenAI 官方 SDK 客户端生效。instructor.from_provider(openai/gpt-4o)这是instructor提供的模型字符串入口。其实现位于 instructor/v2/auto_client.py它要求模型字符串必须满足provider/model的格式例如openai/gpt-4o、anthropic/claude-3-sonnet否则会抛出ConfigurationError随后根据 provider 前缀查找对应的构建器构建底层 SDK 客户端并返回一个已具备response_model、max_retries等能力的Instructor或AsyncInstructor实例。由于传入的是AsyncOpenAI的包装对象得到的将是异步客户端。asyncio.Semaphore(5)限制同时进行中的请求不超过 5 个防止瞬时打爆 API 配额。如果你更习惯显式传入客户端的方式仓库示例同样演示了另一种等价写法见 examples/batch-classification/run.pyclient AsyncOpenAI() client instructor.from_openai(client, modeinstructor.Mode.TOOLS)from_openai的签名定义在 instructor/v2/providers/openai/client.py默认使用Mode.TOOLS即 tool call 模式进行结构化输出。定义分类 SchemaEnum 约束候选类别接下来用 Pydantic 定义分类结果的结构。这里的关键点有两个用Enum收紧候选类别用模型 docstring 注入分类提示词。class QuestionType(Enum): CONTACT CONTACT TIMELINE_QUERY TIMELINE_QUERY DOCUMENT_SEARCH DOCUMENT_SEARCH COMPARE_CONTRAST COMPARE_CONTRAST EMAIL EMAIL PHOTOS PHOTOS SUMMARY SUMMARY然后定义响应模型。模型类的 docstring 会被instructor注入到发给模型的提示中因此可以把每条类别的判定要点写在这里比把提示词全部塞进messages更内聚class QuestionClassification(BaseModel): Predict the type of question that is being asked. Here are some tips on how to predict the question type: CONTACT: Searches for some contact information. TIMELINE_QUERY: When did something happen? DOCUMENT_SEARCH: Find me a document COMPARE_CONTRAST: Compare and contrast two things EMAIL: Find me an email, search for an email PHOTOS: Find me a photo, search for a photo SUMMARY: Summarize a large amount of data # 如果只需要单标签把这里改成 classification: QuestionType 即可 chain_of_thought: str Field( ..., descriptionThe chain of thought that led to the classification ) classification: List[QuestionType] Field( descriptionfAn accuracy and correct prediction predicted class of question. Only allowed types: {[t.value for t in QuestionType]}, should be used, ) field_validator(classification, modebefore) def validate_classification(cls, v): # 有时 API 会返回单个值这里统一包装成 list if not isinstance(v, list): v [v] return v几个值得注意的设计点多标签 vs 单标签classification声明为List[QuestionType]即多标签分类一条问题可能命中多个类别若只需要单标签将其改为classification: QuestionType即可原文档与示例注释中都明确指出了这一切换方式。Chain-of-Thought 字段模型先输出chain_of_thought再给出分类既能提升准确率也让结果可解释。before 校验器field_validator(classification, modebefore)在 Pydantic 类型转换之前执行把模型偶尔返回的单个字符串规整为列表避免校验失败触发无谓的重试。这一模式在instructor的真实使用场景中非常常见。定义带 trace 的异步分类函数分类函数用traceable装饰器标记LangSmith 会据此在面板中生成名为classify-question的 span。函数内部先获取信号量再调用客户端traceable(nameclassify-question) async def classify(data: str) - QuestionClassification: Perform multi-label classification on the input text. Change the prompt to fit your use case. Args: data (str): The input text to classify. async with sem: # some simple rate limiting return data, await client.create( modelgpt-5.4-mini, response_modelQuestionClassification, max_retries2, messages[ { role: user, content: fClassify the following question: {data}, }, ], )modelgpt-5.4-mini与文档开头from_provider(openai/gpt-4o)中的模型名均为示例所用模型名实际部署时请替换为你可用的模型标识。response_modelQuestionClassificationinstructor的魔法参数。底层会将该 Pydantic 模型编译为 tool schema 随请求发出并对模型输出做 Pydantic 校验。max_retries2校验失败或网络异常时自动重试的最大次数。instructor底层create方法的默认值为 3见 instructor/v2/core/client.py这里显式收窄为 2 以减少开销。这里的client.create(...)是instructor客户端的高层 API在 examples/batch-classification/run_langsmith.py 中也可以看到client.chat.completions.create(...)这种 patch 后的原生调用写法两者最终走的是同一套响应模型处理与重试逻辑。批量并发执行与结果收集main函数把问题列表全部转成 task用asyncio.as_completed边完成边收集不依赖任务的提交顺序async def main(questions: List[str]): tasks [classify(question) for question in questions] for task in asyncio.as_completed(tasks): question, label await task resp { question: question, classification: [c.value for c in label.classification], chain_of_thought: label.chain_of_thought, } resps.append(resp) return resps if __name__ __main__: import asyncio questions [ What was that ai app that i saw on the news the other day?, Can you find the trainline booking email?, what did I do on Monday?, Tell me about todays meeting and how it relates to the email on Monday, ] resp asyncio.run(main(questions)) for r in resp: print(q:, r[question]) # q: what did I do on Monday? print(c:, r[classification]) # c: [SUMMARY]输出示例源自原文档问题what did I do on Monday?被分类为[SUMMARY]。若需把结果落盘为 JSONL 用于后续微调或评测可参考 examples/batch-classification/run.py 中的path_to_jsonl追加写入写法。源码级深入这两层包装究竟发生了什么组合wrap_openai与instructor时数据流大致如下wrap_openai(AsyncOpenAI())返回的包装客户端仍暴露chat.completions.create等原生接口LangSmith 在请求层完成 trace 记录。instructor.from_provider(openai/gpt-4o)按openai前缀路由到 OpenAI 构建器instructor/v2/auto_client.py构建器会从 kwargs 中提取base_url、organization、timeout、max_retries、default_headers等参数构造 SDK 客户端并最终调用instructor.from_openai(client, modelmodel_name, mode...或 Mode.TOOLS)返回已具备结构化输出能力的客户端。调用client.create(...)时Instructor.createinstructor/v2/core/client.py会对messages做规范化_normalize_messages透传response_model、context、max_retries、strict等参数并委托给 patch 后的底层创建函数。整个链路中 LangSmith 的 trace 依然有效因为请求最终仍经由被wrap_openai包裹的 HTTP 层发出。Mode枚举定义在 instructor/v2/core/mode.py其中TOOLS tool_call是 OpenAI 兼容提供商的结构化输出默认模式此外还有JSON_SCHEMA json_schema_mode、MD_JSON markdown_json_mode等模式可选它们控制着 schema 如何被注入请求以及输出如何被解析。常见变体与注意事项单标签分类把classification: List[QuestionType]改为classification: QuestionTypebefore 校验器同样生效。限流强度asyncio.Semaphore(5)中的 5 表示并发上限可根据你所用模型与账号的 RPM/TPM 配额调整并发越高asyncio.as_completed的吞吐优势越明显。重试策略max_retries可以传整数也可以传instructor内部的Retrying重试对象做更精细的指数退避与条件重试相关实现见 instructor/core/retry.py。观测数据运行完成后在 LangSmith 面板中可按 trace 名classify-question检索每次分类的输入、输出、耗时与 token 消耗便于定位异常样本与质量回归。关于模型名文档与示例中出现的gpt-4o、gpt-5.4-mini、gpt-4仅为当时演示所用请以你账号实际可用的模型标识为准并注意from_provider要求provider/model双段格式。延伸阅读仓库完整示例examples/batch-classification/run_langsmith.pyLangSmith 组合版、examples/batch-classification/run.py基础并发版、examples/batch-classification/run-cache.py缓存版。自动客户端入口与模型字符串解析instructor/v2/auto_client.py。OpenAI 提供商工厂函数instructor/v2/providers/openai/client.py。客户端create/create_iterable等高层 APIinstructor/v2/core/client.py。各提供商可用模式枚举instructor/v2/core/mode.py。分类相关概念的更多文档docs/concepts/classification、docs/concepts/batch、docs/concepts/iterable。【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考