Assistant API 调用示例（下线中）

import os
import json
import sys
from http import HTTPStatus

from dashscope import Assistants, Messages, Runs, Threads
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'

def create_assistant():
    # create assistant with information
    assistant = Assistants.create(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://www.alibabacloud.com/help/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
    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 os
import json
import sys
from http import HTTPStatus

from dashscope import Assistants, Messages, Runs, Threads
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'

def create_assistant():
    # create assistant with information
    assistant = Assistants.create(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://www.alibabacloud.com/help/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 os
import json
import sys
from http import HTTPStatus

from dashscope import Assistants, Messages, Runs, Steps, Threads
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'

def create_assistant_call_function():
    # create assistant with information
    assistant = Assistants.create(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://www.alibabacloud.com/help/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 os
import json
import sys
from http import HTTPStatus

import dashscope
from dashscope import Assistants, Messages, Runs, Steps, Threads
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'

def create_assistant_call_function():
    # create assistant with information
    assistant = Assistants.create(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://www.alibabacloud.com/help/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 os
import json
import sys
from http import HTTPStatus

import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'

def create_assistant():
    # create assistant with information
    assistant = dashscope.Assistants.create(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://www.alibabacloud.com/help/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 os
import json
import sys
from http import HTTPStatus

import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'

def create_assistant():
    # create assistant with information
    assistant = dashscope.Assistants.create(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # 此处以qwen-max为例，可按需更换模型名称。模型列表：https://www.alibabacloud.com/help/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))