
AI回答采集系统需要稳定调用大模型API,获取结构化响应。本文以腾讯混元大模型为例,介绍从认证、SDK集成、流式输出、结构化响应到错误处理的完整工程实践。适合需要构建AI数据采集或监测系统的开发者参考。前提:已开通腾讯混元API服务,具备基本Python开发环境。
在构建AI回答采集系统时,需要向多个大模型发送问题并获取回答。每次调用看似简单,但进入工程化后,认证、超时、流式解析、结构化输出、错误重试等问题逐一浮现。本文只解决一个问题:如何稳定、可复现地调用腾讯混元API,获取结构化的回答数据。
采集系统与模型调用的关系如下:
模型调用模块是核心,负责认证、请求、响应解析和错误处理。本文聚焦该模块的腾讯混元实现。
pip install tencentcloud-sdk-pythonap-guangzhou,具体以官方文档为准hunyuan-pro(版本可能更新,请以官方为准)具体版本未提供,请根据项目实际环境和官方兼容性要求选择。
腾讯混元使用腾讯云API 3.0签名认证。SDK封装了签名过程,只需配置SecretId和SecretKey。
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models
# 密钥应通过环境变量或密钥管理服务获取,不要硬编码
cred = credential.Credential(
secret_id=os.getenv("TENCENT_SECRET_ID"),
secret_key=os.getenv("TENCENT_SECRET_KEY")
)
client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")说明:
混元API支持流式和非流式输出。采集系统通常需要完整回答,但流式输出可以降低首字延迟。以下示例使用流式输出并拼接完整结果。
def call_hunyuan_stream(prompt: str) -> str:
req = models.ChatCompletionsRequest()
req.Model = "hunyuan-pro"
req.Messages = [
{"Role": "user", "Content": prompt}
]
req.Stream = True
req.Temperature = 0.7
req.TopP = 0.9
try:
resp = client.ChatCompletions(req)
full_content = ""
for event in resp:
if event.Choices and len(event.Choices) > 0:
delta = event.Choices[0].Delta
if delta and delta.Content:
full_content += delta.Content
return full_content
except TencentCloudSDKException as e:
# 错误处理见后文
raise说明:
Stream=True启用流式输出,返回迭代器。event包含增量内容,需要拼接。Temperature和TopP控制随机性,采集场景建议固定值以保证可复现性。采集系统需要将回答解析为结构化字段,如回答文本、Token数、模型版本等。混元API的响应中包含Usage字段,可获取Token消耗。
def parse_response(resp) -> dict:
# 非流式响应示例
choice = resp.Choices[0]
message = choice.Message
usage = resp.Usage
return {
"content": message.Content,
"role": message.Role,
"prompt_tokens": usage.PromptTokens,
"completion_tokens": usage.CompletionTokens,
"total_tokens": usage.TotalTokens,
"model": resp.Model
}说明:
Usage只在最后一个event中返回,需特殊处理。网络抖动、限流、参数错误等都会导致调用失败。必须实现重试和降级。
import time
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
def call_with_retry(prompt: str, max_retries=3, base_delay=1):
last_exception = None
for attempt in range(max_retries):
try:
return call_hunyuan_stream(prompt)
except TencentCloudSDKException as e:
last_exception = e
# 限流错误码:FailedOperation.RequestLimitExceeded
if "RequestLimitExceeded" in str(e):
delay = base_delay * (2 ** attempt)
time.sleep(delay)
continue
# 其他错误直接抛出
raise
raise last_exception说明:
AuthFailure.SignatureFailure。InvalidParameterValue提示模型名称错误。本文以腾讯混元为例,介绍了AI回答采集中模型调用的工程实践:认证初始化、流式输出、结构化响应解析和错误重试。关键实现是使用SDK进行签名认证,流式拼接完整回答,并实现指数退避重试。适用于需要稳定调用混元API的采集系统。需要注意的是,模型版本、地域和价格可能变化,请以官方文档为准;重试机制不能解决所有问题,异常样本应保留原始响应用于排查。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。