Qwen智能体qwen_agent与Assistant功能初探
一、Qwen智能体框架概述
Qwen（通义千问）智能体框架是阿里云推出的新一代AI智能体开发平台，其核心模块qwen_agent.agent提供了一套完整的智能体构建解决方案。该框架通过模块化设计，将LLM（大语言模型）能力与专业工具链相结合，支持开发者快速构建具备复杂问题处理能力的AI助手。
二、qwen_agent核心架构
1. 分层设计理念
- 基础层：封装了Qwen大模型的基础推理能力
- 工具层：集成搜索引擎、代码执行、文档解析等工具
- 控制层：实现任务规划、工具调度和结果验证
- 交互层：提供多轮对话管理和上下文保持
2. 关键技术特性
from qwen_agent.agent import Agent
# 典型初始化示例
agent = Agent(
llm="qwen-max", # 支持不同规模的模型选择
tools=['web_search', 'code_interpreter'], # 可插拔工具集
system_message="你是一个专业数据分析助手" # 角色设定
)
三、Assistant功能详解
1. 多模态任务处理
- 文档解析：支持PDF/Word/Excel等格式的深度理解
- 表格处理：自动提取结构化数据并执行分析
- 图像理解：结合Qwen-VL多模态模型实现图文交互
2. 复杂任务分解，智能选择不同function。
@register_tool('exc_sql')
class ExcSQLTool(BaseTool):
bot = Assistant(
llm=llm_cfg,
name='电商销售助手',
description='订单查询与订单分析',
system_message=system_prompt,
function_list=['exc_sql'], # 只传工具名字符串
)
四、典型应用场景
- 智能客服：处理80%以上标准咨询，转人工率降低60%
- 数据分析：将自然语言需求自动转化为SQL/Python代码
- 知识管理：构建企业专属知识库问答系统
五、完整代码示例
本示例代码是打通我自己笔记本上的虚机和自建的oracle数据库，模拟了一个电商销售助手的功能.
数据库orders表如下：
Python代码如下：
# 1.导入必要的库
import os
import asyncio
from typing import Optional
import dashscope
from qwen_agent.agents import Assistant
from qwen_agent.gui import WebUI
import pandas as pd
from sqlalchemy import create_engine,text
from sqlalchemy.exc import SQLAlchemyError
from qwen_agent.tools.base import BaseTool, register_tool
#测试oracle连接
db_user = "dbtest"
db_password = "test"
db_host = "192.168.43.11:1521"
service_name = "FREEPDB1"
# 1.连接字符串
oracle_connection_string = f"oracle+cx_oracle://{db_user}:{db_password}@{db_host}/?service_name={service_name}"
# 创建SQLAlchemy引擎
try:
engine = create_engine(oracle_connection_string)
connection = engine.connect()
print("数据库连接成功！")
result =connection.execute(text("SELECT * FROM products FETCH FIRST 5 ROWS ONLY"))
# 转换为DataFrame
df = pd.DataFrame(result.fetchall(), columns=result.keys())
print(df) # 以表格形式打印
connection.close()
except SQLAlchemyError as e:
print(f"数据库连接失败: {e}")
数据库测试结果：
数据库连接成功！
product_id category_id product_name product_desc price stock sales \
0 10001 100101 智能手机X 最新款智能手机，6.5英寸屏幕 5999 100 50
1 10002 100101 智能手机Y 高性价比手机，6.1英寸屏幕 2999 200 120
2 10003 100101 智能手机Z 入门级智能手机，5.8英寸屏幕 1999 150 80
3 10004 100102 轻薄笔记本 超薄设计，高性能，13.3英寸 8999 50 30
4 10005 100102 游戏笔记本 高性能游戏本，15.6英寸 12999 30 15
create_time update_time status
0 2022-01-01 2022-06-01 1
1 2022-01-05 2022-06-05 1
2 2022-01-10 2022-06-10 1
3 2022-01-15 2022-06-15 1
4 2022-01-20 2022-06-20 1
主体代码：
# 配置 DashScope
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY', '') # 从环境变量获取 API Key
dashscope.timeout = 30 # 设置超时时间为 30 秒
# ====== 门票助手 system prompt 和函数描述 ======
system_prompt = """我是电商销售助手，以下是关于电商销售订单表orders,货品表order_items和客户表users的相关字段，我可能会编写对应的SQL，对数据进行查询
-- 电商销售订单表
CREATE TABLE "ORDERS"
( "ORDER_ID" NUMBER,
"ORDER_NO" VARCHAR2(50) NOT NULL ENABLE,
"USER_ID" NUMBER NOT NULL ENABLE,
"ADDRESS_ID" NUMBER NOT NULL ENABLE,
"TOTAL_AMOUNT" NUMBER(10,2) NOT NULL ENABLE,
"PAYMENT_AMOUNT" NUMBER(10,2) NOT NULL ENABLE,
"FREIGHT_AMOUNT" NUMBER(10,2) DEFAULT 0,
"ORDER_STATUS" NUMBER(1,0) DEFAULT 0,
"PAYMENT_TIME" DATE,
"DELIVERY_TIME" DATE,
"RECEIVE_TIME" DATE,
"CREATE_TIME" DATE DEFAULT SYSDATE,
PRIMARY KEY ("ORDER_ID")
)
--货品表
CREATE TABLE "ORDER_ITEMS"
( "ITEM_ID" NUMBER,
"ORDER_ID" NUMBER NOT NULL ENABLE,
"PRODUCT_ID" NUMBER NOT NULL ENABLE,
"PRODUCT_NAME" VARCHAR2(100) NOT NULL ENABLE,
"PRODUCT_IMAGE" VARCHAR2(200),
"PRICE" NUMBER(10,2) NOT NULL ENABLE,
"QUANTITY" NUMBER NOT NULL ENABLE,
"TOTAL_PRICE" NUMBER(10,2) NOT NULL ENABLE,
PRIMARY KEY ("ITEM_ID")
）
--客户表
CREATE TABLE USERS"
( "USER_ID" NUMBER,
"USERNAME" VARCHAR2(50) NOT NULL ENABLE,
"PASSWORD" VARCHAR2(100) NOT NULL ENABLE,
"EMAIL" VARCHAR2(100) NOT NULL ENABLE,
"PHONE" VARCHAR2(20),
"REGISTER_DATE" DATE DEFAULT SYSDATE,
"STATUS" NUMBER(1,0) DEFAULT 1,
PRIMARY KEY ("USER_ID")
)
"""
# ====== exc_sql 工具类实现 ======
@register_tool('exc_sql')
class ExcSQLTool(BaseTool):
"""
SQL查询工具，执行传入的SQL语句并返回结果。
功能：
- 执行SQL查询并返回结果
- 支持数据库连接和错误处理
- 限制返回结果数量为10行
参数：
- sql_input: SQL查询语句
- database: 数据库名称（可选，默认为'ubr'）
返回：
- 查询结果（最多10行，以markdown格式返回）
"""
description = '对于生成的SQL，进行SQL查询'
parameters = [{
'name': 'sql_input',
'type': 'string',
'description': '生成的SQL语句',
'required': True
}]
def call(self, params: str, **kwargs) -> str:
"""
执行SQL查询并返回结果
参数：
- params: JSON格式的参数字符串，包含sql_input和可选的database
返回：
- 查询结果（最多10行，以markdown格式返回）
"""
import json
args = json.loads(params)
sql_input = args['sql_input']
database = args.get('database', 'ubr')
# 创建数据库连接
engine = create_engine(oracle_connection_string)
try:
df = pd.read_sql(sql_input, engine)
# 返回前10行，防止数据过多
return df.head(10).to_markdown(index=False)
except Exception as e:
return f"SQL执行出错: {str(e)}"
# ====== 初始化门票助手服务 ======
def init_agent_service():
"""
初始化电商销售助手服务
功能：
- 配置通义千问模型参数
- 初始化助手实例
- 注册SQL查询工具
- 设置助手名称和描述
返回：
- 配置好的助手实例
"""
llm_cfg = {
'model': 'qwen-turbo-2025-04-28',
'timeout': 30,
'retry_count': 3,
}
try:
bot = Assistant(
llm=llm_cfg,
name='电商销售助手',
description='订单查询与订单分析',
system_message=system_prompt,
function_list=['exc_sql'], # 只传工具名字符串
)
print("助手初始化成功！")
return bot
except Exception as e:
print(f"助手初始化失败: {str(e)}")
raise
def app_tui():
"""
终端交互模式
功能：
- 支持连续对话
- 支持文件输入
- 实时响应用户查询
- 异常处理和错误提示
"""
try:
# 初始化助手
bot = init_agent_service()
# 对话历史
messages = []
while True:
try:
# 获取用户输入
query = input('user question: ')
# 获取可选的文件输入
file = input('file url (press enter if no file): ').strip()
# 输入验证
if not query:
print('user question cannot be empty！')
continue
# 构建消息
if not file:
messages.append({'role': 'user', 'content': query})
else:
messages.append({'role': 'user', 'content': [{'text': query}, {'file': file}]})
print("正在处理您的请求...")
# 运行助手并处理响应
response = []
for response in bot.run(messages):
print('bot response:', response)
messages.extend(response)
except Exception as e:
print(f"处理请求时出错: {str(e)}")
print("请重试或输入新的问题")
except Exception as e:
print(f"启动终端模式失败: {str(e)}")
def app_gui():
"""
图形界面模式
功能：
- 基于WebUI提供图形界面
- 预设常用查询建议
- 支持实时对话
- 异常处理和错误提示
"""
try:
print("正在启动 Web 界面...")
# 初始化助手
bot = init_agent_service()
# 配置聊天界面，列举3个典型门票查询问题
chatbot_config = {
'prompt.suggestions': [
'帮我查看订单金额排名',
'帮我从多到小列出前5位购买最多的货物',
'请帮我查询客户订单金额排名前5的订单信息，包括订单号，订的货物清单以及客户姓名',
'请帮我查询客户订单金额排名前5的订单信息，包括订单号，订的货物清单以及客户姓名，请按订单为单位显示，一张订单有多个货物的，将货物名称罗列出来'
]
}
print("Web 界面准备就绪，正在启动服务...")
# 启动 Web 界面
WebUI(
bot,
chatbot_config=chatbot_config
).run()
except Exception as e:
print(f"启动 Web 界面失败: {str(e)}")
print("请检查网络连接和 API Key 配置")
if __name__ == '__main__':
# 运行模式选择
app_gui() # 图形界面模式（默认）
默认运行图形界面
正在启动 Web 界面...
助手初始化成功！
Web 界面准备就绪，正在启动服务...
* Running on local URL: http://127.0.0.1:7860
点开链接，初始化界面如下
可以看到它的自动纠错功能
友情提示你的查询可能有问题：
如此复杂的SQL，它能生成！
SELECT o.ORDER_NO, LISTAGG(oi.PRODUCT_NAME, ', ') WITHIN GROUP (ORDER BY oi.ITEM_ID) AS product_list, u.USERNAME, o.TOTAL_AMOUNT FROM ORDERS o JOIN ORDER_ITEMS oi ON o.ORDER_ID = oi.ORDER_ID JOIN USERS u ON o.USER_ID = u.USER_ID GROUP BY o.ORDER_NO, u.USERNAME, o.TOTAL_AMOUNT ORDER BY o.TOTAL_AMOUNT DESC FETCH FIRST 5 ROWS ONLY"
六、Qwen智能体总结
优点：模块化架构支持灵活扩展，集成多工具链（搜索/代码/文档处理），具备复杂任务分解和动态学习能力，企业级场景适配性强，支持多模态交互。
缺点：工具配置复杂度较高，长任务处理效率依赖模型性能，定制开发需技术基础，实时性较专用系统稍弱。
同类产品：
- OpenAI的Assistant API（功能相似但更封闭）
- LangChain（更开发者导向但集成度低）
- Microsoft Copilot Studio（企业集成强但灵活性弱）
发展动向：
- 强化多智能体协作与竞合机制（目前google正在推A2A，Agent to Agent)
- 开源生态建设（对标LangChain）
*** Qwen在平衡灵活性与开箱即用体验上优势明显，正快速追赶国际头部产品。***