说明

目前天翼云息壤的新老用户均可免费体验2500万Tokens，限时两周。模型广场支持DeepSeek-R1、DeepSeek-V3等多个版本模型的调用。

支持开发者以OpenAI-API-compatible接口方式集成调用。

天翼云官网获取模型调用APP key等信息

模型API终端请求地址

请求地址

API终端请求地址：https://wishub-x1.ctyun.cn/v1/chat/completions

通信协议

接口通过 HTTPS 进行通信，保护用户数据的机密性和完整性，确保网络通信的安全性。

请求方法

支持的 HTTP 请求方法：POST。
POST 请求支持的 Content-Type 类型：application/json。

获取模型调用APP key

访问天翼云息壤deepseek活动页地址：https://www.ctyun.cn/act/xirang/deepseek，选择“立即体验”。

登录天翼云账户，进入“体验中心”后切换到“服务接入”。

在“服务接入”下面，创建建服务组。选择默认勾选的模型即可。

创建服务组后，从服务组上面获取APP KEY。

获取调用的模型名称

左侧导航栏切换到“模型广场”，选择对应想要调用的模型。

获取模型ID作为模型名称。

例如下图中模型ID为 : 4bd107bff85941239e27b1509eccfe98

DeepSeek模型版本选择参考：

DeepSeek-R1：适合需要高级推理能力和高精度的复杂任务，如科研、金融建模等。响应时间相对长。
DeepSeek-V3：适合需要快速响应和高性价比的通用自然语言处理任务，如智能客服、内容生成等。响应时间相对短。

手动验证key的可用性

下列请求如果正常返回deepseek回答，代表没有问题。

注意

以下curl请求中替换成自己的APP KEY后类似“--header 'Authorization: Bearer abcdefghxxsssss' \ ”

 curl --location 'https://wishub-x1.ctyun.cn/v1/chat/completions' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer 替换成自己的APP KEY' \  
--data '{
    "messages": [
        {
            "role": "system",
            "content": "You are a test assistant."
        },
        {
            "role": "user",
            "content": "Testing. Just say hi and nothing else."
        }
    ],
    "model": "4bd107bff85941239e27b1509eccfe98" //DeepSeek-R1模型名
}'

调用代码示例

说明

部分第三方工具调用可能存在接口不适配问题，如涉及工具调用时，建议优先使用下文提及的几款工具。
涉及第三方工具配置时，API key等同于上述内容中创建的服务组APP key。

同步调用

调用主程序

import requests
import json
from prompt import BASIC_CHAT_PROMPT, MATH_PROBLEM_PROMPT, CREATIVE_WRITING_PROMPT, CODE_REVIEW_PROMPT


def send_chat_request(url, app_key, model_id, user_message):
    """
    发送聊天请求到指定的API接口。

    :param url: API接口的URL
    :param app_key: 应用的授权密钥
    :param model_id: 使用的模型ID
    :param user_message: 用户发送的消息内容
    :return: 返回API的响应内容
    """
    # 设置请求头
    headers = {
        'Authorization': f'Bearer {app_key}',
        'Content-Type': 'application/json'
    }

    # 准备请求数据
    data = {
        "model": model_id,
        "messages": [
            {
                "role": "user",
                "content": user_message
            }
        ]
    }

    # 发送POST请求
    try:
        response = requests.post(url, headers=headers, data=json.dumps(data))
        response.raise_for_status()  # 检查请求是否成功
        return response.json()  # 返回JSON格式的响应内容
    except requests.exceptions.RequestException as e:
        # 捕获请求异常，返回错误信息
        return {
            "error": "请求失败",
            "message": str(e),
            "status_code": response.status_code if 'response' in locals() else None
        }


def print_response(response):
    """
    打印API响应内容
    
    :param response: API响应内容
    """
    if response.get("code") == 0:
        content = response.get("choices", [{}])[0].get("message", {}).get("content", "")
        print("返回的content内容为：")
        print(content)
    else:
        print(json.dumps(response, indent=4, ensure_ascii=False))
        print("请求失败，返回的错误信息为：")
        print(response.get("error", "未知错误"))


