Assistant API 调用示例（下线中）

import json
import sys
from http import HTTPStatus

from dashscope import Assistants, Messages, Runs, Threads

def create_assistant():
    # create assistant with information
    assistant = Assistants.create(
        model='qwen-max',  # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        name='smart helper',
        description='A tool helper.',
        instructions='You are a helpful assistant.',  # noqa E501
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        print('Failed: ')
        print(res)
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant()
    print(assistant)
    verify_status_code(assistant)

    # create thread.
    thread = Threads.create(
        messages=[{
            'role': 'user',
            'content': '如何做出美味的牛肉炖土豆？'
        }])
    print(thread)
    verify_status_code(thread)

    # create run
    run = Runs.create(thread.id, assistant_id=assistant.id)
    print(run)
    verify_status_code(run)
    # wait for run completed or requires_action
    run_status = Runs.wait(run.id, thread_id=thread.id)
    print(run_status)

    # get the thread messages.
    msgs = Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, ensure_ascii=False, default=lambda o: o.__dict__, sort_keys=True, indent=4))

import json
import sys
from http import HTTPStatus

from dashscope import Assistants, Messages, Runs, Threads

def create_assistant():
    # create assistant with information
    assistant = Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        model='qwen-max',
        name='smart helper',
        description='A tool helper.',
        instructions='You are a helpful assistant.',  # noqa E501
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        print('Failed: ')
        print(res)
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant()
    print(assistant)
    verify_status_code(assistant)

    # create thread.
    thread = Threads.create(
        messages=[{
            'role': 'user',
            'content': '如何做出美味的牛肉炖土豆？'
        }])
    print(thread)
    verify_status_code(thread)

    # create run with stream.
    run_iterator = Runs.create(thread.id,
                      assistant_id=assistant.id,
                      stream=True)
    # iterator over the event and message.
    for event, msg in run_iterator:
        print(event)
        print(msg)
    # get the thread messages.
    msgs = Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, ensure_ascii=False, default=lambda o: o.__dict__, sort_keys=True, indent=4))

import json
import sys
from http import HTTPStatus

from dashscope import Assistants, Messages, Runs, Steps, Threads

def create_assistant_call_function():
    # create assistant with information
    assistant = Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        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': 'function',
            'function': {
                'name': 'big_add',
                'description': 'Add to number',
                'parameters': {
                    'type': 'object',
                    'properties': {
                        'left': {
                            'type': 'integer',
                            'description': 'The left operator'
                        },
                        'right': {
                            'type': 'integer',
                            'description': 'The right operator.'
                        }
                    },
                    'required': ['left', 'right']
                }
            }
        }],
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        print('Failed: ')
        print(res)
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant_call_function()
    print(assistant)
    verify_status_code(assistant)

    # create thread.
    thread = Threads.create()
    print(thread)
    verify_status_code(thread)

    # create a message.
    message = Messages.create(thread.id, content='Add 87787 to 788988737.')
    print(message)
    verify_status_code(message)

    # create a new run to run message

    message_run = Runs.create(thread.id, assistant_id=assistant.id)
    print(message_run)
    verify_status_code(message_run)

    # get run statue
    run_status = Runs.get(message_run.id, thread_id=thread.id)
    print(run_status)
    verify_status_code(run_status)

    # wait for run completed or requires_action
    run_status = Runs.wait(message_run.id, thread_id=thread.id)
    print(run_status)

    # if prompt input tool result, submit tool result.
    # should call big_add
    if run_status.required_action:
        tool_outputs = [{
            'output':
            '789076524'
        }]
        run = Runs.submit_tool_outputs(message_run.id,
                                       thread_id=thread.id,
                                       tool_outputs=tool_outputs)
        print(run)
        verify_status_code(run)

        # wait for run completed or requires_action
        run_status = Runs.wait(message_run.id, thread_id=thread.id)
        print(run_status)
        verify_status_code(run_status)

    run_steps = Steps.list(run.id,  thread_id=thread.id)
    print(run_steps)
    verify_status_code(run_steps)

    # get the thread messages.
    msgs = Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, ensure_ascii=False, default=lambda o: o.__dict__, sort_keys=True, indent=4))

# yapf: disable
import json
import sys
from http import HTTPStatus

import dashscope
from dashscope import Assistants, Messages, Runs, Steps, Threads

