### 使用 Python 调用星火大模型 API 的教程与代码示例 星火大模型提供了丰富的 API 接口，可以使用 Python 进行调用。以下是一个完整的指南以及代码示例来帮助您理解如何操作。 #### 准备工作 在开始之前，请确保已经注册并获得了星火大模型的 `appid`, `api_secret` 和 `api_key`。这些参数将在后续代码中作为认证凭证使用[^1]。 #### 安装必要的库 为了简化 WebSocket 通信过程，推荐安装 `websockets` 库。如果尚未安装该库，则可通过 pip 工具执行如下命令： ```bash pip install websockets ``` #### 星火大模型 API 调用的核心逻辑 下面展示了一段基于 WebSocket 的 Python 示例代码，用于调用星火大模型 API 并与其进行交互[^1]。 ```python import asyncio import json import uuid from urllib.parse import urlencode import hmac import base64 import hashlib import time import websockets class XFYunStarfireAPI: def __init__(self, appid, api_key, api_secret): self.appid = appid self.api_key = api_key self.api_secret = api_secret self.url = "wss://spark-api.xf-yun.com/v3.1/chat" async def connect(self): timestamp = str(int(time.time())) signa = self.generate_signa(timestamp) query_string = urlencode({ 'app_id': self.appid, 'timestamp': timestamp, 'signa': signa }) websocket_url = f"{self.url}?{query_string}" async with websockets.connect(websocket_url) as websocket: await self.send_message(websocket) await self.receive_messages(websocket) def generate_signa(self, timestamp): """生成签名""" string_to_sign = f"host: spark-api.xf-yun.com\n" \ f"date: {timestamp}\n" \ f"request-line: GET /v3.1/chat HTTP/1.1" hashed = hmac.new( key=self.api_secret.encode(), msg=string_to_sign.encode(), digestmod=hashlib.sha256 ) signature = base64.b64encode(hashed.digest()).decode() return signature async def send_message(self, websocket): """发送消息至星火大模型""" message_id = str(uuid.uuid4()) payload = { "header": { "app_id": self.appid, "uid": "your_user_identifier_here", }, "parameter": { "chat": { "domain": "general", "random_threshold": 0.5, "max_tokens": 2048, "auditing": "default" } }, "payload": { "message": { "text": [ {"role": "user", "content": "你好"}, {"role": "assistant", "content": ""} ] } } } await websocket.send(json.dumps(payload)) print(f"[Sent Message ID={message_id}]") async def receive_messages(self, websocket): """接收来自星火大模型的消息""" while True: response = await websocket.recv() parsed_response = json.loads(response) if 'payload' in parsed_response and 'message' in parsed_response['payload']: content = parsed_response['payload']['message'][0]['content'] role = parsed_response['payload']['message'][0]['role'] if role == 'assistant': print(f"[Received Assistant Response]: {content}") elif 'code' in parsed_response and int(parsed_response['code']) != 0: error_code = parsed_response['code'] error_message = parsed_response['message'] if 'message' in parsed_response else '' print(f"[Error Code={error_code}]: {error_message}") if __name__ == "__main__": APP_ID = "your_app_id_here" API_KEY = "your_api_key_here" API_SECRET = "your_api_secret_here" starfire_client = XFYunStarfireAPI(APP_ID, API_KEY, API_SECRET) asyncio.run(starfire_client.connect()) ``` --- #### 关键点说明 1. **签名算法**：通过 HMAC-SHA256 加密方式生成签名字符串 `signa`，这是连接星火大模型 API 所必需的安全措施之一。 2. **WebSocket URL 构造**：将 `appid`, 时间戳 (`timestamp`) 及计算得到的签名拼接到基础 URL 后面形成最终请求地址[^1]。 3. **JSON 数据结构设计**：按照官方文档规定格式化上传给服务器的内容体部分，包括对话历史记录等字段设置[^1]。 4. **错误处理机制**：考虑到网络波动或者输入非法等情况可能导致失败情形发生，因此增加了简单的异常捕获环节以提升用户体验度[^1]。 ---