Assistant API 代码解释器

代码解释器是一款智能体插件,它能辅助智能体编写和运行程序,解决科学计算和工程应用领域的问题。阅读这篇文章后,您将了解代码解释器的丰富功能,并学习集成代码解释器的方法。

功能特点

作为百炼预置的智能体插件,代码解释器可以帮助智能体执行 Python 代码。有了这个工具,智能体可以根据您的指令构建 Python 代码来处理数据。如果代码无法运行,它还可以帮您尝试改进代码。

配备代码解释器的智能体可以完成下列任务:

  • 格式转换:支持多种数据格式之间的转换,如JSON、CSV、XML等。

  • 数据处理转换:实现多种数据清洗、预处理和转换功能,包括缺失值处理、数据归一化、标准化、特征提取、特征选择等。对数据进行分组、聚合、过滤、排序等操作。

  • 数据统计:执行描述性统计分析、假设检验、回归分析等。

  • 数据可视化:支持生成多种类型的图表,如折线图、柱状图、散点图、热力图等。可以自定义图表样式、颜色、标签等。

  • 数学建模:支持线性模型、非线性模型、时间序列分析等建模类型,并提供模型评估和优化功能。

  • 数值分析:求解微分方程、积分计算、优化问题等。

image数值分析

image图像绘制

image格式转换

image

image

image

集成方法

快速入门,您已经了解 Assistant API 创建智能体应用的简单流程,其中不乏集成文生图工具的方法。类似地,集成代码解释器的思路大致如下:

image

相比基础流程,代码解释器应用需要额外处理大模型输出和代码执行结果之间的交互循环,直到完成整个任务。接下来,您将通过一个典型的示例项目来了解各个环节的开发方法。

典型示例

在这个示例中,您将创建一个 Web 代码教学助手,能够帮助用户编辑、调试和执行代码。您将使用 Assistant API 开发具备代码解释器的智能体应用,Web 界面将使用 Gradio 快速搭建。

环境准备

  • Python 环境:在开始之前,请确保您的开发环境满足以下条件:

    • 推荐使用 Python 3.10 或更高版本。您可以通过以下命令查看您的 Python 版本:

      python3 --version

    • Pip 版本是最新的。您可以通过以下命令升级 Pip 到最新版本:

      pip install --upgrade pip

    • 已安装 DashScope 和 Gradio 软件包。您可以通过以下命令安装软件包:

      pip install dashscope==1.20.11 gradio==5.1.0

  • 百炼 API-KEY:在 Assistant API 的开发场景中,您需要获取配置百炼 API-KEY。

接下来,您将了解开发示例场景的重要方法。建议先浏览并运行以下完整代码,有助于您了解整个示例的开发方法。代码运行后,您可以通过http://127.0.0.1:7860/访问 Web 代码教学助手。

完整代码

"""
代码解释器教学示例

这个示例展示了如何使用 dashscope 的 Assistant API 创建一个具有代码执行能力的AI智能体。
整个过程分为五个主要步骤:
1. 创建智能体 (Assistant) - 配置AI智能体的功能和行为
2. 创建对话线程 (Thread) - 管理一次完整的对话会话
3. 发送消息 (Message) - 在对话线程中添加用户输入
4. 运行智能体 (Run) - 处理用户输入并生成响应
5. 处理步骤 (Steps) - 跟踪并处理智能体的执行过程
"""

import dashscope
import gradio as gr
import logging
import asyncio

# 配置日志以跟踪API调用和执行过程
logging.basicConfig(level=logging.INFO)

