langchain-openai 从第一行 invoke 到 Azure 部署:ChatOpenAI 流式、工具调用、Responses API 完整实战指南
【免费下载链接】langchainThe agent engineering platform.项目地址: https://gitcode.com/GitHub_Trending/la/langchain
如果你要在 Python 里把 GPT 系列模型接进自己的业务,langchain-openai是目前最省事的桥梁:它把官方openaiSDK 的能力对齐成 LangChain 的标准接口,ChatOpenAI负责聊天与流式输出,OpenAIEmbeddings负责向量化,Azure 三件套负责企业部署。本文按"场景 + 能力"两条线拆解这个包:先给你一张能力总表定位每个类,再按"跑通 → 流式统计 → 工具与结构化 → 高级 API → Azure → 排错"的实际落地顺序逐节展开,每一节都只保留最短可运行片段和必须知道的坑。
能力速查:先对号入座找类
在动手之前,先花一分钟确认你要用的到底是哪个入口。下表是 langchain_openai/ 顶层导出的全部成员,直接按场景对号入座:
| 类 / 符号 | 模块 | 适用场景 | 一句话定位 |
|---|---|---|---|
ChatOpenAI | chat_models/ | 绝大多数聊天场景 | 继承BaseChatOpenAI,同时覆盖 Chat Completions 与 Responses API 两条通道 |
AzureChatOpenAI | 同上 | Azure 上部署的 OpenAI 服务 | 增加azure_deployment、api_version等 Azure 专属字段 |
OpenAI | llms/ | 非聊天的纯文本补全 | 对应gpt-3.5-turbo-instruct这类 instruct 模型,继承BaseLLM |
AzureOpenAI | 同上 | Azure 上的传统补全 | OpenAI的 Azure 变体 |
OpenAIEmbeddings | embeddings/ | 文本向量化 | 内置超长文本分块 + 加权平均 |
AzureOpenAIEmbeddings | 同上 | Azure 向量化 | OpenAIEmbeddings的 Azure 变体 |
custom_tool | tools/ | OpenAI 自定义工具 | 装饰器,让工具接受自由格式的长字符串输入 |
StreamChunkTimeoutError | chat_models._client_utils | 流式排错 | 流式分块等待超时时抛出的异常 |
它本质上是"包装层"而非替代品:网络请求仍然走openaiSDK,这个包负责参数对齐、response_metadata/usage_metadata的归一化,以及把 OpenAI 的异常翻译成语义更清晰的 LangChain 异常。版本信息上,当前发行版为1.6.7,要求 Python>=3.10,<4.0,核心依赖是langchain-core>=1.6.6,<2.0.0、openai>=2.45.0,<4.0.0、tiktoken>=0.7.0,<1.0.0和certifi,具体见 pyproject.toml。
从安装到密钥:把第一条消息跑起来
装包与配置密钥
两条命令就能让环境就绪。官方推荐uv,pip完全等价:
uv add langchain-openai # 或者 pip install -U langchain-openai密钥通过环境变量注入,不用写进代码:
export OPENAI_API_KEY="your-api-key"从 chat_models/base.py 的客户端参数定义可以确认解析优先级:api_key缺省时读OPENAI_API_KEY;organization回退到OPENAI_ORG_ID;base_url则是"显式传参 →OPENAI_API_BASE→OPENAI_BASE_URL"三级回退,最后一级由底层 SDK 接管。
在 langchain 仓库内联调时,pyproject.toml 的
[tool.uv.sources]把langchain-core、langchain、langchain-tests指向../../core、../../langchain_v1、../../standard-tests的可编辑安装,改动核心包无需重新打包。
第一条消息与返回结构
from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4.1-mini", temperature=0, max_retries=2) ai_msg = model.invoke([ ("system", "You are a helpful translator. Translate to French."), ("human", "I love programming."), ]) print(ai_msg.content) print(ai_msg.usage_metadata) # input / output / total tokens返回值是标准AIMessage,两个字段最常用:usage_metadata给出input_tokens/output_tokens/total_tokens;response_metadata携带model_name、finish_reason、system_fingerprint、token_usage,开了logprobs时还有logprobs。生成侧常用参数是model、temperature、max_tokens、logprobs、stream_options;客户端侧是timeout、max_retries、api_key、base_url、organization。
这里有个很实用的透传机制:任何没有被显式声明为类字段的参数,都会原样塞进底层chat.completions.create(...)。所以ChatOpenAI(..., frequency_penalty=0.2)、初始化后.bind(frequency_penalty=0.2)、或者在invoke时临时传frequency_penalty=0.2,三者在请求层面完全等价——OpenAI SDK 的参数面在这里是完整可达的。
流式输出与 token 统计:两个最容易踩的坑
边生成边打印
for chunk in model.stream(messages): print(chunk.text, end="")stream吐出的每个chunk都是AIMessageChunk,需要聚合完整消息时直接累加:full = next(stream),然后循环full += chunk。异步场景把invoke/stream/batch换成ainvoke/astream/abatch即可,用法一一对应。
流式里为什么看不到 usage
这是生产环境最高频的疑问:非流式调用自带usage_metadata,流式却拿不到。原因是 OpenAI 默认不在流里推送用量,必须显式要求。
for chunk in model.stream(messages, stream_usage=True): if chunk.usage_metadata: print(chunk.usage_metadata)
stream_usage=True等价于在请求里下发stream_options={"include_usage": True},初始化时传stream_options也能达到同样效果。忘记开这个开关,流式日志里就永远只有内容、没有账本。
工具调用与结构化输出:让模型"按规矩办事"
bind_tools 绑定 Pydantic 工具
ChatOpenAI原生对接 OpenAI 函数调用协议,工具定义就是普通的 Pydantic 模型:
from pydantic import BaseModel, Field class GetWeather(BaseModel): """Get the current weather in a given location""" location: str = Field(..., description="City and state, e.g. San Francisco, CA") model_with_tools = model.bind_tools([GetWeather]) ai_msg = model_with_tools.invoke("Which city is hotter today: LA or NY?") print(ai_msg.tool_calls)三个控制旋钮值得记住:strict=True强制模型严格遵循参数的 JSON Schema;parallel_tool_calls默认True(OpenAI SDK>=1.32起支持),想禁用可以在invoke(..., parallel_tool_calls=False)临时关,也可以.bind(parallel_tool_calls=False)固定;还能通过model_kwargs在初始化时写死。
结构化输出、JSON 模式与 custom_tool
class Joke(BaseModel): setup: str = Field(description="The setup of the joke") punchline: str = Field(description="The punchline to the joke") rating: int | None = Field(description="How funny, 1 to 10") result = model.with_structured_output(Joke).invoke("Tell me a joke about cats")with_structured_output直接返回 Pydantic 实例;而纯 JSON 约束走model.bind(response_format={"type": "json_object"}),模型返回原始 JSON 字符串,需要自己解析。
另一条路是 OpenAI 的自定义工具(Custom Tools):工具输入是一段自由格式的长字符串,而不是结构化参数。tools/custom_tool.py 里的装饰器会把函数返回值包装成{"type": "custom_tool_call_output", "output": ...}结构:
from langchain_openai import custom_tool @custom_tool def execute_code(code: str) -> str: """Execute python code.""" return "27"装饰器还支持format参数,为工具输出附加上下文无关文法(grammar)约束。
Responses API:推理展示、服务端会话与 flex 计费
开启方式与 output_version
Chat Completions 之外的新通道/v1/responses,两个开关即可启用:
model = ChatOpenAI( model="...", use_responses_api=True, output_version="responses/v1", )output_version="responses/v1"的作用是把推理摘要、内置工具调用这类响应条目直接格式化进消息的content字段,而不是塞在additional_kwargs里——新应用建议直接采用这种输出版本。
推理模型与 reasoning 参数
推理模型的"思考过程"可以要求服务端摘要后回传,类比成"先给你剧透版剧情,再给正文":
reasoning = {"effort": "medium", "summary": "auto"} # effort: low|medium|high; summary: detailed|auto|None model = ChatOpenAI(model="...", reasoning=reasoning, output_version="responses/v1") response = model.invoke("What is 3^3?") for block in response.content: if block["type"] == "reasoning": for summary in block["summary"]: print(summary["text"])会话状态续接与 service_tier
Responses API 允许服务端替你保管上下文,像电话免提一样:只要把上一轮的响应 ID 回传,历史消息就不用重复发送。
first = model.invoke("Hi, I'm Bob.") second = model.invoke( "What is my name?", previous_response_id=first.response_metadata["id"], )初始化时设use_previous_response_id=True可以全自动:请求载荷自动丢弃到最近一次响应为止的输入,并用最近响应 ID 填充previous_response_id。
对接 OpenAI 兼容端点又启用了推理时,如果频繁出现请求格式错误,优先切换到
use_previous_response_id=True,让会话状态交给服务端保管,比客户端拼装历史更稳。
成本档位方面,service_tier="flex"选择更便宜的 flex 层级,代价是响应更慢、资源不保证可用(beta 特性,仅部分模型支持),适合批量评测、数据增强这类离线任务。
Azure OpenAI 部署:参数、环境变量与令牌认证
环境变量与最小配置
Azure 的寻址方式与官方 API 完全不同——不是模型名,而是"部署名 + REST API 版本":
export AZURE_OPENAI_API_KEY="your-api-key" export AZURE_OPENAI_ENDPOINT="https://your-endpoint.openai.azure.com/"from langchain_openai import AzureChatOpenAI model = AzureChatOpenAI( azure_deployment="your-deployment-name", api_version="2024-...", # Azure REST API 版本,注意不是模型版本 )字段定义见 chat_models/azure.py:azure_endpoint缺省读AZURE_OPENAI_ENDPOINT;openai_api_type用于兼容旧式配置。Embeddings 与补全模型的 Azure 变体AzureOpenAIEmbeddings、AzureOpenAI分别位于 embeddings/azure.py 和 llms/azure.py,字段约定与聊天模型一致。
用 Entra ID 令牌替代 API Key
企业内网环境常常禁用静态密钥,Azure 支持基于 Microsoft Entra ID(原 Azure AD)的令牌认证:
model = AzureChatOpenAI( azure_deployment="your-deployment-name", azure_ad_token="token", # 或 azure_ad_token_provider 动态提供令牌 )azure_ad_token适合短生命周期的静态令牌,azure_ad_token_provider适合需要自动刷新的场景,二者选其一即可替代api_key。
生产排错:异常映射、兼容服务与工程化
九类异常一张表
OpenAI SDK 的裸异常对业务代码不友好,chat_models/base.py 做了完整映射,让你用except精准分层处理:
| 异常类 | 继承关系 | 触发场景 |
|---|---|---|
OpenAIContextOverflowError | openai.BadRequestError+ContextOverflowError | 上下文超长 |
OpenAIAuthenticationError | openai.AuthenticationError+ModelAuthenticationError | 密钥无效 / 认证失败 |
OpenAIPermissionDeniedError | 权限拒绝类 | 无访问权限 |
OpenAIInvalidRequestError | openai.BadRequestError+ModelInvalidRequestError | 请求体非法 |
OpenAIModelNotFoundError | openai.NotFoundError+ModelNotFoundError | 模型名不存在 |
OpenAIRateLimitError | openai.RateLimitError+ModelRateLimitError | 触发限流 |
OpenAIAPIError | openai.InternalServerError+ModelAPIError | 服务端 5xx |
OpenAIConnectionError | openai.APIConnectionError+ModelConnectionError | 连接失败 |
OpenAITimeoutError | openai.APITimeoutError+ModelTimeoutError | 请求超时 |
统一捕获策略建议:
OpenAIRateLimitError走指数退避重试;OpenAIContextOverflowError走截断或压缩上下文;OpenAIConnectionError/OpenAITimeoutError交给max_retries自动兜底。
兼容服务:base_url 与 extra_body 的分工
LM Studio、vLLM、Ollama 这类 OpenAI 兼容服务,只需要换base_url就能接进来,但参数放置位置有讲究:
# LM Studio:支持 TTL 自动卸载 model = ChatOpenAI( base_url="http://localhost:1234/v1", api_key="lm-studio", # 任意非空字符串即可 model="mlx-community/QwQ-32B-4bit", extra_body={"ttl": 300}, ) # vLLM:自定义采样参数 model = ChatOpenAI( base_url="http://localhost:8000/v1", api_key="EMPTY", model="meta-llama/Llama-2-7b-chat-hf", extra_body={"use_beam_search": True, "best_of": 4}, )参数路由规则(源码 docstring 明确约定):
| 参数容器 | 落点 | 典型用途 |
|---|---|---|
model_kwargs | 合并进请求体顶层 | 标准 OpenAI 参数:max_completion_tokens、stream_options、modalities、audio |
extra_body | 嵌套在extra_body键下 | 兼容服务商私有参数:vLLM 的use_beam_search、LM Studio 的ttl |
两条硬性提醒:私有参数误放进
model_kwargs会直接触发 API 报错;同时模型名可能触发 Responses API 路由(/v1/chat/completions与/v1/responses的选择部分依据模型名推断,与base_url无关),对接兼容服务请显式设use_responses_api=False。另外ChatOpenAI只遵循官方 API 规范,第三方响应里的非标准字段(如reasoning_content)不会被提取保留,接 DeepSeek、OpenRouter 建议用各自的专属包。
工程化:测试、快照与静态检查
单测集中在 tests/unit_tests,覆盖chat_models、embeddings、llms、token 计数、兼容层与密钥加载;集成测试放在 tests/integration_tests,用vcrpy录制回放真实 API 交互,避免重复计费。pyproject.toml 中的质量门槛:pytest 默认带--snapshot-warn-unused --strict-markers --cov=langchain_openai;ruff 开启select = ["ALL"]并采用 Google 风格 docstring;mypy 设了disallow_untyped_defs = true。跨包一致性则由 libs/standard-tests/ 的标准测试套件统一校验。
向量化与遗留补全:补齐剩下的两块拼图
OpenAIEmbeddings 的分块加权机制
默认模型是text-embedding-ada-002,text-embedding-3系列可以加dimensions指定输出维度:
from langchain_openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings(model="text-embedding-3-large", dimensions=1024) vec = embeddings.embed_query("hello") vecs = embeddings.embed_documents(["hello", "goodbye"])关键默认值一览(定义见 embeddings/base.py):
| 参数 | 默认值 | 说明 |
|---|---|---|
embedding_ctx_length | 8191 | 单条文本嵌入前允许的最大 token 数 |
chunk_size | 1000 | 单批请求的文本条数 |
max_retries | 2 | 最大重试次数 |
request_timeout | None | float /httpx.Timeout/ None |
tiktoken_enabled | True | 本地分词计数;False时改用 transformers 分词 |
tiktoken_model_name | None | 指定计费用 tiktoken 模型(Azure / 兼容服务适用) |
show_progress_bar | False | 批量嵌入时显示进度条 |
skip_empty | False | 是否跳过空字符串 |
它的长文本策略可以类比"把长文拆成几段分别打分,再按篇幅加权出总分":超过embedding_ctx_length的文本会被切块分别嵌入,最后按各块 token 数加权平均并归一化,逻辑在_process_batched_chunked_embeddings里。对 OpenRouter、Ollama、vLLM 这类兼容服务,很多实现不支持本地分词,建议check_embedding_ctx_length=False直接发原文,必要时再加encoding_format="float"绕开 base64 编码问题。
传统补全:OpenAI 类
非聊天的 instruct 风格补全模型走 llms/base.py 的OpenAI类(继承BaseLLM),接口形态与ChatOpenAI类似但输入是纯字符串:
from langchain_openai import OpenAI llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_retries=2) print(llm.invoke("The meaning of life is ")) for piece in llm.stream("The meaning of life is "): print(piece, end="|")选型落点:聊天、工具、结构化输出、Responses API 一律用
ChatOpenAI;只有维护 instruct 系列老模型时才用OpenAI;向量入库用OpenAIEmbeddings;企业 Azure 环境把三个类分别换成Azure前缀版本,其余代码基本不动。
【免费下载链接】langchainThe agent engineering platform.项目地址: https://gitcode.com/GitHub_Trending/la/langchain
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考