Python集成OpenAI与百度文心一言API的代码示例

在实际的AI应用开发中，很多系统需要同时接入国际主流的OpenAI模型与国内合规的百度文心一言大模型，以兼顾推理能力、网络连通性与合规审计要求。本文通过具体的Python代码，依次演示OpenAI官方接口与百度千帆大模型平台（文心一言）接口的鉴权通信流程，并给出一套统一的调用抽象层实现，帮助开发者在一个项目中无缝切换与调度两种API。

一、集成准备与依赖环境安装

在编写代码前，需要分别完成两家平台的凭据获取与本地Python运行环境配置：

• OpenAI凭据：前往OpenAI开发者平台创建并复制API Key。在有代理或反向代理需求的环境下，还可配置自定义base_url。
• 百度千帆凭据：登录百度智能云千帆大模型平台，在“应用接入”页面创建应用，获取对应应用的API Key (AK)和Secret Key (SK)。

在终端中执行以下命令安装官方支持包与网络请求库：

pip install openai qianfan requests

二、OpenAI API 的标准调用实现

在OpenAI Python SDK 1.0版本及以上中，采用全新的OpenAI客户端对象进行调用。以下代码展示初始化、构造对话请求及提取回复结果的标准过程：

import os
from openai import OpenAI

# 1. 客户端初始化：从环境变量读取 OPENAI_API_KEY
client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
    # 如使用代理或内网网关，可指定 base_url="https://api.openai-proxy.com/v1"
)

def call_openai_chat(prompt: str, model: str = "gpt-4o-mini") -> str:
    """
    调用 OpenAI Chat Completions 接口
    """
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "你是一个严谨且实用的技术助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7,
            max_tokens=1000
        )
        # 获取回复文本
        content = response.choices[0].message.content
        return content.strip()
    except Exception as e:
        print(f"[OpenAI 调用异常]: {e}")
        raise e

if __name__ == "__main__":
    test_prompt = "请用一句话解释什么是RESTful API。"
    result = call_openai_chat(test_prompt)
    print("OpenAI 回复结果：\n", result)

三、百度文心一言 API 的调用实现

百度文心一言由百度千帆大模型平台托管。集成百度文心一言通常有两种路径：使用千帆官方SDK，或使用轻量级requests直接调用原生REST接口。生产环境中，了解原生REST通信有助于在没有第三方依赖或自研网关时排查鉴权问题。

方式一：使用百度千帆官方SDK调用

官方SDK内置了Token缓存维护与自动续期，是使用百度模型最直接的方式：

import os
import qianfan

# 显式配置凭据，或提前在系统中导出 QIANFAN_AK 与 QIANFAN_SK 环境变量
os.environ["QIANFAN_AK"] = os.environ.get("QIANFAN_AK", "YOUR_BAIDU_AK")
os.environ["QIANFAN_SK"] = os.environ.get("QIANFAN_SK", "YOUR_BAIDU_SK")

def call_wenxin_sdk(prompt: str, model: str = "ERNIE-Speed-128K") -> str:
    """
    使用百度千帆 Python SDK 调用文心一言模型
    """
    try:
        # 初始化对话对象
        chat_comp = qianfan.ChatCompletion()
        
        # 发送对话请求
        resp = chat_comp.do(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7
        )
        return resp["result"].strip()
    except Exception as e:
        print(f"[千帆SDK 调用异常]: {e}")
        raise e

方式二：使用原生 requests 请求百度 REST 接口

如果项目需要保持轻量，不愿引入额外SDK，可以通过以下两步完成：先用AK和SK向鉴权服务换取有效时长30天的access_token，再将Token附加在请求参数中调用文心模型端点：

import os
import requests

def get_baidu_access_token(ak: str, sk: str) -> str:
    """
    获取百度千帆鉴权 Token (通常有效期为30天，可加内存缓存)
    """
    url = f"https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id={ak}&client_secret={sk}"
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    
    response = requests.post(url, headers=headers, timeout=10)
    response.raise_for_status()
    data = response.json()
    if "access_token" in data:
        return data["access_token"]
    raise RuntimeError(f"获取百度Token失败: {data}")

