from fastapi.datastructures import URL
import requests
import json
from typing import List, Dict, Optional

3m7of64lb 文本生成

https://cloud.baidu.com/doc/qianfan/s/rmh4stp0j 模型列表

class Qianfan:

"""
百度千帆大模型
接口地址：https://qianfan.baidubce.com/v2/chat/completions
认证方式：Bearer bce-v3签名
"""

def __init__(self, bearer_token: str):
    self.base_url = "https://qianfan.baidubce.com/v2/chat/completions"
    self.headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {bearer_token}",  # 完全匹配curl认证头
    }

def send_chat(
    self,
    messages,
    model: str = "ernie-3.5-8k",
    temperature: float = 0.7,
    top_p: float = 0.95,
    max_tokens: int = 2048,  # 平台限制最大值2048
    stream: bool = False,
):
    """
    调用千帆V2对话接口
    :param messages: 对话消息列表，格式和OpenAI一致
    :param model: 模型名称，默认ernie-3.5-8k
    :param temperature: 生成温度
    :param top_p: 采样阈值
    :param max_tokens: 最大生成token数（限制1-2048）
    :param stream: 是否流式输出
    :return: 接口响应结果
    """

    if not (1 <= max_tokens <= 2048):
        raise ValueError(f"max_tokens必须在[1, 2048]之间，当前值：{max_tokens}")

    # 构造请求体
    data = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "top_p": top_p,
        "max_tokens": max_tokens,
        "stream": stream,
    }

    try:
        if not stream:
            response = requests.post(
                url=self.base_url,
                headers=self.headers,
                json=data,
                timeout=30,  # 设置超时时间，避免卡顿
            )
            response.raise_for_status()  # 抛出HTTP异常（如401/403/500等）
            return response.json()
    except requests.exceptions.HTTPError as e:
        # 捕获HTTP错误，返回详细信息
        error_detail = f"HTTP错误 {e.response.status_code}: {e.response.text}"
        raise Exception(error_detail)
    except requests.exceptions.Timeout:
        raise Exception("接口调用超时（超过30秒），请检查网络或接口状态")
    except requests.exceptions.ConnectionError:
        raise Exception("网络连接失败，请检查网络或接口地址是否正确")
    except Exception as e:
        raise Exception(f"接口调用失败: {str(e)}")

# 提取有用信息
def get_answer(self, messages: List[Dict[str, str]], **kwargs) -> str:
    """
    简化调用：直接返回模型的回答内容（封装chat_completion）
    :param messages: 对话消息列表
    :param kwargs: 其他参数（model/temperature/max_tokens等）
    :return: 模型回答的文本内容
    """
    response = self.chat_completion(messages=messages, **kwargs)
    # 提取核心回答内容
    if "choices" in response and len(response["choices"]) > 0:
        return response["choices"][0]["message"]["content"]
    else:
        raise Exception(
            f"响应格式异常，无法提取回答：{json.dumps(response, ensure_ascii=False)}"
        )

--------------------- 测试/使用示例 ---------------------

if name == "__main__":

# 1. 替换为你自己的完整Bearer token（从curl命令中复制）
BEARER_TOKEN = (
    "bce-v3/ALTAK-AFJjxQqQDcYb4msG4P16P/317ea76939b7*****5ea88eae6fe1745ab3ecac5"
)
print("=== 方式1：基础调用 ===")
try:
    # 创建客户端实例
    client = Qianfan(bearer_token=BEARER_TOKEN)

    messages = [
        {"role": "user", "content": "介绍一下你自己"},
    ]

    # 调用接口（可自定义参数）
    response = client.send_chat(
        messages=messages,
        model="ernie-3.5-8k",
        temperature=0.7,
        max_tokens=1024,  # 自定义token数（1-2048）
        stream=False,
    )
    print(json.dumps(response, ensure_ascii=False, indent=2))
    # answer = client.get_answer(messages=messages)

except Exception as e:
    print(f"调用失败: {str(e)}")