背景
公司想对接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: str
default_model: str = "qwen-plus"
timeout: int = 30
max_retries: int = 3
class GPTChatResponse(BaseModel):
content: str
tool_calls: Optional[List[Dict]] = None
def 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'] = temperature
for 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:
raise
if __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: str
default_model: str = "qwen-plus"
timeout: int = 30
max_retries: int = 3
class GPTChatResponse(BaseModel):
content: str
tool_calls: Optional[List[Dict]] = None
def 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'] = tools
content_list = []
tool_calls = None
try:
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]':
continue
try:
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):
break
except json.JSONDecodeError:
log.warning(f"Failed to decode chunk: {line}")
continue
except requests.exceptions.RequestException as e:
log.error(f"Stream request failed: {str(e)}")
raise
return 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)
结束
我们成功构建了基于阿里云DashScope的高效智能对话系统。这套方案不仅适用于客服场景，也可扩展应用于智能助手、内容生成等多种AI应用场景。DashScope服务的稳定性和易用性为中小企业快速部署AI能力提供了可靠选择。
