在构建自然语言处理应用时，大模型的响应速度往往直接决定了用户体验。传统的HTTP请求采用一问一答的模式，必须等待模型生成完整结果后才能返回给客户端，这在处理长文本时会导致明显的等待时间。讯飞星火API提供了基于WebSocket的实时流式对话能力，允许服务端在模型生成过程中逐字推送结果，从而实现类似打字机的效果。不仅如此，通过功能调用机制，大模型还能与外部业务系统进行交互，极大地拓展了应用边界。
WebSocket协议与星火API的鉴权机制
WebSocket是一种在单个TCP连接上进行全双工通信的协议。与HTTP每次请求都需要建立新连接不同，WebSocket建立连接后，客户端和服务端可以随时互相发送数据。这种特性非常适合流式对话场景，因为大模型在推理时是逐个Token生成的，通过WebSocket可以立即将生成的Token推送给客户端，而不必等待整段文本生成完毕。
要建立WebSocket连接，首先需要解决鉴权问题。讯飞星火API的鉴权机制基于HMAC-SHA256算法生成签名。开发者需要使用APIKey和APISecret对请求参数进行加密，生成一个带有时间戳和签名的URL。服务端接收到连接请求后，会验证签名是否有效以及时间戳是否在有效期内。这种机制确保了请求的合法性和不可篡改性，防止API密钥被恶意盗用。
import base64
import hashlib
import hmac
from datetime import datetime
from urllib.parse import urlencode
def generate_auth_url(api_key, api_secret, host, path, method="GET"):
# 生成RFC1123格式的日期时间戳
utc_now = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
# 拼接签名原文
signature_origin = f"host: {host}\ndate: {utc_now}\n{method} {path} HTTP/1.1"
# 使用HMAC-SHA256进行加密
signature_sha = hmac.new(api_secret.encode('utf-8'), signature_origin.encode('utf-8'), hashlib.sha256).digest()
signature = base64.b64encode(signature_sha).decode('utf-8')
# 构建请求参数
authorization_origin = f'api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature}"'
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode('utf-8')
# 生成最终鉴权URL
params = {
'authorization': authorization,
'date': utc_now,
'host': host
}
url = f"wss://{host}{path}?{urlencode(params)}"
return url
在上述代码中，生成签名原文时必须严格按照指定的格式拼接，包括host、date和request-line。日期参数必须使用UTC时间并符合RFC1123格式，如果时区或格式不正确，服务端会直接拒绝连接。此外，生成的鉴权URL通常具有较短的有效期，一般为5分钟左右，因此客户端在发起连接前应实时生成，避免使用过期的URL导致连接失败。
构建实时流式对话请求与解析
建立WebSocket连接后，客户端需要发送特定结构的JSON数据来触发对话。星火API的请求体主要包含header、parameter和payload三个部分。header中通常包含应用ID和对话场景标识，parameter用于设置模型参数如最大Token数、温度值等，payload则包含了实际的对话历史记录和用户当前提问。通过维护对话历史，可以让模型理解上下文，实现多轮对话。
当服务端接收到请求后，会开始进行推理，并通过WebSocket连接持续推送JSON格式的响应数据。每个响应数据包包含当前生成的文本片段、本次对话的唯一标识以及结束标志。客户端需要不断接收这些数据包，提取其中的文本片段并追加到界面上，从而实现流式输出的视觉效果。当响应数据中的状态码指示会话结束时，客户端可以关闭连接或等待下一次提问。
import json
import websocket
def on_message(ws, message):
data = json.loads(message)
# 解析返回的文本片段
if data['header']['code'] != 0:
print(f"请求出错: {data['header']['message']}")
ws.close()
return
# 提取AI生成的文本
choices = data.get('payload', {}).get('choices', {})
if choices:
content = choices.get('text', {}).get('content', '')
print(content, end='', flush=True)
# 判断是否结束
if data['header']['status'] == 2:
print("\n对话结束")
ws.close()
def on_error(ws, error):
print(f"WebSocket错误: {error}")
def on_open(ws):
# 构造请求参数
payload = {
"header": {
"app_id": "your_app_id",
"uid": "user_123"
},
"parameter": {
"chat": {
"domain": "generalv3",
"max_tokens": 1024,
"temperature": 0.5
}
},
"payload": {
"message": {
"text": [
{"role": "user", "content": "请用一句话介绍WebSocket协议"}
]
}
}
}
ws.send(json.dumps(payload))
# 启动WebSocket客户端
# ws = websocket.WebSocketApp(auth_url, on_message=on_message, on_error=on_error, on_open=on_open)
# ws.run_forever()
在解析流式数据时，最关键的是判断header中的status字段。通常status为1表示正在生成中，为2表示当前这轮对话已经全部生成完毕。如果在网络不稳定的情况下，可能会出现数据包乱序或丢失的情况，因此建议在每轮对话结束后，根据完整的对话历史重新发起请求进行重试，而不是依赖断点续传。同时，为了防止长时间占用连接，应当设置合理的超时时间。
深度解析Function Call功能调用
大模型虽然具备丰富的知识储备，但其训练数据是静态的，无法直接获取实时天气、数据库中的订单状态等动态信息。Function Call功能就是为了解决这一痛点而设计的。开发者可以预先定义一组外部函数的名称、描述和参数结构，并将其随同用户提问一起发送给星火API。模型在理解用户意图后，如果发现需要调用外部工具，就会返回一个包含函数名和参数的指令，而不是直接生成自然语言回复。
在接收到模型返回的函数调用指令后，客户端应用需要解析出函数名和参数，在本地执行对应的业务逻辑，例如查询数据库或调用第三方API。获取到执行结果后，再将结果以特定的格式返回给大模型。大模型会结合这个结果和之前的对话上下文，为用户生成最终的自然语言总结。这种机制将大模型的推理能力与外部系统的数据获取能力完美结合。
# 定义外部函数结构
functions = [
{
"name": "get_weather",
"description": "获取指定城市的实时天气信息",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称，例如：北京"
}
},
"required": ["location"]
}
}
]
# 构造带有Function Call的请求Payload
payload_with_fc = {
"header": {
"app_id": "your_app_id"
},
"parameter": {
"chat": {
"domain": "generalv3",
"functions": functions
}
},
"payload": {
"message": {
"text": [
{"role": "user", "content": "今天北京天气怎么样？"}
]
}
}
}
# 模拟接收到的模型响应
mock_response = {
"header": {"code": 0, "status": 2},
"payload": {
"choices": {
"text": [
{
"role": "assistant",
"content": "",
"function_call": {
"name": "get_weather",
"arguments": "{\"location\":\"北京\"}"
}
}
]
}
}
}
# 客户端处理函数调用
def handle_function_call(response_data):
choices = response_data.get('payload', {}).get('choices', {})
text_list = choices.get('text', [])
if text_list:
func_call = text_list[0].get('function_call')
if func_call:
func_name = func_call['name']
args = json.loads(func_call['arguments'])
# 执行本地函数
if func_name == "get_weather":
weather_info = query_local_weather(args['location'])
# 将结果返回给大模型进行下一轮对话
return weather_info
return None
def query_local_weather(location):
# 模拟查询外部天气API
return f"{location}今天晴，气温25度"
使用Function Call时，函数描述的清晰度直接决定了模型调用的准确率。在定义description时，应当尽可能详细地说明函数的作用、适用场景以及参数的具体含义。如果参数是一个枚举值，最好在描述中列出所有可能的选项。此外，当本地函数执行失败或超时时，也应当将错误信息以结构化的方式返回给大模型，让模型能够自行判断是否需要向用户道歉或尝试调用其他备用函数。通过合理设计函数调用链路，可以构建出高度智能化的Agent系统。