# 示例调用
if __name__ == "__main__":
    # API配置
    url = "https://wishub-x1.ctyun.cn/v1/chat/completions"
    # 从环境变量获取API密钥，如果没有设置，也可以直接终端执行export XIRANG_API_KEY="xxx"
    openai_api_key=os.getenv("XIRANG_API_KEY")
    DeepSeek_R1_model_id = "4bd107bff85941239e27b1509eccfe98"
    DeepSeek_V3_model_id ="9dc913a037774fc0b248376905c85da5"

    # 测试不同的prompt
    print("\n=== 基础对话测试 ===")
    response = send_chat_request(url, app_key, DeepSeek_V3_model_id, BASIC_CHAT_PROMPT)
    print_response(response)

prompt模型代码示例

"""
This module contains various prompt examples for different scenarios.
"""

# 基础问答prompt
BASIC_CHAT_PROMPT = "你是谁"

# 数学长文本对话prompt
MATH_PROBLEM_PROMPT = """请你扮演一位耐心的数学老师，帮我解答以下数学问题。
这是一道关于概率论的问题：
在一个袋子里有5个红球和3个蓝球。小明随机取出两个球，然后放回袋子，接着又取出一个球。
请问：
1. 第三次取出红球的概率是多少？
2. 如果已知前两次都取出了红球，那么第三次取出红球的概率是多少？
3. 请详细解释计算过程，并说明为什么这两个概率可能不同。

要求：
1. 请用通俗易懂的语言解释
2. 给出详细的计算步骤
3. 说明概率论中的条件概率概念
"""

# 可以继续添加更多的prompt示例

流式调用

流式调用相对复杂。

APIClient代码：

import requests
import json
import urllib3
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class APIClient:
    def __init__(self, url, app_key):
        """
        初始化API客户端
        
        :param url: API接口的URL
        :param app_key: 应用的授权密钥
        """
        self.url = url
        self.headers = {
            'Authorization': f'Bearer {app_key}',
            'Content-Type': 'application/json'
        }
        self.session = requests.Session()
        self.session.verify = False
        self.timeout = 30  # 设置较长的超时时间

    def send_chat_request_stream(self, model_id, user_message):
        """
        发送流式聊天请求
        
        :param model_id: 使用的模型ID
        :param user_message: 用户发送的消息内容
        :yield: 流式响应内容
        """
        data = {
            "model": model_id,
            "messages": [
                {
                    "role": "user",
                    "content": user_message
                }
            ],
            "stream": True
        }

        logger.debug(f"Sending request to {self.url}")
        logger.debug(f"Request data: {json.dumps(data, ensure_ascii=False)}")

        try:
            with self.session.post(
                self.url, 
                headers=self.headers, 
                json=data, 
                stream=True,
                timeout=self.timeout
            ) as response:
                logger.debug(f"Response status code: {response.status_code}")
                response.raise_for_status()
                
                for line in response.iter_lines():
                    if not line:
                        continue
                        
                    try:
                        # 处理原始行数据
                        line_str = line.decode('utf-8')
                        logger.debug(f"Received line: {line_str}")
                        
                        # 处理data:前缀，无论有没有空格
                        if line_str.startswith('data:'):
                            json_str = line_str[5:].strip()  # 移除'data:'前缀并清理空白字符
                        else:
                            json_str = line_str.strip()
                            
                        if json_str == '[DONE]':
                            logger.debug("Received [DONE] signal")
                            break
                            
                        data = json.loads(json_str)
                        logger.debug(f"Parsed JSON: {json.dumps(data, ensure_ascii=False)}")
                        
                        if data.get("code") == 0:
                            choices = data.get("choices", [])
                            if choices:
                                delta = choices[0].get("delta", {})
                                # 检查是否只有role字段
                                if delta.get("role") and not delta.get("content"):
                                    logger.debug("Skipping role-only delta")
                                    continue
                                    
                                content = delta.get("content")
                                if content:
                                    logger.debug(f"Yielding content: {content}")
                                    yield content
                                else:
                                    logger.debug("No content in delta")
                            else:
                                logger.warning("No choices in response")
                        else:
                            logger.warning(f"Unexpected response code: {data.get('code')}")
                            
                    except json.JSONDecodeError as e:
                        logger.error(f"JSON decode error: {str(e)}, line: {line}")
                        continue
                    except Exception as e:
                        logger.error(f"Error processing response: {str(e)}")
                        yield f"\nError processing stream: {str(e)}"
                        break
                        
        except requests.exceptions.Timeout:
            error_msg = "Request timed out"
            logger.error(error_msg)
            yield f"\nError: {error_msg}"
        except requests.exceptions.RequestException as e:
            error_msg = f"Request failed: {str(e)}"
            logger.error(error_msg)
            yield f"\nError: {error_msg}"
        except Exception as e:
            error_msg = f"Unexpected error: {str(e)}"
            logger.error(error_msg)
            yield f"\nError: {error_msg}"

