最佳实践
更新时间：
本文提供Assistant API的几个最佳实践，帮助您快速入门并开发您自己的业务应用。
前提条件
设置您的API-KEY，替换YOUR_DASHCOPE_API_KEY为您自己的API key
export DASHSCOPE_API_KEY=YOUR_DASHSCOPE_API_KEY1、调用Assistant API夸克搜索回答问题
在使用基础通义千问模型基础上，可以通过Assistant API调用夸克搜索，实现更好地回答问题。下面给出了一个使用Assistant API调用夸克搜索的示例。
from http import HTTPStatus
import json
import sys
from typing import Any
import dashscope
def validate_request_result(response: Any):
"""Verify the response of result.
If the request is success, the status_code is HTTPStatus.OK.
Otherwise, the request is failed, you can get error code and error message.
Args:
response (Any): The api response.
"""
if response.status_code != HTTPStatus.OK:
print('Get run failed, status code: %s, code: %s, message: %s'%(response.status_code, response.code, response.message))
sys.exit(response.status_code)
else:
print(response)
# create assistant with information
assistant = dashscope.Assistants.create(
model='qwen-max',
name='smart helper',
description='A tool helper.',
instructions='You are a helpful assistant. When asked a question, use tools wherever possible.', # noqa E501
tools=[{
'type': 'quark_search'
}],
)
validate_request_result(assistant)
# create a thread.
thread = dashscope.Threads.create()
# check result is success.
validate_request_result(thread)
# create a message.
message = dashscope.Messages.create(thread.id, content="今日北京天气")
validate_request_result(message)
# create a new run to run message
run = dashscope.Runs.create(thread.id, assistant_id=assistant.id)
validate_request_result(message)
# wait for run completed
run = dashscope.Runs.wait(run.id, thread_id=thread.id)
validate_request_result(run)
# get the thread messages.
msgs = dashscope.Messages.list(thread.id)
validate_request_result(msgs)
# print the messages
print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4, ensure_ascii=False))
最终可以得到回复（去除前面打印的结果的running信息）：
{
"data": [
{
"assistant_id": "asst_6dc44244-b3f0-458c-8396-3db71023bb8e",
"content": [
{
"text": {
"value": "2024年4月1日，北京市的天气情况如下：\n\n- 白天：阴天，湿度为34%，最低气温为10.0摄氏度，最高气温为16.0摄氏度，风向为东南风，风力1级。\n- 夜间：多云，转为北风，风力同样为1级。\n\n总体来说，今天北京气温较低且有阴云，建议出门携带外套，并关注晚上的天气变化。由于湿度较大，体感温度可能会更低，要注意保暖。同时，空气质量良好，空气质量指数为66。"
},
"type": "text"
}
],
"created_at": 1711937044469,
"file_ids": [],
"from": "text",
"id": "message_2a01c9f1-8d2e-43ef-a7bb-a5a70a84ecbd",
"metadata": {},
"name": "",
"object": "thread.message",
"plugin_call": {},
"role": "assistant",
"run_id": "run_700f67ff-5a48-4ada-a35e-c2450670f164",
"status": {},
"thread_id": "thread_fc6a8f26-8900-43a8-a61e-e4ff6d11216b",
"tool_calls": []
},
{
"assistant_id": "",
"content": [
{
"text": {
"value": "今日北京天气"
},
"type": "text"
}
],
"created_at": 1711937030391,
"file_ids": [],
"from": "",
"id": "message_ee8a95c4-8254-43d3-9dd8-bbf5769ec87f",
"metadata": {},
"name": "",
"object": "thread.message",
"plugin_call": {},
"role": "user",
"run_id": "",
"status": {},
"thread_id": "thread_fc6a8f26-8900-43a8-a61e-e4ff6d11216b",
"tool_calls": []
}
],
"first_id": "message_2a01c9f1-8d2e-43ef-a7bb-a5a70a84ecbd",
"has_more": false,
"last_id": "message_ee8a95c4-8254-43d3-9dd8-bbf5769ec87f",
"object": "list",
"request_id": "a891e571-f36b-9630-98fb-a6577a79d93f",
"status_code": 200
}2、Agent智能体构建
在构建具体的Agent应用的过程中，用户可以自定义多种AgentModule类来完成一系列子任务，并根据需要实现复杂、灵活的应用编排。以下提供一种示范性的编排实例。
Step 1: 创建Agent类
首先，定义基类AgentModule，及其子类APIAssistantAgent，对各个API接口顺序实现封装。
用户通过name、description、model、instructions以及tools、functions等参数创建APIAssistantAgent。
用户通过functions参数为APIAssistantAgent传入function-call可能调用的函数实体，在APIAssistantAgent.query()中自动识别、调用函数并向AssistantAPI上传结果。
from typing import Dict, List
import json
import dashscope
class AgentModule():
def __init__(self, name: str, description: str):
self.name = name
self.description = description
def __call__(self, query:str):
return self.query(query)
def query(self, query:str):
pass
class APIAssistantAgent(AgentModule):
'''
Assistant with LLM capability, which is built with Assistant API.
'''
def __init__(self, name:str="",
description:str="",
model:str="",
instructions:str="",
tools: List=[],
functions: Dict={},
assistant_id: str=None):
super().__init__(name, description)
# create a new assistant
if not assistant_id:
self.assistant = dashscope.Assistants.create(
model=model,
name=name,
description=description,
instructions=instructions,
tools=tools,
)
verify_status_code(self.assistant)
else: # reload the assistant via id
self.assistant = dashscope.Assistants.retrieve(assistant_id = assistant_id)
self.name, self.description = self.assistant.name, self.assistant.description
self.assistant_id = self.assistant.id
self.funcs = functions
self.thread = dashscope.Threads.create()
verify_status_code(self.thread)
def _forward_and_submit_outputs(self, run_status):
'''get the function name, run and return the function output to the server'''
func_obj = run_status.required_action.submit_tool_outputs.tool_calls[0].function
func_name = func_obj.name
arguments = json.loads(func_obj.arguments)
tool_outputs = [{
'tool_call_id':
run_status.required_action.submit_tool_outputs.tool_calls[0].id,
'output': json.dumps(self.funcs[func_name](**arguments))
}]
run_status = dashscope.Runs.submit_tool_outputs(run_status.id,
thread_id=self.thread.id,
tool_outputs=tool_outputs)
return
def query(self, query:str):
'''
query: string, the query string to the assistant. e.g. Who is the Jack Chou ?
'''
message = dashscope.Messages.create(self.thread.id, content=query)
message_run = dashscope.Runs.create(self.thread.id, assistant_id=self.assistant.id)
run_status = dashscope.Runs.wait(message_run.id, thread_id=self.thread.id)
if run_status.required_action:
self._forward_and_submit_outputs(run_status)
run_status = dashscope.Runs.wait(run_status.id, thread_id=self.thread.id)
msgs = dashscope.Messages.list(self.thread.id)
answer = json.loads(json.dumps(msgs, default=lambda o: o.__dict__))['data'][0]['content'][0]['text']['value']
return answerStep 2: 编排多个Agent实现特定应用
用户可以将多个Agent进行编排，实现特定的应用。
编排后的应用本身也作为一个Agent类，以供更上层的应用编排。
以下提供一个智能财经助手的例子，通过顺序编排新闻检索专家与新闻分析专家来进行实现。
class SequentialAgent(AgentModule):
'''
用于顺序编排的AgentModuel
'''
def __init__(self, name: str, description: str, sequential: List,):
'''
sequential: List[assistant_object or function]
'''
super().__init__(name, description)
self.sequential = sequential
def query(self, query:str):
msg = query
for node in self.sequential:
msg = node(msg)
return msg
news_args = {"name":"新闻检索专家",
"description":"新闻检索专家",
"instructions":"你是一个新闻检索专家。请根据用户提供的信息，适当使用工具查询与其相关的新闻资讯。请务必查询真实的新闻。",
"model":"qwen-max",
"tools":[{
'type': 'quark_search'
}]}
news_assistant = APIAssistantAgent(**news_args)
analytic_args = {"name":"公司新闻分析专家",
"description":"一个公司新闻分析专家",
"instructions":"你是一个公司新闻分析专家，请根据用户提供的新闻，对相关公司进行分析并给出适当的详细的投资建议。",
"model":"qwen-max",
}
analytic_assistant = APIAssistantAgent(**analytic_args)
finance_assistant = SequentialAgent(name="财经助手", description="一个财经助手", sequential = [news_assistant, analytic_assistant])
print(finance_assistant.name)
print(finance_assistant("你对SpaceX怎么看?"))执行结果
基于以上信息，我们可以分析SpaceX的业务进展、技术实力和未来潜力：
发射成功率与频率：尽管在年初遇到了挑战，但SpaceX迅速调整并持续完成多次成功发射任务，显示了其强大的技术研发实力和运营能力。频繁且成功的发射活动有助于提高公司的市场占有率和行业地位。
星链计划进展：星链卫星部署的顺利推进表明SpaceX正在逐步实现其大规模卫星互联网服务的目标，这将为公司带来新的收入来源，并可能颠覆全球通信行业的格局。一旦全面商业化运作，将极大提升公司估值。
载人航天成就：Crew Dragon Demo-1的成功首飞不仅增强了商业航天市场的信心，也为NASA及国际合作伙伴提供了可靠的载人运输方案，确保了SpaceX在政府合同中占据有利位置。
风险承受能力与创新精神：马斯克对SpaceX存活几率的评估反映了公司早期面临的高风险环境，但也彰显了管理层面对未知敢于探索的决心，这种精神在高科技行业中尤为宝贵，有助于吸引投资者关注长期价值。
投资建议： 对于投资者而言，SpaceX展现了在多个前沿领域的领导地位和技术优势，尤其是星链计划的商业化进程值得密切关注。然而，考虑到其业务性质的复杂性和高度不确定性，以及可能面临的法规、竞争和技术难题，投资者应保持审慎乐观态度。理想的投资策略可能是多元化投资组合，同时结合定期评估SpaceX的项目进度、财务状况和市场反馈进行动态调整。在等待更多关于盈利模式、营收增长和市场份额等实质性数据的同时，长期投资者可适当布局以期分享公司在新兴太空经济中的潜在收益。
反馈
- 本页导读 （1）
文档反馈