diff --git a/examples/cls/v20201016/chat_completions.py b/examples/cls/v20201016/chat_completions.py new file mode 100644 index 000000000..de745b93c --- /dev/null +++ b/examples/cls/v20201016/chat_completions.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +import json +import os + +from tencentcloud.common import credential +from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException +from tencentcloud.common.profile.client_profile import ClientProfile +from tencentcloud.cls.v20201016 import cls_client, models + +# 1. 创建认证凭证（Credential） +# 从环境变量中获取密钥 ID 和密钥 Key，这是访问云 API 的必需凭证。 +cred = credential.Credential( + os.environ.get("TENCENTCLOUD_SECRET_ID"), + os.environ.get("TENCENTCLOUD_SECRET_KEY")) + +# 2. 创建客户端配置（ClientProfile） +cpf = ClientProfile() +cpf.httpProfile.reqTimeout = 300 # 流式接口可能耗时较长，因此将请求超时时间设置为 300 秒。 +cpf.httpProfile.endpoint = "cls.ai.tencentcloudapi.com" # 通过SSE流式调用此接口时，请务必设置请求域名（Endpoint）为 cls.ai.tencentcloudapi.com。 +client = cls_client.ClsClient(cred, "ap-guangzhou", cpf) + +# 3. 构造公共请求参数 +msg = models.Message() +msg.Role = "user" +msg.Content = "状态码200, 统计日志条数" + +meta_region = models.MetadataItem() +meta_region.Key = "topic_region" +meta_region.Value = "ap-guangzhou" +meta_topic = models.MetadataItem() +meta_topic.Key = "topic_id" +meta_topic.Value = "xxxxxxxx-xxxx" + +# 4. 流式调用 +# CLS ChatCompletions 同时支持流式和非流式两种模式。 +# 将 Stream 参数设置为 True，表示以流式方式获取响应。 +try: + req = models.ChatCompletionsRequest() + req.Model = "text2sql" + req.Messages = [msg] + req.Metadata = [meta_region, meta_topic] + req.Stream = True + resp = client.ChatCompletions(req) + + for event in resp: + # 跳过心跳等 data 为空的事件（如服务端每隔一段时间发送的 :heartbeat 注释） + if not event.get('data'): + continue + # 服务端通过 event: error 返回应用层错误 + if event.get("event") == "error": + err = json.loads(event["data"]) + print("\n[流式] error", err["Response"]["Error"]) + continue + data = json.loads(event['data']) + for choice in data.get('Choices', []): + # 实时输出思考过程（ReasoningContent） + if choice['Delta'].get('ReasoningContent'): + print(choice['Delta']['ReasoningContent'], end='', flush=True) + # 实时输出回复内容（Content） + if choice['Delta'].get('Content'): + print(choice['Delta']['Content'], end='', flush=True) + print() + print('[流式] 完成，开始非流式调用...') + +except TencentCloudSDKException as err: + print("[流式] error", err) + +# 5. 非流式调用 +# 通过 Stream=False 参数来指定非 stream 协议，一次性拿到结果。 +try: + req = models.ChatCompletionsRequest() + req.Model = "text2sql" + req.Messages = [msg] + req.Metadata = [meta_region, meta_topic] + req.Stream = False + resp = client.ChatCompletions(req) + + if resp.Choices[0].Message.ReasoningContent: + print("思考过程：", resp.Choices[0].Message.ReasoningContent) + print("[非流式] 回复内容：\n" + resp.Choices[0].Message.Content) + +except TencentCloudSDKException as err: + print("[非流式] error", err) diff --git a/examples/cls/v20201016/chat_completions_async.py b/examples/cls/v20201016/chat_completions_async.py new file mode 100644 index 000000000..da8e31d11 --- /dev/null +++ b/examples/cls/v20201016/chat_completions_async.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*- +# Copyright 2017-2021 Tencent Ltd. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +这是一个使用腾讯云 SDK 异步 API 的示例。 +该脚本演示了如何以流式（stream）方式调用 CLS ChatCompletions 接口。 +""" +import asyncio +import json +import os + +from tencentcloud.common import credential +from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException +from tencentcloud.common.profile.client_profile import ClientProfile +from tencentcloud.cls.v20201016 import cls_client_async, models + + +async def main(): + """ + 主函数，异步执行与 CLS 的对话。 + """ + # 1. 创建认证凭证（Credential） + # 从环境变量中获取密钥 ID 和密钥 Key，这是访问云 API 的必需凭证。 + cred = credential.Credential( + os.environ.get("TENCENTCLOUD_SECRET_ID"), + os.environ.get("TENCENTCLOUD_SECRET_KEY")) + + # 2. 创建客户端配置（ClientProfile） + cpf = ClientProfile() + cpf.httpProfile.reqTimeout = 300 # 流式接口可能耗时较长，因此将请求超时时间设置为 300 秒。 + cpf.httpProfile.endpoint = "cls.ai.tencentcloudapi.com" # 通过SSE流式调用此接口时，请务必设置请求域名（Endpoint）为 cls.ai.tencentcloudapi.com。 + + # 3. 创建异步 ClsClient 客户端 + # 使用 'async with' 语法可以确保客户端在使用完毕后自动关闭连接。 + async with cls_client_async.ClsClient(cred, "ap-guangzhou", cpf) as client: + # 4. 构造公共请求参数 + msg = models.Message() + msg.Role = "user" + msg.Content = "状态码200, 统计日志条数" + + meta_region = models.MetadataItem() + meta_region.Key = "topic_region" + meta_region.Value = "ap-guangzhou" + meta_topic = models.MetadataItem() + meta_topic.Key = "topic_id" + meta_topic.Value = "xxxxxxxx-xxxx" + + # 5. 流式调用 + # CLS ChatCompletions 同时支持流式和非流式两种模式。 + # 将 Stream 参数设置为 True，表示以流式方式获取响应。 + try: + req = models.ChatCompletionsRequest() + req.Model = "text2sql" + req.Messages = [msg] + req.Metadata = [meta_region, meta_topic] + req.Stream = True + resp = await client.ChatCompletions(req) + + async for event in resp: + # 跳过心跳等 data 为空的事件（如服务端每隔一段时间发送的 :heartbeat 注释） + if not event.get('data'): + continue + # 服务端通过 event: error 返回应用层错误 + if event.get("event") == "error": + err = json.loads(event["data"]) + print("\n[流式] error", err["Response"]["Error"]) + continue + data = json.loads(event['data']) + for choice in data.get('Choices', []): + # 实时输出思考过程（ReasoningContent） + if choice['Delta'].get('ReasoningContent'): + print(choice['Delta']['ReasoningContent'], end='', flush=True) + # 实时输出回复内容（Content） + if choice['Delta'].get('Content'): + print(choice['Delta']['Content'], end='', flush=True) + print() + print('[流式] 完成，开始非流式调用...') + + except TencentCloudSDKException as err: + print("[流式] error", err) + + # 6. 非流式调用 + # 通过 Stream=False 参数来指定非 stream 协议，一次性拿到结果。 + try: + req = models.ChatCompletionsRequest() + req.Model = "text2sql" + req.Messages = [msg] + req.Metadata = [meta_region, meta_topic] + req.Stream = False + resp = await client.ChatCompletions(req) + + if resp.Choices[0].Message.ReasoningContent: + print("思考过程：", resp.Choices[0].Message.ReasoningContent) + print("[非流式] 回复内容：\n" + resp.Choices[0].Message.Content) + + except TencentCloudSDKException as err: + print("[非流式] error", err) + + +# 7. 运行主函数 +asyncio.run(main())