调用主程序：

def cleanup(client):
    """
    清理资源
    """
    try:
        del client
    except Exception:
        pass

def main():
    # API配置
    url = "https://wishub-x1.ctyun.cn/v1/chat/completions"
    # 从环境变量获取API密钥，如果没有设置，也可以直接终端执行export XIRANG_API_KEY="xxx"
    openai_api_key=os.getenv("XIRANG_API_KEY")
    DeepSeek_R1_model_id = "4bd107bff85941239e27b1509eccfe98"
    DeepSeek_V3_model_id ="9dc913a037774fc0b248376905c85da5"

    # 创建API客户端
    client = APIClient(url, app_key)
    atexit.register(cleanup, client)  # 注册清理函数

    try:

        # # 测试流式输出
        print("\n=== 流式输出：数学问题测试 ===")
        stream = client.send_chat_request_stream(DeepSeek_R1_st_model_id, MATH_PROBLEM_PROMPT)
        print_stream_response(stream)


    except Exception as e:
        print(f"\n程序执行出错: {str(e)}")
    finally:
        cleanup(client)


if __name__ == "__main__":
    main()

Langchain调用

参考langchain的自定义模型文档通过OpenAI协议调用即可。

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage

def main():
    # 初始化LangChain模型，配置自定义API端点
    llm = ChatOpenAI(
        model_name="9dc913a037774fc0b248376905c85da5",  # 自定义模型名称
       # 从环境变量获取API密钥，如果没有设置，也可以直接终端执行export XIRANG_API_KEY="xxx"
        openai_api_key=os.getenv("XIRANG_API_KEY")
        openai_api_base="https://wishub-x1.ctyun.cn/v1",  # API基础URL
        temperature=0.7
    )

    # 创建一个聊天消息
    message = HumanMessage(content="你是谁")
    
    # 使用LangChain进行对话
    response = llm([message])
    print(response.content)


if __name__ == "__main__":
    main()

Dify

说明

前置条件：本地下载并安装Docker。

访问 Docker 官方网站，点击“Download Docker Desktop”按钮，选择合适的 Windows、macOS 或 Linux版本下载并安装。

终端执行命令行

git clone  https://gitcode.com/gh_mirrors/di/dify.git 
cd docker
mv .env.example .env

#在.env文件添加
CUSTOM_MODEL_ENABLED=true
#执行:
docker compose up -d
# 等服务启动访问
http://localhost/apps

启动后，浏览器访问http://localhost/apps

打开设置界面添加自定义模型

Spring AI

注意

Spring AI要求使用JDK 17或更高版本，Spring AI需要在Spring Boot 3.2.x或更高版本的环境中运行。

核心代码如下：

OpenAIConfig配置类

package com.demo.aitest;

import jakarta.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.ai.openai.OpenAiChatClient;
import org.springframework.ai.openai.OpenAiChatOptions;
import org.springframework.ai.openai.api.OpenAiApi;

@Configuration
public class OpenAIConfig {

    @Value("${spring.ai.openai.api-key}")
    private String apiKey;

    @Value("${spring.ai.openai.base-url}")
    private String baseUrl;

    @Value("${spring.ai.openai.chat.options.model}")
    private String model;

    @Bean
    public OpenAiChatClient openAiChatClient() {
        OpenAiApi openAiApi = new OpenAiApi(baseUrl, apiKey);
        OpenAiChatOptions options = new OpenAiChatOptions();
        options.setModel(model);
        options.setFrequencyPenalty(0.0f);
        return new OpenAiChatClient(openAiApi, options);
    }

}

ChatController配置类

package com.demo.aitest;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.ai.chat.ChatClient;
import org.springframework.ai.chat.ChatResponse;
import org.springframework.ai.chat.prompt.Prompt;
import org.springframework.web.bind.annotation.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@RestController
public class ChatController {
    private static final Logger logger = LoggerFactory.getLogger(ChatController.class);

