百度千帆大模型API调用保姆级教程：从申请Key到实现流式对话（附完整Python代码）

百度千帆大模型API实战指南：从零构建智能对话系统

第一次接触大模型API时，我盯着百度智能云控制台密密麻麻的菜单栏，手指悬在键盘上却不知从何下手。那种既兴奋又忐忑的心情至今记忆犹新——兴奋于即将解锁AI的强大能力，忐忑于可能遇到的复杂技术门槛。这份指南正是为了帮助像当时的我这样的开发者，用最直接的方式跨越从注册账号到实现流式对话的全过程。

1. 环境准备与账号配置

在开始编码之前，我们需要先搭建好开发环境并完成必要的账号设置。这个过程看似简单，但往往隐藏着许多新手容易踩坑的细节。

1.1 开发环境搭建

推荐使用Python 3.8或更高版本，这是目前最稳定兼容千帆大模型API的Python环境。如果你尚未安装Python，可以从官网下载对应操作系统的安装包：

# 检查Python版本 python --version # 如果同时安装了Python3，可能需要使用 python3 --version

安装必要的依赖库，这些库将帮助我们处理HTTP请求和JSON数据：

pip install requests json

对于IDE的选择，PyCharm或VS Code都是不错的选项。我个人更倾向于VS Code，因为它轻量且插件丰富，特别是对Python的支持非常完善。

1.2 百度智能云账号注册

访问百度智能云官网，点击注册按钮。这里有个小技巧：使用企业邮箱注册通常会比个人邮箱获得更高的初始配额。注册过程中需要：

1. 填写基本信息（邮箱/手机号）
2. 完成实名认证（个人/企业）
3. 设置支付方式（即使使用免费额度也需要）

注意：虽然Yi-34B-Chat有免费额度，但百度智能云要求账户必须完成实名认证和支付方式绑定才能调用API。

1.3 创建应用获取API密钥

登录控制台后，按照以下路径创建应用：

1. 进入"产品服务"→"人工智能"→"千帆大模型平台"
2. 在左侧菜单选择"应用接入"
3. 点击"创建应用"按钮

填写应用基本信息时，"应用名称"和"应用描述"可以自由填写，但建议使用有意义的名称以便后续管理。创建成功后，系统会生成两个关键凭证：

• API Key：用于标识你的应用
• Secret Key：用于获取访问令牌

这两个密钥相当于你的账号密码，务必妥善保管。我习惯将它们存储在环境变量中，而不是直接硬编码在脚本里：

# 安全提示：不要将密钥直接写在代码中 import os API_KEY = os.getenv('QIANFAN_API_KEY') SECRET_KEY = os.getenv('QIANFAN_SECRET_KEY')

2. 理解API基础调用流程

千帆大模型的API调用遵循标准的OAuth2.0认证流程，理解这个流程对后续调试和问题排查非常重要。

2.1 认证机制解析

每次调用API前，都需要先用API Key和Secret Key获取access_token。这个token的有效期通常是30天，但建议每次调用都获取新的token以确保安全。认证过程分为两步：

1. 向认证服务器发送请求获取access_token
2. 使用access_token调用大模型API

sequenceDiagram 开发者->>认证服务器: 发送API Key和Secret Key 认证服务器-->>开发者: 返回access_token 开发者->>千帆API: 发送请求+access_token 千帆API-->>开发者: 返回AI响应

2.2 获取access_token的代码实现

下面是一个健壮的access_token获取函数，包含了错误处理和重试机制：

def get_access_token_with_retry(api_key, secret_key, max_retries=3): url = f"https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id={api_key}&client_secret={secret_key}" for attempt in range(max_retries): try: response = requests.post(url) response.raise_for_status() # 检查HTTP错误 token_data = response.json() if 'access_token' not in token_data: raise ValueError("响应中缺少access_token字段") return token_data['access_token'] except requests.exceptions.RequestException as e: print(f"获取token失败，尝试 {attempt + 1}/{max_retries}: {str(e)}") if attempt == max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避

2.3 首次API调用测试

让我们先实现一个最简单的单轮对话功能，验证整个流程是否畅通：

def simple_chat(prompt): token = get_access_token(API_KEY, SECRET_KEY) url = f"https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/yi_34b_chat?access_token={token}" payload = { "messages": [ { "role": "user", "content": prompt } ] } response = requests.post(url, json=payload) return response.json()['result'] # 测试调用 print(simple_chat("你好，介绍一下你自己"))

如果一切正常，你将看到Yi-34B模型的自我介绍。如果遇到错误，常见的排查点包括：

• 密钥是否正确（注意区分大小写）
• 账号是否已完成实名认证
• 网络是否能正常访问百度服务器
• 是否超过了免费额度限制

3. 实现流式对话功能