def create_assistant_call_function():
    # create assistant with information
    assistant = Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        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': 'function',
            'function': {
                'name': 'big_add',
                'description': 'Add to number',
                'parameters': {
                    'type': 'object',
                    'properties': {
                        'left': {
                            'type': 'integer',
                            'description': 'The left operator'
                        },
                        'right': {
                            'type': 'integer',
                            'description': 'The right operator.'
                        }
                    },
                    'required': ['left', 'right']
                }
            }
        }],
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        print('Failed: ')
        print(res)
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant_call_function()
    print(assistant)
    verify_status_code(assistant)

    # create run
    run_iterator = Runs.create_thread_and_run(thread={'messages': [{
            'role': 'user',
            'content': 'What is transformer? Explain it in simple terms.'
        }]}, assistant_id=assistant.id, stream=True)
    thread = None
    for event, run in run_iterator:
        print(event)
        print(run)
        if event == 'thread.created':
            thread = run
    verify_status_code(run)

    run_status = Runs.get(run.id, thread_id=thread.id)
    print(run_status)
    verify_status_code(run_status)

    # list run steps
    run_steps = Steps.list(run.id, thread_id=thread.id)
    print(run_steps)
    verify_status_code(run_steps)

    # create a message.
    message = Messages.create(thread.id, content='Add 87787 to 788988737.')
    print(message)
    verify_status_code(message)

    # create a new run to run message
    responses = Runs.create(thread.id, assistant_id=assistant.id, stream=True)
    for event, message_run in responses:
        print(event)
        print(message_run)
    verify_status_code(message_run)

    # get run statue
    run_status = Runs.get(message_run.id, thread_id=thread.id)
    print(run_status)
    verify_status_code(run_status)

    # if prompt input tool result, submit tool result.
    # should call big_add
    if run_status.required_action:
        tool_outputs = [{
            'output':
            '789076524'
        }]
        # submit output with stream.
        responses = Runs.submit_tool_outputs(message_run.id,
                                             thread_id=thread.id,
                                             tool_outputs=tool_outputs,
                                             stream=True)
        for event, run in responses:
            print(event)
            print(run)
        verify_status_code(run)

    run_status = Runs.get(run.id, thread_id=thread.id)
    print(run_status)
    verify_status_code(run_status)

    run_steps = Steps.list(run.id,  thread_id=thread.id)

    print(run_steps)
    verify_status_code(run_steps)

    # get the thread messages.
    msgs = Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, ensure_ascii=False, default=lambda o: o.__dict__, sort_keys=True, indent=4))

import json
import sys
from http import HTTPStatus

import dashscope

def create_assistant():
    # create assistant with information
    assistant = dashscope.Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        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'
        }],
    )
    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant()
    print(assistant)
    verify_status_code(assistant)

    # create a thread.
    thread = dashscope.Threads.create()
    print(thread)
    verify_status_code(thread)

    # create a message.
    message = dashscope.Messages.create(thread.id, content='请帮忙查询今日北京天气？')
    print(message)
    verify_status_code(message)

    # create a new run to run message
    message_run = dashscope.Runs.create(thread.id, assistant_id=assistant.id)
    print(message_run)
    verify_status_code(message_run)

    # wait for run completed
    run = dashscope.Runs.wait(message_run.id, thread_id=thread.id)
    print(run)

    run = dashscope.Runs.get(run.id, thread_id=thread.id)
    print(run)
    verify_status_code(run)

    run_steps = dashscope.Steps.list(run.id, thread_id=thread.id)

    print(run_steps)
    verify_status_code(run_steps)

    # get the thread messages.
    msgs = dashscope.Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4, ensure_ascii=False))

import json
import sys
from http import HTTPStatus

import dashscope

def create_assistant():
    # create assistant with information
    assistant = dashscope.Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        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'
        }],
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        print('Failed: ')
        print(res)
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant()
    print(assistant)
    verify_status_code(assistant)

    # create a thread.
    thread = dashscope.Threads.create()
    print(thread)
    verify_status_code(thread)

    # create a message.
    message = dashscope.Messages.create(thread.id,
                                        content='今天北京天气怎么样？')
    print(message)
    verify_status_code(message)

    # create a new run to run message
    stream_iterator = dashscope.Runs.create(thread.id,
                                        assistant_id=assistant.id,
                                        stream=True)
    for event, msg in stream_iterator:
        print(event)
        print(msg)
    verify_status_code(msg)

    # get run statue
    # run_9fa03862-aa36-4e1c-b2a7-9fdd91cb9a1d
    run = dashscope.Runs.get(msg.id, thread_id=thread.id)
    print(run)
    verify_status_code(run)
    # print run status, to verify run is completed.
    print(run.status)

    run_steps = dashscope.Steps.list(run.id, thread_id=thread.id)

    print(run_steps)
    verify_status_code(run_steps)

    # get the thread messages.
    msgs = dashscope.Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4, ensure_ascii=False))

import json
import sys
from http import HTTPStatus

import dashscope

