百度文心一言API调用ERNIE-3.5-8K的Python开发指南

1. 环境准备与API接入基础

1.1 前置条件

在调用百度文心一言API前，需完成以下准备工作：

• 注册百度智能云账号并完成企业实名认证
• 开通「文心一言」服务并创建应用
• 获取API Key和Secret Key（保管好密钥，建议通过环境变量配置）

1.2 安装Python SDK

推荐使用官方aip库简化调用流程：

pip install baidu-aip

或直接通过HTTP请求调用：

pip install requests

2. 认证与鉴权机制

2.1 Access Token获取

ERNIE-3.5-8K的API调用采用OAuth2.0认证，需先获取access_token：

import requests
API_KEY = 'your_api_key'
SECRET_KEY = 'your_secret_key'
def get_access_token():
    url = f"https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id={API_KEY}&client_secret={SECRET_KEY}"
    response = requests.post(url)
    return response.json().get('access_token')

2.2 Token缓存策略

建议实现token缓存机制（有效期30天）：

from datetime import datetime, timedelta
class TokenManager:
    def __init__(self):
        self._token = None
        self._expire_time = datetime.now()
    def get_token(self):
        if not self._token or datetime.now() > self._expire_time:
            self._refresh_token()
        return self._token
    def _refresh_token(self):
        self._token = get_access_token()
        self._expire_time = datetime.now() + timedelta(days=25)  # 提前5天更新

3. ERNIE-3.5-8K核心API调用

3.1 基础文本生成

def ernie_35_8k(prompt, temperature=0.7, top_p=0.9):
    url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token_manager.get_token()}"
    }
    payload = {
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "top_p": top_p,
        "model": "ERNIE-3.5-8K"
    }
    response = requests.post(url, headers=headers, json=payload)
    return response.json()

3.2 高级参数说明

参数	类型	说明
max_output_tokens	int	最大输出token数（8K版本上限8000）
presence_penalty	float	重复惩罚系数（-2.0~2.0）
stop	list	停止生成标识词列表

4. 流式输出处理

对于长文本生成，建议使用流式接口减少等待时间：

def stream_ernie(prompt):
    url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions_pro"
    payload = {
        "messages": [{"role": "user", "content": prompt}],
        "stream": True
    }
    with requests.post(url, headers=headers, json=payload, stream=True) as r:
        for chunk in r.iter_content(chunk_size=None):
            if chunk:
                print(chunk.decode('utf-8'), end='', flush=True)

5. 异常处理与监控

5.1 错误代码处理

def safe_call(prompt):
    try:
        response = ernie_35_8k(prompt)
        if 'error_code' in response:
            if response['error_code'] == 17:  # QPS超限
                time.sleep(1)
                return safe_call(prompt)
            raise Exception(f"API Error: {response['error_msg']}")
        return response['result']
    except requests.exceptions.RequestException as e:
        # 重试逻辑
        return handle_retry(e, prompt)

5.2 性能监控建议

• 记录API响应时间（P99应<2s）
• 监控token消耗速率
• 设置QPS熔断机制

6. 最佳实践案例

6.1 客服问答系统

def customer_service(query):
    context = """你是一个专业客服，请用中文回答用户问题。
    已知产品信息：..."""
    return ernie_35_8k(context + query)

6.2 批量文本处理

使用异步提高吞吐量：

import asyncio
async def batch_process(prompts):
    semaphore = asyncio.Semaphore(10)  # 并发控制
    async with aiohttp.ClientSession() as session:
        tasks = [process_one(session, p, semaphore) for p in prompts]
        return await asyncio.gather(*tasks)

7. 注意事项

1. 严格遵守《文心一言API使用规范》
2. 重要业务需实现fallback机制
3. 生产环境建议使用私有化部署方案
4. 敏感数据需进行脱敏处理

通过本文指南，开发者可快速构建基于ERNIE-3.5-8K的智能文本处理系统。建议结合官方文档（https://cloud.baidu.com/doc/WENXINWORKSHOP/）获取最新API变更信息。