class CodeTeachingAssistant:
    """
    代码教学助手类
    
    工作流程:
    1. 初始化时创建助手和对话线程
    2. 接收用户消息后,将其添加到对话线程
    3. 创建运行实例处理用户输入
    4. 跟踪运行过程中的每个步骤
    5. 返回处理结果
    """
    def __init__(self):
        # 第一步:创建Assistant实例
        # - model: 选择支持代码解释器的模型
        # - tools: 启用代码解释器功能
        # - instructions: 定义Assistant的行为准则
        self.assistant = dashscope.Assistants.create(
            model='qwen-plus',  # 使用支持代码执行的千问模型
            name='编程教学助手',
            instructions='''
                你是一个编程教师,负责:
                1. 解释代码原理
                2. 提供代码示例
                3. 执行代码并分析结果
                请用通俗易懂的方式教学。
            ''',
            tools=[{'type': 'code_interpreter'}]  # 启用代码解释器
        )
        
        # 第二步:创建对话线程
        # Thread用于维护一次完整的对话上下文
        self.thread = dashscope.Threads.create()
        
    async def chat(self, message: str):
        """
        处理用户消息并返回助手回复
        
        工作流程:
        1. 将用户消息添加到对话线程
        2. 创建运行实例处理消息
        3. 跟踪运行过程中的每个步骤
        4. 逐步返回处理结果
        """
        # 第三步:发送用户消息
        # 将用户输入添加到对话线程中
        dashscope.Messages.create(
            thread_id=self.thread.id,
            role="user",
            content=message
        )
        
        # 第四步:创建运行实例
        # Run 负责处理用户输入并生成响应
        run = dashscope.Runs.create(
            thread_id=self.thread.id,
            assistant_id=self.assistant.id
        )
        
        # 用于跟踪已处理的步骤,避免重复处理
        processed_steps = set()
        
        # 第五步:处理运行步骤
        while True:
            # 获取运行状态
            run_status = dashscope.Runs.retrieve(
                thread_id=self.thread.id,
                run_id=run.id
            )
            
            # 获取执行步骤列表
            # Steps 包含了Assistant执行过程中的所有操作
            steps = dashscope.Steps.list(
                thread_id=self.thread.id,
                run_id=run.id
            )
            
            # 处理新的步骤
            if hasattr(steps, 'data'):
                for step in steps.data:
                    # 跳过已处理的步骤
                    if step.id not in processed_steps:
                        processed_steps.add(step.id)
                        
                        # 处理消息创建步骤
                        # 这类步骤包含Assistant的文本响应
                        if step.step_details.type == 'message_creation':
                            message_id = step.step_details.message_creation.message_id
                            message = dashscope.Messages.retrieve(
                                thread_id=self.thread.id,
                                message_id=message_id
                            )
                            if hasattr(message, 'content'):
                                yield message.content[0].text.value + "\n"
                        
                        # 处理代码执行步骤
                        # 这类步骤包含代码执行的过程和结果
                        elif step.step_details.type == 'tool_calls':
                            for tool_call in step.step_details.tool_calls:
                                if tool_call.type == 'code_interpreter':
                                    # 显示执行的代码
                                    yield f"\n执行代码:\n{tool_call.code_interpreter.arguments}\n"
                                    # 显示执行结果
                                    yield f"\n执行结果:\n{tool_call.code_interpreter.output}\n"
            
            # 检查运行状态
            if run_status.status == 'completed':
                break  # 处理完成
            elif run_status.status == 'failed':
                yield "处理失败,请重试"
                break  # 处理失败
                
            # 等待新的步骤
            await asyncio.sleep(1)

def create_web_ui():
    """
    创建Web交互界面
    
    使用Gradio构建简单的聊天界面,包括:
    - 对话历史显示
    - 消息输入框
    - 示例问题
    """
    assistant = CodeTeachingAssistant()
    
    css = """
        .contain { display: flex; flex-direction: column; }
        .gradio-container { height: 100vh !important; }
        #component-0 { height: 100%; }
        #chatbot { flex-grow: 1; overflow: auto;}
    """
    
    with gr.Blocks(css=css) as demo:
        chatbot = gr.Chatbot(elem_id="chatbot", label="编程学习")
        msg = gr.Textbox(label="输入问题或代码")
        
        async def respond(message, history):
            history.append((message, ""))  # 立即添加用户消息
            yield "", history  # 清空输入框并显示用户消息
            
            # 创建一个新的响应条目
            response = ""
            async for chunk in assistant.chat(message):
                response += chunk
                # 实时更新最后一条消息
                history[-1] = (message, response)
                yield "", history  # 保持输入框为空,更新对话历史
            
        msg.submit(respond, [msg, chatbot], [msg, chatbot])
        
        # 添加示例问题
        gr.Examples([
            "请用Python写一个计算斐波那契数列的函数,并画出前20个数的趋势图",
            "帮我写一个冒泡排序算法,并用随机数据测试它的效果",
            "请用matplotlib画一个简单的正弦波图形"
        ], inputs=msg)
        
    return demo

if __name__ == "__main__":
    demo = create_web_ui()
    demo.launch()

创建代码教学助手

您可以构建一个代码教学助手组件,包含智能体应用的全部组件和工作流程。在初始化方法__init__中您将创建智能体实例,包括大模型、系统指令和工具配置。在对话方法chat中您将处理用户和智能体的所有消息。

