feat(pipeline/llm): OpenAI 官方 ChatGPT 适配器并整理 providers 目录

新增 providers/adapters/openai_official_chatgpt、shared 抽离归一化与 token 启发式；MA_LLM_TEXT_PROVIDER 支持 openai_official/openai_chatgpt/chatgpt。补充单测与 try_openai_official_llm 试跑脚本、.env.example 说明。

Made-with: Cursor

.env.example

@@ -29,9 +29,19 @@ CSRF_TRUSTED_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
 # OPENAI_TEXT_MODEL=
 # 别名：LLM_API_KEY、LLM_BASE_URL、LLM_MODEL
 
-# 报告/策略等「纯文本」任务的后端选择（默认经 crawler 副本内 AI_crawler 调 OpenAI 兼容网关；其它取值待扩展）
+# 报告/策略等「纯文本」任务的后端选择：
+#   crawler_openai_compatible（默认）= 经 crawler 副本内 AI_crawler 调自建 OpenAI 兼容网关
+#   openai_official / openai_chatgpt / chatgpt = 直连 OpenAI 官方 api.openai.com（与上项凭据独立）
 # MA_LLM_TEXT_PROVIDER=crawler_openai_compatible
 
+# OpenAI 官方 Chat Completions（仅当 MA_LLM_TEXT_PROVIDER 为 openai_official / openai_chatgpt / chatgpt 时需要）
+# OPENAI_OFFICIAL_API_KEY=sk-...
+# OPENAI_OFFICIAL_BASE_URL=https://api.openai.com/v1
+# OPENAI_OFFICIAL_TEXT_MODEL=gpt-4o-mini
+# 与预检一致；gpt-4o 等可设 128000
+# OPENAI_OFFICIAL_CONTEXT_WINDOW=128000
+# OPENAI_OFFICIAL_TIMEOUT=600
+
 # 独立策略稿 / 报告内「策略与机会」LLM 块采样温度（默认 0.1，低于 chat_completion_text 的 0.2，减轻同提示多轮漂移）。设 0 更稳、设 0.2 与旧默认接近。
 # MA_STRATEGY_LLM_TEMPERATURE=0.1
 

backend/pipeline/demos/try_openai_official_llm.py