def create_assistant():
    # create assistant with information
    assistant = dashscope.Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        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': 'text_to_image'
        }],
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant()
    print(assistant)
    verify_status_code(assistant)

    # create a thread.
    thread = dashscope.Threads.create()
    print(thread)
    verify_status_code(thread)

    # create a message.
    message = dashscope.Messages.create(thread.id, content='Draw a picture of a cute kitten lying on the sofa.')
    print(message)
    verify_status_code(message)

    # create a new run to run message
    message_run = dashscope.Runs.create(thread.id, assistant_id=assistant.id)
    print(message_run)
    verify_status_code(message_run)

    # get run statue
    run = dashscope.Runs.get(message_run.id, thread_id=thread.id)
    print(run)
    verify_status_code(run)
    # print run status, to verify run is completed.
    print(run.status)

    # wait for run completed or requires_action
    run = dashscope.Runs.wait(run.id, thread_id=thread.id)
    print(run)

    run = dashscope.Runs.get(run.id, thread_id=thread.id)
    print(run)
    verify_status_code(run)

    run_steps = dashscope.Steps.list(run.id, thread_id=thread.id)

    print(run_steps)
    verify_status_code(run_steps)

    # get the thread messages.
    msgs = dashscope.Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4, ensure_ascii=False))

import json
import sys
from http import HTTPStatus

import dashscope

def create_assistant():
    # create assistant with information
    assistant = dashscope.Assistants.create(
    # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
        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': 'code_interpreter'
        }],
    )

    return assistant

def verify_status_code(res):
    if res.status_code != HTTPStatus.OK:
        sys.exit(res.status_code)

if __name__ == '__main__':
    # create assistant
    assistant = create_assistant()
    print(assistant)
    verify_status_code(assistant)

    # create a thread.
    thread = dashscope.Threads.create()
    print(thread)
    verify_status_code(thread)

    # create a message.
    message = dashscope.Messages.create(thread.id, content='有若干只鸡兔同在一个笼子里，从上面数，有100个头，从下面数，有334只脚。请用工具计算笼中各有多少只鸡和兔?, ')
    print(message)
    verify_status_code(message)

    # create a new run to run message
    message_run = dashscope.Runs.create(thread.id, assistant_id=assistant.id)
    print(message_run)
    verify_status_code(message_run)

    # get run statue
    run = dashscope.Runs.get(message_run.id, thread_id=thread.id)
    print(run)
    verify_status_code(run)
    # print run status, to verify run is completed.
    print(run.status)

    # wait for run completed or requires_action
    run = dashscope.Runs.wait(run.id, thread_id=thread.id)
    print(run)

    run = dashscope.Runs.get(run.id, thread_id=thread.id)
    print(run)
    verify_status_code(run)

    run_steps = dashscope.Steps.list(run.id, thread_id=thread.id)

    print(run_steps)
    verify_status_code(run_steps)

    # get the thread messages.
    msgs = dashscope.Messages.list(thread.id)
    print(msgs)
    print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4, ensure_ascii=False))

from dashscope import Assistants, Messages, Runs, Threads

assistant = Assistants.create(
# 此处以qwen-max为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
    model='qwen-max',
    name='smart helper',
    description='一个智能助手，可以通过用户诉求，调用已有的插件能力帮助用户。',
    instructions='你是一个智能助手，请记住以下信息。${document1}',  # from rag library
    tools=[
        {
            "type": "rag",
            "prompt_ra": {
                "pipeline_id": ["fqcioazfej","fqcioazfej"],  # 知识库 id ["9c0simc8q8", "13loysdy91"]
                "multiknowledge_rerank_top_n":10,  # 多个知识库总共召回的片段数
                "rerank_top_n":5, # 单个知识库召回的片段数
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query_word": {
                            "type": "str",
                            "value": "${document1}"
                        }

                    }
                }
            }

        },

    ]
)

def send_message(assistant, message='百炼是什么？'):
    print(f"Query: {message}")

    # create thread.
    # create a thread.
    thread = Threads.create()

    print(thread)

    # create a message.
    message = Messages.create(thread.id, content=message)
    # create run

    run = Runs.create(thread.id, assistant_id=assistant.id)
    print(run)

    # # get run statue
    # run_status = Runs.get(run.id, thread_id=thread.id)
    # print(run_status)

    # wait for run completed or requires_action
    run_status = Runs.wait(run.id, thread_id=thread.id)
    # print(run_status)

    # if prompt input tool result, submit tool result.

    run_status = Runs.get(run.id, thread_id=thread.id)
    print(run_status)
    # verify_status_code(run_status)

    # get the thread messages.
    msgs = Messages.list(thread.id)
    # print(msgs)
    # print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4))

    print("运行结果:")
    for message in msgs['data'][::-1]:
        print("content: ", message['content'][0]['text']['value'])
    print("\n")

if __name__ == "__main__":
    send_message(assistant, message='百炼是什么?')