class CodeTeachingAssistant:
    """
    代码教学助手接口
    主要职责:
    1. 初始化时创建助手和对话线程
    2. 接收用户消息后,将其添加到对话线程
    3. 创建运行实例处理用户输入
    4. 跟踪运行过程中的每个步骤
    5. 返回处理结果
    """
    def __init__(self):
        """创建Assistant实例"""
        pass

    async def chat(self, message: str):
        """
        处理用户消息并返回助手回复

        工作流程:
        1. 将用户消息添加到对话线程
        2. 创建运行实例处理消息
        3. 跟踪运行过程中的每个步骤
        4. 逐步返回处理结果
        """
        pass

这里您先声明了代码教学助手组件和两种方法,接下来将讨论它们的实现方法。

创建智能体与线程

在 Assistant API 中,我们需要两个核心组件来启动对话系统:

  • 智能体(Assistant)

    智能体是对话系统的大脑,负责理解和处理用户输入。创建智能体时需要配置:

    • model:选择支持代码执行的模型(如 qwen-plus)

    • name:为智能体命名

    • instructions:定义智能体的行为准则

    • tools:启用代码解释器等工具

  • 线程(Thread)

    线程用于管理完整的对话上下文,确保对话的连贯性。每个用户会话对应一个独立的线程。

这两个组件在 CodeTeachingAssistant 类的初始化方法中配置:

def __init__(self):
    # 第一步:创建Assistant实例
    # - model: 选择支持代码解释器的模型
    # - tools: 启用代码解释器功能
    # - instructions: 定义Assistant的行为准则
    self.assistant = dashscope.Assistants.create(
        model='qwen-plus',  # 使用支持代码执行的千问模型
        name='编程教学助手',
        instructions='''
            你是一个编程教师,负责:
            1. 解释代码原理
            2. 提供代码示例
            3. 执行代码并分析结果
            请用通俗易懂的方式教学。
        ''',
        tools=[{'type': 'code_interpreter'}]  # 启用代码解释器
    )
    # 第二步:创建对话线程
    # Thread用于维护一次完整的对话上下文
    self.thread = dashscope.Threads.create()

创建这些组件后,它们可以在整个会话过程中重复使用,无需重新初始化。智能体将根据配置的指令和工具来处理用户的每一个请求,而线程则会保持对话的上下文连续性。

处理对话消息

在 Assistant API 中,处理用户消息是一个异步过程,需要经过多个步骤来完成完整的对话交互。chat 方法实现了这个复杂的处理流程:

发送用户消息

首先,我们需要将用户的输入添加到对话线程中:

async def chat(self, message: str):
    """
    处理用户消息并返回助手回复
    
    工作流程:
    1. 将用户消息添加到对话线程
    2. 创建运行实例处理消息
    3. 跟踪运行过程中的每个步骤
    4. 逐步返回处理结果
    """
    # 第三步:发送用户消息
    # 将用户输入添加到对话线程中
    dashscope.Messages.create(
        thread_id=self.thread.id,
        role="user",
        content=message
    )
    

创建运行实例

创建一个运行实例来处理这条消息:

async def chat(self, message: str):
    """
    处理用户消息并返回助手回复
    
    工作流程:
    1. 将用户消息添加到对话线程
    2. 创建运行实例处理消息
    3. 跟踪运行过程中的每个步骤
    4. 逐步返回处理结果
    """
    # 第四步:创建运行实例
    # Run 负责处理用户输入并生成响应
    run = dashscope.Runs.create(
        thread_id=self.thread.id,
        assistant_id=self.assistant.id
    )
    

步骤追踪与处理

系统会将智能体的运行过程分解为多个步骤,每个步骤可能是“消息创建”或“工具调用”。在实际示例中,这两种步骤会交替进行,但智能体总是先“创建消息”,且最后也以“创建消息”结尾。因此我们需要:

  • 追踪运行状态

  • 获取步骤列表

  • 处理每个新步骤

  • 避免重复处理

  1. 处理消息创建步骤

    当智能体生成文本响应时:

async def chat(self, message: str):
    # 这类步骤包含Assistant的文本响应
    if step.step_details.type == 'message_creation':
        message_id = step.step_details.message_creation.message_id
        message = dashscope.Messages.retrieve(
            thread_id=self.thread.id,
            message_id=message_id
        )
        if hasattr(message, 'content'):
            yield message.content[0].text.value + "\n"
            
  1. 处理代码执行步骤

    当智能体执行代码时:

    async def chat(self, message: str):
        # 处理代码执行步骤
        # 这类步骤包含代码执行的过程和结果
        elif step.step_details.type == 'tool_calls':
            for tool_call in step.step_details.tool_calls:
                if tool_call.type == 'code_interpreter':
                    # 显示执行的代码
                    yield f"\n执行代码:\n{tool_call.code_interpreter.arguments}\n"
                    # 显示执行结果
                    yield f"\n执行结果:\n{tool_call.code_interpreter.output}\n"
                    