@@ -0,0 +1,27 @@
+"""
+试跑 OpenAI 官方「ChatGPT」文本适配器（不经过完整报告管线）。
+
+准备：在 .env 中设置 MA_LLM_TEXT_PROVIDER=chatgpt 与 OPENAI_OFFICIAL_API_KEY（见 .env.example）。
+
+用法（在 backend 目录下）：
+
+    .venv\\Scripts\\python.exe -m pipeline.demos.try_openai_official_llm
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+if __name__ == "__main__":
+    if not (os.environ.get("OPENAI_OFFICIAL_API_KEY") or "").strip():
+        print("未设置 OPENAI_OFFICIAL_API_KEY，退出。", file=sys.stderr)
+        sys.exit(1)
+    # 与生产一致时可在 .env 中设 MA_LLM_TEXT_PROVIDER=chatgpt；本脚本也强制用官方适配器
+    from pipeline.llm.providers.adapters.openai_official_chatgpt import OpenAiOfficialChatGptTextLlm
+
+    out = OpenAiOfficialChatGptTextLlm().complete_text(
+        "You reply in one short English sentence only.",
+        "What is 2+2?",
+        temperature=0.0,
+    )
+    print(out)

backend/pipeline/llm/llm_client.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from .providers.factory import get_text_llm
-from .providers.output_normalize import strip_outer_markdown_fence
+from .providers.shared.output_normalize import strip_outer_markdown_fence
 
 
 def call_llm(

backend/pipeline/llm/providers/__init__.py

@@ -1,10 +1,14 @@
 """文本大模型调用的协议、适配器与工厂（与具体提示词/业务生成逻辑解耦）。"""
+
 from __future__ import annotations
 
+from .adapters import CrawlerOpenAiCompatibleTextLlm, OpenAiOfficialChatGptTextLlm
 from .factory import get_text_llm, reset_text_llm_client_for_tests
 from .protocol import TextLlmClient
 
 __all__ = [
+    "CrawlerOpenAiCompatibleTextLlm",
+    "OpenAiOfficialChatGptTextLlm",
     "TextLlmClient",
     "get_text_llm",
     "reset_text_llm_client_for_tests",

backend/pipeline/llm/providers/adapters/__init__.py

@@ -0,0 +1,11 @@
+"""具体大模型通道实现：经统一协议暴露给 `factory`。"""
+from __future__ import annotations
+
+from .crawler_openai_compatible import CrawlerOpenAiCompatibleTextLlm, ensure_ai_crawler_path
+from .openai_official_chatgpt import OpenAiOfficialChatGptTextLlm
+
+__all__ = [
+    "CrawlerOpenAiCompatibleTextLlm",
+    "OpenAiOfficialChatGptTextLlm",
+    "ensure_ai_crawler_path",
+]

backend/pipeline/llm/providers/adapters/crawler_openai_compatible.py

@@ -9,7 +9,7 @@ from pathlib import Path
 
 from django.conf import settings
 
-from .token_heuristics import estimate_crawler_style_input_tokens
+from ..shared.token_heuristics import estimate_crawler_style_input_tokens
 
 
 def ensure_ai_crawler_path() -> None:

backend/pipeline/llm/providers/adapters/openai_official_chatgpt.py

@@ -0,0 +1,133 @@
+"""
+OpenAI 官方 `https://api.openai.com`（ChatGPT 系列）`chat/completions` 直连接口。
+
+**凭据与网关与爬虫副本中的自建网关独立**，避免与 `OPENAI_BASE_URL` 指向的兼容网关共用时互相串环境。
+通过 `MA_LLM_TEXT_PROVIDER=openai_official`（或 `openai_chatgpt` / `chatgpt`）启用。
+"""
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import requests
+
+from ..shared.openai_message_content import normalize_message_content
+from ..shared.token_heuristics import estimate_crawler_style_input_tokens
+
+_DEFAULT_BASE = "https://api.openai.com/v1"
+_DEFAULT_MODEL = "gpt-4o-mini"
+# gpt-4o / 4.1 等常见上限；可按模型在 .env 中覆盖
+_DEFAULT_CTX = 128_000
+
+_BUF = 256
+_WANT_MAX = 8192
+
+
+def _read_timeout() -> tuple[float, float]:
+    read = 600
+    raw = (os.environ.get("OPENAI_OFFICIAL_TIMEOUT") or os.environ.get("LLM_CHAT_TIMEOUT") or os.environ.get("OPENAI_TIMEOUT") or "").strip()
+    if raw:
+        try:
+            read = max(60, int(raw))
+        except ValueError:
+            pass
+    conn = 30.0
+    raw_c = (os.environ.get("LLM_CHAT_CONNECT_TIMEOUT") or "").strip()
+    if raw_c:
+        try:
+            conn = max(5.0, float(raw_c))
+        except ValueError:
+            pass
+    return (conn, float(read))
+
+
+def _resolve_credentials() -> tuple[str, str, str]:
+    key = (os.environ.get("OPENAI_OFFICIAL_API_KEY") or "").strip()
+    if not key:
+        msg = "使用 openai_official 适配器需设置环境变量 OPENAI_OFFICIAL_API_KEY（与自建网关/爬虫副本的 key 可分开）。"
+        raise ValueError(msg)
+    base = (os.environ.get("OPENAI_OFFICIAL_BASE_URL") or _DEFAULT_BASE).strip().rstrip("/")
+    model = (
+        os.environ.get("OPENAI_OFFICIAL_TEXT_MODEL")
+        or os.environ.get("OPENAI_OFFICIAL_MODEL")
+        or _DEFAULT_MODEL
+    ).strip()
+    return key, base, model
+
+
+def _context_window() -> int:
+    raw = (os.environ.get("OPENAI_OFFICIAL_CONTEXT_WINDOW") or str(_DEFAULT_CTX)).strip()
+    try:
+        return max(4096, int(raw))
+    except ValueError:
+        return _DEFAULT_CTX
+
+
+def _default_temperature() -> float:
+    return 0.2
+
+
+class OpenAiOfficialChatGptTextLlm:
+    """
+    直连 OpenAI 官方「Chat Completions」；请求体与 `AI_crawler.chat_completion_text` 同形，
+    并在本地做与爬虫网关一致的 `max_tokens` 收紧，减少 400。
+    """
+
+    def complete_text(
+        self,
+        system_prompt: str,
+        user_prompt: str,
+        *,
+        temperature: float | None = None,
+    ) -> str:
+        api_key, base, model = _resolve_credentials()
+        body: dict[str, Any] = {
+            "model": model,
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt},
+            ],
+            "temperature": _default_temperature() if temperature is None else float(temperature),
+            "max_tokens": _WANT_MAX,
+        }
+        est = estimate_crawler_style_input_tokens(system_prompt, user_prompt)
+        context_window = _context_window()
+        if est >= context_window - _BUF - 256:
+            raise ValueError(
+                f"提示词过长（估算输入约 {est} tokens，OPENAI_OFFICIAL_CONTEXT_WINDOW={context_window}），"
+                "请缩小输入或调大 OPENAI_OFFICIAL_CONTEXT_WINDOW。"
+            )
+        avail = context_window - est - _BUF
+        want = int(body.get("max_tokens") or _WANT_MAX)
+        body["max_tokens"] = max(256, min(want, max(avail, 256)))
+        r = requests.post(
+            f"{base}/chat/completions",
+            headers={
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json",
+            },
+            json=body,
+            timeout=_read_timeout(),
+        )
+        try:
+            r.raise_for_status()
+        except requests.HTTPError as e:
+            snippet = ""
+            if e.response is not None:
+                snippet = (e.response.text or "")[:1200].replace("\r\n", "\n").replace("\n", " ")
+            if snippet:
+                raise requests.HTTPError(
+                    f"{e!s} | body: {snippet}",
+                    response=e.response,
+                    request=e.request,
+                ) from e
+            raise
+        data = r.json()
+        msg = (data.get("choices") or [{}])[0].get("message") or {}
+        return normalize_message_content(msg.get("content"))
+
+    def estimate_input_tokens(self, system_prompt: str, user_prompt: str) -> int:
+        return estimate_crawler_style_input_tokens(system_prompt, user_prompt)
+
+    def context_window_tokens(self) -> int:
+        return _context_window()

backend/pipeline/llm/providers/factory.py

@@ -5,7 +5,8 @@ from __future__ import annotations
 
 import os
 
-from .crawler_openai_compatible import CrawlerOpenAiCompatibleTextLlm
+from .adapters.crawler_openai_compatible import CrawlerOpenAiCompatibleTextLlm
+from .adapters.openai_official_chatgpt import OpenAiOfficialChatGptTextLlm
 from .protocol import TextLlmClient
 
 # 模块级单例：避免重复构造；测试可用 `reset_text_llm_client_for_tests` 切换实现。
@@ -29,9 +30,14 @@ def _build_client(pid: str) -> TextLlmClient:
         "openai_compatible",
     ):
         return CrawlerOpenAiCompatibleTextLlm()
+    if pid in ("openai_official", "openai_chatgpt", "chatgpt"):
+        return OpenAiOfficialChatGptTextLlm()
+    known = (
+        "crawler_openai_compatible, crawler, default, openai_compatible, "
+        "openai_official, openai_chatgpt, chatgpt"
+    )
     raise ValueError(
-        f"不支持的 {_ENV_KEY}={pid!r}；"
-        f"当前仅实现 {_DEFAULT_ID}（经 AI_crawler 的 OpenAI 兼容 `chat/completions`）。"
+        f"不支持的 {_ENV_KEY}={pid!r}；已知取值：{known}。",
     )
 
 

backend/pipeline/llm/providers/shared/__init__.py

@@ -0,0 +1,12 @@
+"""与具体后端无关的轻量工具：去围栏、token 启发式、OpenAI 风格 message 正文解析。"""
+from __future__ import annotations
+
+from .openai_message_content import normalize_message_content
+from .output_normalize import strip_outer_markdown_fence
+from .token_heuristics import estimate_crawler_style_input_tokens
+
+__all__ = [
+    "estimate_crawler_style_input_tokens",
+    "normalize_message_content",
+    "strip_outer_markdown_fence",
+]

backend/pipeline/llm/providers/shared/openai_message_content.py

@@ -0,0 +1,26 @@
+"""
+解析 OpenAI chat.completions 返回的 `message.content`：可能是 str 或 part 列表。
+与 `AI_crawler._normalize_chat_content` 行为一致。
+"""
+from __future__ import annotations
+
+from typing import Any
+
+
+def normalize_message_content(content: Any) -> str:
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content.strip()
+    if isinstance(content, list):
+        parts: list[str] = []
+        for item in content:
+            if isinstance(item, dict):
+                if item.get("type") == "text":
+                    parts.append(str(item.get("text") or ""))
+                elif "text" in item:
+                    parts.append(str(item.get("text") or ""))
+            elif isinstance(item, str):
+                parts.append(item)
+        return "".join(parts).strip()
+    return str(content).strip()

backend/pipeline/llm/providers/shared/token_heuristics.py

@@ -1,6 +1,6 @@
 """
 与 `crawler_copy/.../AI_crawler` 中 `_estimate_chat_input_tokens` 同口径的保守估算，