    @Value("${spring.ai.openai.base-url}")
    private String baseUrl;

    @Value("${spring.ai.openai.api-key}")
    private String apiKey;

    @Value("${spring.ai.openai.chat.options.model}")
    private String model;

    @Autowired
    private ChatClient chatClient;

    @GetMapping("/chat")
    public String chat(@RequestParam String message) {
        try {
            logger.info("OpenAI Request Info:");
            logger.info("Base URL: {}", baseUrl);
            logger.info("API Key: {}", apiKey);
            logger.info("Model: {}", model);
            logger.info("Request URL: {}/chat/completions", baseUrl);
            logger.info("Request message: {}", message);
            
            ChatResponse response = chatClient.call(new Prompt(message));
            String content = response.getResult().getOutput().getContent();
            
            logger.info("Response received successfully");
            return content;
            
        } catch (Exception e) {
            logger.error("Error processing chat request: {}", e.getMessage());
            throw e;
        }
    }
}

application.yml配置如下

spring:
  ai:
    openai:
      api-key: Your_APP_KEY   #替换成自己的API KEY
      base-url: https://wishub-x1.ctyun.cn
      chat:
        options:
          model: 4bd107bff85941239e27b1509eccfe98. #替换成自己要用的模型

pom.xml依赖如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.demo</groupId>
    <artifactId>aitest</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>aitest</name>
    <description>aitest</description>

    <properties>
        <java.version>17</java.version>
        <spring-ai.version>0.8.1</spring-ai.version>
        <spring-cloud.version>2023.0.0</spring-cloud.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.ai</groupId>
            <artifactId>spring-ai-openai-spring-boot-starter</artifactId>
            <version>${spring-ai.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-function-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.ai</groupId>
                <artifactId>spring-ai-bom</artifactId>
                <version>${spring-ai.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
    <repositories>
        <repository>
            <id>spring-snapshots</id>
            <name>Spring Snapshots</name>
            <url>https://repo.spring.io/snapshot</url>
            <releases>
                <enabled>false</enabled>
            </releases>
        </repository>
        <repository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>
</project>

访问地址：http://localhost:8080/chat?message=Hello

常见报错

触发限流

报错信息为：[openai_api_compatible]Error:APIrequestfailedwithstatuscode429:{"code":700007,"detail":"Deepseek模型并发已达上限","message":"DEEP_SEEK_MODEL_CONCURRENCY_LIMIT","error":{"code":"700007","message":"Deepseek模型并发已达上限","type":"DEEP_SEEK_MODEL_CONCURRENCY_LIMIT"}}

问题原因：大模型并发是大模型的并发能力，和用户、tokens是否用完都无关，只和机器负载能力有关。目前单个模型有固定的并发度，这是所有ds模型共享的并发度，只能支持固定个请求同时访问。
解决方案：等一段时间再调用，或增加重试机制。

同步调用响应耗时长

问题原因：大模型API同步耗时长是因为其推理过程复杂，需要逐个生成token并拼接结果，且在处理长文本或复杂问题时，等待完整结果返回的时间过长。而流式API可以在推理过程中逐步返回结果，用户无需等待全部结果生成，从而提升交互效率。
解决方案：改为流式调用。

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

查看所有产品

一站式智算服务平台

一站式智算服务平台

天翼云官网获取模型调用APP key等信息

模型API终端请求地址

请求地址

通信协议

请求方法

获取模型调用APP key

获取调用的模型名称

手动验证key的可用性

调用代码示例

同步调用

调用主程序

prompt模型代码示例

流式调用

Langchain调用

Dify

Spring AI

核心代码如下：

常见报错

触发限流

同步调用响应耗时长

活动

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

查看所有产品

一站式智算服务平台

一站式智算服务平台

天翼云官网获取模型调用APP key等信息

模型API终端请求地址

请求地址

通信协议

请求方法

获取模型调用APP key

获取调用的模型名称

手动验证key的可用性

调用代码示例

同步调用

调用主程序

prompt模型代码示例

流式调用

Langchain调用

Dify

Spring AI

核心代码如下：

常见报错

触发限流

同步调用响应耗时长