状态监控

系统会持续监控运行状态,直到处理完成或失败:

  • completed:处理成功完成

  • failed:处理失败,需要重试

async def chat(self, message: str):
    # 检查运行状态
    if run_status.status == 'completed':
        break  # 处理完成
    elif run_status.status == 'failed':
        yield "处理失败,请重试"
        break  # 处理失败
        
    # 等待新的步骤
    await asyncio.sleep(1)
    

其他实现细节

  • 使用 async/await 异步处理,确保响应的实时性

  • 使用 yield 逐步返回结果,提供流式体验

  • 使用 set 数据结构跟踪已处理步骤,避免重复处理

  • 通过 asyncio.sleep(1) 控制轮询间隔

注意事项

  • 所有API调用都是异步的,需要适当处理等待时间

  • 步骤处理是顺序进行的,确保消息按正确顺序显示

  • 错误处理机制确保在处理失败时能够正常退出

创建Web交互界面

使用 Gradio 框架构建一个直观的 Web 界面,让用户能够方便地与代码教学助手进行交互。

创建基础界面

使用gr.Blocks()创建一个灵活的界面容器:

with gr.Blocks() as demo:
    chatbot = gr.Chatbot(label="编程学习")
    msg = gr.Textbox(label="输入问题或代码")

实现消息处理

定义异步响应函数,处理用户输入并更新对话历史:

async def respond(message, history):
    """处理用户输入并更新对话历史"""
    response = ""
    async for chunk in assistant.chat(message):
        response += chunk
    history.append((message, response))
    return "", history

关键特性

  • 使用异步函数处理消息

  • 流式接收响应内容

  • 动态更新对话历史

  • 清空输入框以准备下一次输入

配置交互事件

提供预设的示例问题,帮助用户快速开始:

gr.Examples(
    examples=[
        "请用Python写一个计算斐波那契数列的函数,并画出前20个数的趋势图",
        "帮我写一个冒泡排序算法,并用随机数据测试它的效果",
        "请用matplotlib画一个简单的正弦波图形"
    ],
    inputs=msg
)

启动应用

if __name__ == "__main__":
    demo = create_web_ui()
    demo.launch()

界面组件说明

  1. Chatbot 组件

    • 显示完整的对话历史

    • 支持代码和文本的格式化显示

    • 自动滚动到最新消息

    • 优化页面样式

  2. Textbox 组件

    • 接收用户输入

    • 支持多行文本

    • 可通过回车提交

  3. Examples 组件

    • 提供常用示例

    • 一键填充输入框

    • 帮助用户理解系统功能

使用提示

  • 界面支持实时响应,无需手动刷新

  • 代码执行结果会自动格式化显示

  • 可以通过示例快速体验系统功能

  • 支持连续对话,保持上下文关联

总结

通过这个示例,您已经了解了如何使用代码解释器来创建一个教学智能体。代码解释器可以帮助用户学习编程概念并执行代码。您可以根据自己的需求,进一步扩展和定制这个智能体,以满足不同的教学和学习场景。

关于 Assistant API,您还可以了解:

常见问题

  1. 使用代码解释器会收取额外费用吗?

    目前,百炼预置的代码解释器限时免费。在本文的示例中,智能体运行时产生的所有“code_interpreter”步骤暂不计费。

  2. 代码解释器支持哪些 Python 软件包?

    Python 软件包的支持情况可能发生变化,您可以尝试询问示例智能体获取最新的支持信息。

  3. 代码解释器存在哪些限制

    由于运行在沙盒环境中,代码解释器存在一些限制:

    • 只能执行 Python 代码

    • 无法读写外部文件

    • 无法安装外部软件包

    • 无法访问外部网络

    • 计算资源和时长受限

  4. 如何处理数据文件?

    代码解释器目前不支持直接处理数据文件。建议您将数据转换为文本格式,并通过输入框发送到智能体进行处理。

  5. 如何处理错误信息?

    当代码执行失败时,智能体会返回详细的错误信息。请根据错误提示进行相应调整。如果遇到网络不稳定导致的响应延迟或超时,请稍后重试。

  6. 如何获取技术支持?

    如果以上内容无法解决您的问题,请访问技术支持页面获取帮助。