-供预检、策略档位与 OpenAI 兼容适配器共用（无 tiktoken 时避免 max_tokens 400）。
+供预检、策略档位与各适配器共用（无 tiktoken 时避免 max_tokens 400）。
 """
 from __future__ import annotations
 

backend/pipeline/tests/test_text_llm_providers.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from pipeline.llm.providers import (
+    CrawlerOpenAiCompatibleTextLlm,
+    OpenAiOfficialChatGptTextLlm,
+    get_text_llm,
+    reset_text_llm_client_for_tests,
+)
+
+
+def test_openai_official_complete_text_uses_post_and_returns_content(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENAI_OFFICIAL_API_KEY", "sk-test")
+
+    def _fake_post(
+        url: str,
+        headers: dict,
+        json: dict,
+        timeout: object,
+    ) -> MagicMock:
+        assert "/chat/completions" in url
+        assert json["messages"][0]["role"] == "system"
+        r = MagicMock()
+        r.json.return_value = {"choices": [{"message": {"content": "ok_out"}}]}
+        r.raise_for_status = MagicMock()
+        return r
+
+    with patch(
+        "pipeline.llm.providers.adapters.openai_official_chatgpt.requests.post",
+        side_effect=_fake_post,
+    ):
+        llm = OpenAiOfficialChatGptTextLlm()
+        out = llm.complete_text("S", "U", temperature=0.1)
+    assert out == "ok_out"
+
+
+def test_factory_selects_openai_official_when_env_set(monkeypatch: pytest.MonkeyPatch) -> None:
+    reset_text_llm_client_for_tests()
+    monkeypatch.setenv("MA_LLM_TEXT_PROVIDER", "chatgpt")
+    monkeypatch.setenv("OPENAI_OFFICIAL_API_KEY", "sk-x")
+    try:
+        c = get_text_llm()
+        assert isinstance(c, OpenAiOfficialChatGptTextLlm)
+    finally:
+        reset_text_llm_client_for_tests()
+        monkeypatch.delenv("MA_LLM_TEXT_PROVIDER", raising=False)
+        monkeypatch.delenv("OPENAI_OFFICIAL_API_KEY", raising=False)
+
+
+def test_factory_unknown_provider_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    reset_text_llm_client_for_tests()
+    monkeypatch.setenv("MA_LLM_TEXT_PROVIDER", "no_such_provider")
+    try:
+        with pytest.raises(ValueError, match="不支持的"):
+            get_text_llm()
+    finally:
+        reset_text_llm_client_for_tests()
+        monkeypatch.delenv("MA_LLM_TEXT_PROVIDER", raising=False)
+
+
+def test_default_provider_is_crawler_when_env_cleared(monkeypatch: pytest.MonkeyPatch) -> None:
+    """未设置 MA_LLM_TEXT_PROVIDER 时仍为爬虫副本网关；此处只断言类型，不调真实网络。"""
+    reset_text_llm_client_for_tests()
+    monkeypatch.delenv("MA_LLM_TEXT_PROVIDER", raising=False)
+    c = get_text_llm()
+    assert isinstance(c, CrawlerOpenAiCompatibleTextLlm)
+    reset_text_llm_client_for_tests()