def call_wenxin_rest(prompt: str, ak: str, sk: str) -> str:
    """
    通过原生 HTTP POST 请求调用文心一言 (以 ERNIE-Speed-128K 为例)
    """
    token = get_baidu_access_token(ak, sk)
    # ERNIE-Speed-128K 对应官方发布端点
    endpoint = f"https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/ernie-speed-128k?access_token={token}"
    
    payload = {
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7
    }
    headers = {"Content-Type": "application/json"}
    
    resp = requests.post(endpoint, json=payload, headers=headers, timeout=30)
    resp.raise_for_status()
    res_data = resp.json()
    
    # 百度接口如果发生业务错误，HTTP可能返回200但在体中附带 error_code
    if "error_code" in res_data:
        raise RuntimeError(f"文心一言服务返回错误 [{res_data['error_code']}]: {res_data.get('error_msg')}")
        
    return res_data.get("result", "").strip()

四、编写统一适配抽象层 (Unified LLM Client)

在实际工程中，为了避免业务代码与具体的模型厂商强耦合，通常需要构建一个统一接口。该适配层负责：

• 统一输入：仅接收标准字符串或标准消息结构。
• 统一输出：统一返回过滤后的回答文本，隐去choices[0].message.content与result的数据字段差异。
• 统一错误转换：对外抛出标准自定义异常，方便上层做重试或降级。

import os
from typing import Optional
from openai import OpenAI
import qianfan

class UnifiedLLMClient:
    """
    统一大语言模型适配客户端，支持在 OpenAI 与百度文心一言间自由切换
    """
    def __init__(
        self,
        openai_key: Optional[str] = None,
        qianfan_ak: Optional[str] = None,
        qianfan_sk: Optional[str] = None
    ):
        # 1. 注册 OpenAI 客户端
        self.openai_key = openai_key or os.environ.get("OPENAI_API_KEY")
        self.openai_client = OpenAI(api_key=self.openai_key) if self.openai_key else None
        
        # 2. 注册 百度千帆 凭据
        self.qianfan_ak = qianfan_ak or os.environ.get("QIANFAN_AK")
        self.qianfan_sk = qianfan_sk or os.environ.get("QIANFAN_SK")
        if self.qianfan_ak and self.qianfan_sk:
            os.environ["QIANFAN_AK"] = self.qianfan_ak
            os.environ["QIANFAN_SK"] = self.qianfan_sk
            self.qianfan_client = qianfan.ChatCompletion()
        else:
            self.qianfan_client = None

    def generate(self, prompt: str, provider: str = "openai", model: Optional[str] = None) -> str:
        """
        统一调用入口：
        provider 选项: 'openai' 或 'wenxin'
        """
        provider = provider.lower()
        if provider == "openai":
            if not self.openai_client:
                raise ValueError("未检测到有效 OPENAI_API_KEY，无法调用 OpenAI 接口。")
            target_model = model or "gpt-4o-mini"
            resp = self.openai_client.chat.completions.create(
                model=target_model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.7
            )
            return resp.choices[0].message.content.strip()

        elif provider in ["wenxin", "qianfan", "baidu"]:
            if not self.qianfan_client:
                raise ValueError("未检测到有效 QIANFAN_AK/SK，无法调用百度文心接口。")
            target_model = model or "ERNIE-Speed-128K"
            resp = self.qianfan_client.do(
                model=target_model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.7
            )
            return resp["result"].strip()

        else:
            raise ValueError(f"不支持的模型提供商: {provider}。目前仅支持 'openai' 与 'wenxin'")

# 示例验证
if __name__ == "__main__":
    client = UnifiedLLMClient()
    question = "请用20字以内概括数据结构中栈的特点。"

    # 1. 统一接口调度 OpenAI
    # answer_openai = client.generate(question, provider="openai", model="gpt-4o-mini")
    # print("OpenAI:", answer_openai)

    # 2. 统一接口调度 百度文心一言
    # answer_baidu = client.generate(question, provider="wenxin", model="ERNIE-Speed-128K")
    # print("百度文心:", answer_baidu)

五、注意事项与生产实践建议

• 错误码截获差异：OpenAI发生限流或鉴权失败时，HTTP状态码会直接返回401或429，Python中会抛出openai.RateLimitError或openai.AuthenticationError；而百度千帆直接调用HTTP接口时，网络状态码可能依然为200，需要在返回的JSON字典中显式检查error_code（例如17代表并发超限，18代表配额用尽）。
• Token生命周期管理：如果采用纯HTTP方式调用百度文心，换取的access_token有效期为30天，应在应用内存或Redis中缓存该Token并在过期前重新获取，避免每次请求都额外产生一次OAuth往返。
• 流式响应支持：当需要打字机输出效果时，两家均支持在入参中设置stream=True。OpenAI返回的是分块Chunk生成器，通过chunk.choices[0].delta.content提取；千帆SDK同样返回生成器，通过遍历resp["result"]逐步输出。