时间：   |   分类：     | 字数： 866 字 | 阅读： 2分钟 | 阅读次数：

自定义 LangChain OpenAI 聊天模型（CustomChatOpenAI）重构实践

在 AI 应用开发中，LangChain 提供了强大的链式调用能力，而 OpenAI 的接口则是主流大模型服务的事实标准。本文将介绍如何基于 LangChain 的 ChatOpenAI，重构并扩展出一个支持推理内容（reasoning_content）流式输出的自定义聊天模型——CustomChatOpenAI，并详细解析其设计思路与实现细节。

背景与需求

在实际业务中，除了常规的对话内容（content），我们还希望模型能输出推理过程（reasoning_content），并且在流式响应时能以 <think>...</think> 标签包裹推理内容，便于前端或下游系统做进一步处理。原生的 ChatOpenAI 并不支持这一需求，因此需要自定义扩展。

设计思路

1. 继承与扩展

我们通过继承 ChatOpenAI，重写其核心方法，增加对 reasoning_content 的处理能力。核心思路如下：

  • 流式输出：在流式响应中，优先检测 reasoning_content，并用 <think>...</think> 标签包裹，仅允许出现一次，避免重复嵌套。
  • 普通输出：在非流式响应中，将推理内容和最终回复拼接输出，格式统一。

2. 关键方法解析

_create_client

重写客户端创建方法，支持自定义 base_url 和 api_key，兼容多种 OpenAI 兼容服务。

_process_stream

核心流式处理逻辑。遍历大模型返回的流式数据块，判断 delta 中是否包含 reasoning_content 或 content，并按需插入 <think> 标签。例如：

// ... existing code ...
if hasattr(delta, 'reasoning_content') and delta.reasoning_content:
    if not in_think_tag and not think_closed:
        yield AIMessageChunk(content="<think>\n")
        in_think_tag = True
    if in_think_tag and not think_closed:
        yield AIMessageChunk(content=delta.reasoning_content)
elif hasattr(delta, 'content') and delta.content:
    if in_think_tag and not think_closed:
        yield AIMessageChunk(content="\n</think>\n")
        in_think_tag = False
        think_closed = True
    yield AIMessageChunk(content=delta.content)
// ... existing code ...

_generate

非流式响应时，将 reasoning_content 和 content 拼接，保证输出格式一致。

stream 与 invoke

  • stream 方法用于流式输出，适配 LangChain 的消息格式。
  • invoke 方法根据 streaming 参数自动选择流式或非流式处理，提升接口易用性。

实践效果

通过 test.py 中的测试用例，可以看到 CustomChatOpenAI 能够正确输出带 <think> 标签的推理内容，并兼容原有的对话内容输出，极大提升了模型输出的可解释性和可用性。

总结与展望

本次重构不仅满足了业务对推理内容的特殊需求，还为后续扩展（如多标签、多段推理等）打下了良好基础。未来可以考虑：

  • 支持多段推理内容的分段输出
  • 推理内容与最终回复的样式自定义
  • 更丰富的流式消息处理能力

如需完整代码实现，请参考项 https://raw.githubusercontent.com/luckfu/chat/refs/heads/main/custom_chat.py 文件。欢迎交流与指正！

0%