流式对话(Streaming)是大模型交互中的重要特性，它允许我们逐步接收响应内容，而不是等待整个响应完成。这种模式特别适合长文本生成场景，能显著提升用户体验。

3.1 流式对话原理

传统API调用是"请求-等待-完整响应"模式，而流式对话则是"请求-持续接收部分响应"模式。技术实现上，服务器会保持连接开放，通过Server-Sent Events(SSE)持续发送数据片段。

def stream_chat(messages): token = get_access_token(API_KEY, SECRET_KEY) url = f"https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/yi_34b_chat?access_token={token}" payload = { "messages": messages, "stream": True # 关键参数，启用流式 } with requests.post(url, json=payload, stream=True) as response: for line in response.iter_lines(): if line: decoded_line = line.decode('utf-8') if decoded_line.startswith('data:'): json_str = decoded_line[5:].strip() try: data = json.loads(json_str) yield data.get('result', '') except json.JSONDecodeError: continue

3.2 上下文保持技巧

要实现真正的多轮对话，关键在于正确维护对话历史。每次调用都需要包含之前的对话上下文，模型才能理解当前对话的语境。

class Conversation: def __init__(self): self.history = [] def add_user_message(self, content): self.history.append({"role": "user", "content": content}) def add_assistant_message(self, content): self.history.append({"role": "assistant", "content": content}) def get_response_stream(self): token = get_access_token(API_KEY, SECRET_KEY) url = f"https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/yi_34b_chat?access_token={token}" payload = { "messages": self.history, "stream": True } with requests.post(url, json=payload, stream=True) as response: full_response = "" for line in response.iter_lines(): if line: decoded_line = line.decode('utf-8') if decoded_line.startswith('data:'): json_str = decoded_line[5:].strip() try: data = json.loads(json_str) chunk = data.get('result', '') full_response += chunk yield chunk except json.JSONDecodeError: continue self.add_assistant_message(full_response)

3.3 完整对话示例

结合上述组件，我们可以构建一个完整的命令行聊天程序：

def interactive_chat(): conv = Conversation() print("输入'quit'退出对话") while True: user_input = input("\nYou: ") if user_input.lower() == 'quit': break conv.add_user_message(user_input) print("\nAI: ", end='', flush=True) for chunk in conv.get_response_stream(): print(chunk, end='', flush=True) print()

这个实现包含了所有核心功能：

• 流式输出，实时显示模型响应
• 上下文保持，支持多轮对话
• 简洁的用户界面

4. 高级应用与优化技巧

掌握了基础调用后，我们可以进一步探索API的高级功能，提升应用性能和用户体验。

4.1 参数调优

千帆API提供了多个参数来控制模型行为，合理设置这些参数可以获得更符合需求的输出：

示例调用：

payload = { "messages": messages, "stream": True, "temperature": 0.7, # 稍微降低随机性 "max_output_tokens": 1024, # 限制响应长度 "penalty_score": 1.2 # 减少重复 }

4.2 错误处理与重试

网络服务难免会遇到临时故障，健壮的应用应该具备错误处理和自动重试能力：

def robust_api_call(messages, max_retries=3): for attempt in range(max_retries): try: token = get_access_token(API_KEY, SECRET_KEY) url = f"https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/yi_34b_chat?access_token={token}" payload = { "messages": messages, "stream": True } response = requests.post(url, json=payload, stream=True, timeout=30) response.raise_for_status() # 处理流式响应... return process_stream(response) except requests.exceptions.RequestException as e: print(f"API调用失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}") if attempt == max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避

4.3 性能优化

对于高频调用的应用，可以考虑以下优化措施：

1. Token缓存：access_token通常有30分钟有效期，可以缓存而不是每次都重新获取
2. 连接池：使用requests.Session()重用HTTP连接
3. 异步调用：对于高并发场景，使用aiohttp等异步库

import aiohttp import asyncio async def async_chat(session, messages): token = get_access_token(API_KEY, SECRET_KEY) url = f"https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/yi_34b_chat?access_token={token}" async with session.post(url, json={"messages": messages}) as response: return await response.json()

4.4 实际应用案例

在实际项目中，我们可以将千帆API集成到各种应用场景中：

1. 客服机器人：结合业务知识库提供智能客服
2. 内容生成：自动生成产品描述、营销文案等
3. 代码助手：帮助开发者编写和优化代码
4. 教育应用：作为智能辅导老师解答学生问题

一个电商客服集成的伪代码示例：

def handle_customer_query(query, product_db): # 从数据库获取产品信息 product_info = product_db.query(query) # 构建提示词 prompt = f"你是一位电商客服助手。用户询问：{query}\n\n相关产品信息：{product_info}\n\n请用友好的语气回答用户问题，突出产品优势但不要夸大。" # 调用API response = simple_chat(prompt) # 后处理 return response.replace("作为一个AI", "作为客服")