当前位置：首页 > news >正文

基于阿里云DashScope API构建智能对话指南

news 来源：原创 2025/5/23 5:56:53

背景

公司想对接AI智能体，用于客服系统，经过调研和实施，觉得DashScope 符合需求。
阿里云推出的DashScope灵积模型服务为开发者提供了便捷高效的大模型接入方案。本文将详细介绍如何基于DashScope API构建一个功能完善的智能对话系统，包含流式对话、工具调用等高级特性。

项目背景与技术选型

我们的项目目标是构建一个企业级智能客服系统，需要满足以下核心需求：

支持多轮自然语言对话
实现低延迟的流式响应
可扩展的工具调用能力
稳定的生产环境部署

经过技术评估，我们选择了阿里云DashScope服务，主要基于以下优势：

模型多样性：提供QWEN系列等多种大语言模型
API兼容性：兼容OpenAI API格式，降低迁移成本
性能保障：阿里云基础设施确保服务稳定性
成本效益：相比自建模型集群更具性价比

模型地址：https://help.aliyun.com/zh/model-studio/videos/yi-large-quick-start

在这里插入图片描述

核心代码实现与优化

基础对话功能实现

我们首先实现了基础的对话功能模块，这是整个系统的核心：


import json
from typing import List, Dict, Optional
import requests
from pydantic import BaseModel
from utils.LogHandler import log# 配置管理使用Pydantic模型，便于验证和文档化
class DashScopeConfig(BaseModel):base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"api_key: strdefault_model: str = "qwen-plus"timeout: int = 30max_retries: int = 3class GPTChatResponse(BaseModel):content: strtool_calls: Optional[List[Dict]] = Nonedef gpt_chat(messages: List[Dict[str, str]],config: DashScopeConfig,model: Optional[str] = None,temperature: Optional[float] = None,max_tokens: Optional[int] = 512
) -> str:"""标准对话API实现:param messages: 对话消息列表:param config: 服务配置:param model: 指定模型，默认使用配置中的default_model:param temperature: 生成多样性控制:param max_tokens: 最大输出token数:return: 模型生成的文本内容"""headers = {'Content-Type': 'application/json','Authorization': f'Bearer {config.api_key}'}payload = {'model': model or config.default_model,'messages': messages,'max_tokens': max_tokens,}if temperature is not None:payload['temperature'] = temperaturefor attempt in range(config.max_retries):try:resp = requests.post(f"{config.base_url}/chat/completions",headers=headers,json=payload,timeout=config.timeout)resp.raise_for_status()json_data = resp.json()if choices := json_data.get('choices'):if len(choices) > 0:return choices[0]['message']['content']raise ValueError("No valid response from model")except requests.exceptions.RequestException as e:log.error(f"Attempt  {attempt + 1} failed: {str(e)}")if attempt == config.max_retries - 1:raiseif  __name__ == "__main__":config = DashScopeConfig(api_key="sk-xxxxxx")messages = [{"role": "system", "content": "你是一名专业客服人员"},{"role": "user", "content": "怎么处理客户争吵问题"}]response = gpt_chat(messages, config)print(response)

在这里插入图片描述

流式对话高级实现

为提升用户体验，我们实现了流式对话功能，并进行了多项优化：


import json
from typing import List, Dict, Optional
import requests
from pydantic import BaseModel
from utils.LogHandler import log# 配置管理使用Pydantic模型，便于验证和文档化
class DashScopeConfig(BaseModel):base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"api_key: strdefault_model: str = "qwen-plus"timeout: int = 30max_retries: int = 3class GPTChatResponse(BaseModel):content: strtool_calls: Optional[List[Dict]] = Nonedef gpt_stream_chat(messages: List[Dict[str, str]],config: DashScopeConfig,model: Optional[str] = None,tools: Optional[List[Dict]] = None,on_content: Optional[callable] = None
) -> GPTChatResponse:"""流式对话实现，支持实时内容处理和工具调用:param messages: 对话消息列表:param config: 服务配置:param model: 指定模型:param tools: 可用工具列表:param on_content: 内容回调函数:return: GPTChatResponse对象"""headers = {'Content-Type': 'application/json','Authorization': f'Bearer {config.api_key}'}payload = {'model': model or config.default_model,'messages': messages,'stream': True}if tools:payload['tools'] = toolscontent_list = []tool_calls = Nonetry:resp = requests.post(f"{config.base_url}/chat/completions",headers=headers,json=payload,stream=True,timeout=config.timeout)resp.raise_for_status()for line in resp.iter_lines():if not line or line == b'data: [DONE]':continuetry:chunk = json.loads(line.decode('utf-8')[6:])delta = chunk['choices'][0]['delta']# 处理推理过程内容if content := delta.get('reasoning_content'):if on_content:on_content(content, 'reasoning')content_list.append(content)# 处理最终回复内容if content := delta.get('content'):if on_content:on_content(content, 'content')content_list.append(content)# 处理工具调用if 'tool_calls' in delta:tool_calls = delta['tool_calls']if tool_calls and 'name' in str(tool_calls):breakexcept json.JSONDecodeError:log.warning(f"Failed  to decode chunk: {line}")continueexcept requests.exceptions.RequestException as e:log.error(f"Stream  request failed: {str(e)}")raisereturn GPTChatResponse(content=''.join(content_list), tool_calls=tool_calls)def handle_stream_content(content: str, content_type: str):print(content, end='', flush=True)  # 实时输出，不换行if __name__ == "__main__":config = DashScopeConfig(api_key="sk-xxxxxxxxxxxxxxxxxxxxxx")messages = [{"role": "system", "content": "你是一名专业客服人员"},{"role": "user", "content": "你好？"}]response = gpt_stream_chat(messages, config, on_content=handle_stream_content)# print(response.content)