Mistral Python 客户端

从 v1 迁移到 v2

如果您正在从 v1 升级到 v2，请查看迁移指南，了解破坏性变更以及如何更新您的代码。

API 密钥设置

在开始之前，您需要一个 Mistral AI 的 API 密钥。

1. 获取您自己的 Mistral API 密钥： https://docs.mistral.ai/#api-access
2. 将您的 Mistral API 密钥设置为环境变量。您只需执行一次此操作。

# 设置 Mistral API 密钥（以 zsh 为例）
$ echo 'export MISTRAL_API_KEY=[your_key_here]' >> ~/.zshenv

# 重新加载环境变量（或直接退出并打开一个新的终端）
$ source ~/.zshenv

摘要

Mistral AI API：我们的聊天完成和嵌入 API 规范。在 La Plateforme 上创建您的账户以获取访问权限，并阅读文档，了解如何使用它。

目录

• Mistral Python 客户端
  • 从 v1 迁移
  • API 密钥设置
  • SDK 安装
  • SDK 示例用法
  • 提供商 SDK 示例用法
  • 可用资源和操作
  • 服务器发送事件流
  • 分页
  • 文件上传
  • 重试
  • 错误处理
  • 服务器选择
  • 自定义 HTTP 客户端
  • 身份验证
  • 资源管理
  • 调试
  • IDE 支持
• 开发
  • 贡献

SDK 安装

[!NOTE] Python 版本升级政策

一旦某个 Python 版本达到其官方终止支持日期，将为用户提供 3 个月的宽限期以便升级。在此宽限期之后，SDK 所支持的最低 Python 版本将会更新。

SDK 可以通过 uv、pip 或 poetry 包管理器进行安装。

uv

uv 是一个快速的 Python 包安装和解析工具，旨在作为 pip 和 pip-tools 的替代品。由于其速度和现代 Python 工具链功能，推荐使用。

uv add mistralai

PIP

PIP 是 Python 的默认包管理器，可通过命令行轻松地从 PyPI 安装和管理软件包。

pip install mistralai

Poetry

Poetry 是一种现代化的工具，通过使用单个 pyproject.toml 文件来管理项目元数据和依赖关系，从而简化了依赖管理和软件包发布。

poetry add mistralai

使用 uv 进行 Shell 和脚本操作

您可以在 Python Shell 中使用此 SDK，借助 uv 和随附的 uvx 命令，如下所示：

uvx --from mistralai python

也可以编写独立的 Python 脚本，而无需设置整个项目，如下所示：

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "mistralai",
# ]
# ///

from mistralai.client import Mistral

sdk = Mistral(
  # SDK 参数
)

# 脚本其余部分...

保存到文件后，您可以使用 uv run script.py 来运行脚本，其中 script.py 可以替换为实际的文件名。

代理额外依赖项

当使用与代理相关的功能时，需要添加 agents 额外依赖项。这可以在安装软件包时一并添加：

pip install "mistralai[agents]"

注意：这些功能需要 Python 3.10 或更高版本（SDK 的最低要求）。

其他软件包

其他 mistralai-* 软件包（例如 mistralai-workflows）可以单独安装，并且属于 mistralai 命名空间：

pip install mistralai-workflows

SDK 示例用法

创建聊天完成

此示例展示了如何创建聊天完成。

# 同步示例
from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.chat.complete(model="mistral-large-latest", messages=[
        {
            "role": "user",
            "content": "谁是法国最伟大的画家？用一句话回答。",
        },
    ], stream=False, response_format={
        "type": "text",
    })

    # 处理响应
    print(res)

同样的 SDK 客户端也可以通过导入 asyncio 来发出异步请求。

# 异步示例
import asyncio
from mistralai.client import Mistral
import os

async def main():

    async with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:

        res = await mistral.chat.complete_async(model="mistral-large-latest", messages=[
            {
                "role": "user",
                "content": "谁是法国最伟大的画家？用一句话回答。",
            },
        ], stream=False, response_format={
            "type": "text",
        })

        # 处理响应
        print(res)

asyncio.run(main())

上传文件

此示例展示了如何上传文件。

# 同步示例
from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.files.upload(file={
        "file_name": "example.file",
        "content": open("example.file", "rb"),
    }, visibility="workspace")

    # 处理响应
    print(res)

同样的 SDK 客户端也可以通过导入 asyncio 发出异步请求。

# 异步示例
import asyncio
from mistralai.client import Mistral
import os

async def main():

    async with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:

        res = await mistral.files.upload_async(file={
            "file_name": "example.file",
            "content": open("example.file", "rb"),
        }, visibility="workspace")

        # 处理响应
        print(res)

asyncio.run(main())

创建代理完成

此示例展示了如何创建代理完成。

# 同步示例
from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),Diaspora: A Social Network for the People
By: Noura Al-Mutairi
Introduction
In today's digital age, social media platforms have become an integral part of our daily lives. They offer us a way to connect with friends and family, share our thoughts and experiences, and stay updated on current events. However, despite their popularity, many people are becoming increasingly concerned about privacy issues, data security, and the lack of transparency in how these platforms operate. This has led to the rise of alternative social networks that prioritize user privacy and control. One such platform is Diaspora, which was created by four students at New York University in 2010. In this essay, I will explore what makes Diaspora unique, its benefits and challenges, and why it might be a better choice for those who value their online privacy.
What is Diaspora?
Diaspora is an open-source social networking platform that allows users to create their own independent servers, known as "pods," where they can host their personal data and interact with others. Unlike traditional social media platforms like Facebook or Twitter, which store all user data on centralized servers owned by the company, Diaspora gives users full control over their data and how it is shared. Each pod can be customized to meet the specific needs of its users, and pods can communicate with each other through a decentralized network called the "federation." This means that users can join any pod they choose, regardless of where it is located, and still maintain a seamless connection with other pods around the world.
Benefits of Diaspora
One of the main advantages of Diaspora is its emphasis on privacy and user control. By allowing users to host their own data on independent servers, Diaspora eliminates the risk of data breaches and unauthorized access that often occur on centralized platforms. Additionally, since users have full control over their data, they can decide who can see their posts, photos, and personal information, and how much of it they want to share with others. This level of control is particularly important for individuals who are concerned about their online safety and want to protect their personal information from prying eyes.
Another benefit of Diaspora is its flexibility and customization options. With thousands of pods available to choose from, users can find one that suits their preferences and needs. Whether they want a pod that focuses on art and culture, technology and innovation, or community building and activism, there is always a pod that matches their interests. Moreover, users can also create their own pods if none of the existing ones meet their requirements. This level of freedom and creativity is rare in traditional social media platforms, where users are often limited to the features and settings provided by the company.
Challenges of Diaspora
Despite its many benefits, Diaspora also faces some challenges that may hinder its growth and adoption. One of the main challenges is its lack of popularity and awareness among the general public. Since Diaspora is still a relatively new platform compared to Facebook or Twitter, many people are not familiar with its features and benefits. As a result, it is difficult for Diaspora to attract a large number of users and compete with established players in the market. Another challenge is its technical complexity and learning curve. Unlike traditional social media platforms, which are designed to be easy to use and intuitive, Diaspora requires some technical knowledge and skills to set up and manage a pod. This may discourage some users who are not comfortable with technology or do not have access to reliable internet connections.
Conclusion
In conclusion, Diaspora is a unique and promising social networking platform that offers a refreshing alternative to traditional social media platforms. By giving users full control over their data and interactions, Diaspora empowers them to take charge of their online presence and express themselves freely without fear of censorship or surveillance. Moreover, Diaspora's decentralized nature and flexible structure make it a perfect fit for diverse communities and groups that seek to build strong bonds and promote mutual understanding. However, in order for Diaspora to reach its full potential, more efforts are needed to increase its visibility and accessibility among the general public. Only then can Diaspora become a viable option for those who value their online privacy and want to reclaim their digital sovereignty.

# 同步示例
from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.agents.complete(messages=[
        {
            "role": "user",
            "content": "谁是法国最伟大的画家？用一句话简短回答。",
        },
    ], agent_id="<id>", stream=False, response_format={
        "type": "text",
    })

    # 处理响应
    print(res)

同样的 SDK 客户端也可以通过导入 asyncio 来进行异步请求。

# 异步示例
import asyncio
from mistralai.client import Mistral
import os

async def main():

    async with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:

        res = await mistral.agents.complete_async(messages=[
            {
                "role": "user",
                "content": "谁是法国最伟大的画家？用一句话简短回答。",
            },
        ], agent_id="<id>", stream=False, response_format={
            "type": "text",
        })

        # 处理响应
        print(res)

asyncio.run(main())

创建嵌入请求

本示例展示了如何创建嵌入请求。

# 同步示例
from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.embeddings.create(model="mistral-embed", inputs=[
        "对这句话进行嵌入。",
        "也对这句话进行嵌入。",
    ])

    # 处理响应
    print(res)

同样的 SDK 客户端也可以通过导入 asyncio 来进行异步请求。

# 异步示例
import asyncio
from mistralai.client import Mistral
import os

async def main():

    async with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:

        res = await mistral.embeddings.create_async(model="mistral-embed", inputs=[
            "对这句话进行嵌入。",
            "也对这句话进行嵌入。",
        ])

        # 处理响应
        print(res)

asyncio.run(main())

更多示例

您可以使用 uv run 在 examples/ 目录中运行这些示例。

各云服务商 SDK 示例用法

Azure AI

先决条件

在开始之前，请确保您已准备好 AZURE_ENDPOINT 和 AZURE_API_KEY。要获取这些信息，您需要在 Azure AI 上部署 Mistral 模型。 有关在 Azure AI 上部署 Mistral 的说明，请参阅此处。

步骤 1：安装

pip install mistralai

步骤 2：示例用法

以下是一个基本示例，帮助您入门。您也可以运行 examples 目录中的示例。

import os
from mistralai.azure.client import MistralAzure

# SDK 会自动将 api-version 作为查询参数注入
client = MistralAzure(
    api_key=os.environ["AZURE_API_KEY"],
    server_url=os.environ["AZURE_ENDPOINT"],
    api_version="2024-05-01-preview",  # 可选，默认值为该版本
)

res = client.chat.complete(
    model=os.environ["AZURE_MODEL"],
    messages=[
        {
            "role": "user",
            "content": "你好！",
        }
    ],
)
print(res.choices[0].message.content)

Google Cloud

先决条件

在开始之前，您需要创建一个 Google Cloud 项目并启用 Mistral API。为此，请按照此处的说明操作。 要在本地运行此示例，您还需要确保已通过 Google Cloud 进行身份验证。您可以通过运行以下命令来完成：

gcloud auth application-default login

步骤 1：安装

pip install mistralai
# 如果需要 GCP 身份验证支持（必需）：
pip install "mistralai[gcp]"

步骤 2：示例用法

以下是一个基本示例，帮助您入门。您也可以运行 examples 目录中的示例。

SDK 自动执行以下操作：

• 通过 google.auth.default() 检测凭据
• 在令牌过期时自动刷新
• 根据 project_id 和 region 构建 Vertex AI 的 URL

import os
from mistralai.gcp.client import MistralGCP

# SDK 会自动检测凭据并构建 Vertex AI 的 URL
client = MistralGCP(
    project_id=os.environ.get("GCP_PROJECT_ID"),  # 可选：从凭据中自动检测
    region="us-central1",  # 默认值：europe-west4
)

res = client.chat.complete(
    model="mistral-small-2503",
    messages=[
        {
            "role": "user",
            "content": "你好！",
        }
    ],
)
print(res.choices[0].message.content)

可用资源与操作

可用方法

Agents

• complete - 代理完成
• stream - 流式代理完成

Audio.Speech

• complete - 语音

Audio.Transcriptions

• complete - 创建转录
• stream - 创建流式转录（SSE）

Audio.Voices

• list - 列出所有语音
• create - 创建新语音
• delete - 删除自定义语音
• update - 更新语音元数据
• get - 获取语音详情
• get_sample_audio - 获取语音样本音频

Batch.Jobs

• list - 获取批量作业
• create - 创建批量作业
• get - 获取批量作业
• delete - 删除批量作业
• cancel - 取消批量作业

Beta.Agents

• create - 创建一个可在会话中使用的代理。
• list - 列出所有代理实体。
• get - 获取一个代理实体。
• update - 更新一个代理实体。
• delete - 删除一个代理实体。
• update_version - 更新代理的一个版本。
• list_versions - 列出代理的所有版本。
• get_version - 获取代理的特定版本。
• create_version_alias - 创建或更新代理版本别名。
• list_version_aliases - 列出代理的所有别名。
• delete_version_alias - 删除代理版本别名。

Beta.Connectors

• create - 创建一个新的连接器。
• list - 列出所有连接器。
• get_auth_url - 获取连接器的授权 URL。
• call_tool - 调用连接器工具。
• list_tools - 列出连接器的工具。
• get - 获取一个连接器。
• update - 更新一个连接器。
• delete - 删除一个连接器。

Beta.Conversations

• start - 创建一个会话并为其添加条目。
• list - 列出所有已创建的会话。
• get - 获取会话信息。
• delete - 删除一个会话。
• append - 向现有会话添加新条目。
• get_history - 获取会话中的所有条目。
• get_messages - 获取会话中的所有消息。
• restart - 从指定条目重新开始一个会话。
• start_stream - 创建一个会话并为其添加条目。
• append_stream - 向现有会话添加新条目。
• restart_stream - 从指定条目重新开始一个会话。

Beta.Libraries

• list - 列出您有权访问的所有库。
• create - 创建一个新的库。
• get - 获取特定库的详细信息。
• delete - 删除一个库及其所有文档。
• update - 更新一个库。

Beta.Libraries.Accesses

• list - 列出对该库的所有访问权限。
• update_or_create - 创建或更新访问级别。
• delete - 删除一个访问级别。

Beta.Libraries.Documents

• list - 列出给定库中的文档。
• upload - 上传新文档。
• get - 获取特定文档的元数据。
• update - 更新特定文档的元数据。
• delete - 删除文档。
• text_content - 获取特定文档的文本内容。
• status - 获取特定文档的处理状态。
• get_signed_url - 获取特定文档的签名 URL。
• extracted_text_signed_url - 获取从给定文档中提取的文本的签名 URL。
• reprocess - 对文档进行重新处理。

Beta.Observability.Campaigns

• create - 创建并启动一个新的活动。
• list - 获取所有活动。
• fetch - 根据 ID 获取活动。
• delete - 删除一个活动。
• fetch_status - 根据活动 ID 获取活动状态。
• list_events - 获取由给定活动选择的事件 ID。

Beta.Observability.ChatCompletionEvents

• search - 获取聊天完成事件。
• search_ids - 作为 /search 的替代方案，仅返回 ID，并且可以一次性返回大量 ID。
• fetch - 获取聊天完成事件。
• fetch_similar_events - 获取相似的聊天完成事件。
• judge - 根据给定选项对事件进行评判。

Beta.Observability.ChatCompletionEvents.Fields

• list - 获取聊天完成字段。
• fetch_options - 获取聊天完成字段选项。
• fetch_option_counts - 获取聊天完成字段选项的数量。

Beta.可观测性.数据集

• create - 创建一个新的空数据集
• list - 列出现有数据集
• fetch - 根据ID获取数据集
• delete - 删除数据集
• update - 部分更新数据集
• list_records - 列出数据集中现有的记录
• create_record - 向数据集中添加一条对话
• import_from_campaign - 使用市场活动填充数据集
• import_from_explorer - 使用探索器中的样本填充数据集
• import_from_file - 使用上传的文件中的样本填充数据集
• import_from_playground - 使用游乐场中的样本填充数据集
• import_from_dataset_records - 使用另一个数据集中的样本填充数据集
• export_to_jsonl - 导出到文件API，并获取用于下载生成的JSONL文件的预签名URL
• fetch_task - 获取数据集导入任务的状态
• list_tasks - 列出给定数据集的导入任务

Beta.可观测性.数据集.记录

• fetch - 从数据集中获取指定对话的内容
• delete - 从数据集中删除一条记录
• bulk_delete - 从数据集中批量删除多条记录
• judge - 根据给定选项对数据集中的记录运行Judge
• update_payload - 更新数据集记录的对话负载
• update_properties - 更新对话属性

Beta.可观测性.Judge

• create - 创建一个新的Judge
• list - 获取Judge，支持可选的过滤和搜索
• fetch - 根据ID获取Judge
• delete - 删除Judge
• update - 更新Judge
• judge_conversation - 对一条对话运行保存的Judge

聊天

• complete - 聊天完成
• stream - 流式聊天完成

分类器

• moderate - 内容审核
• moderate_chat - 聊天内容审核
• classify - 分类
• classify_chat - 聊天分类

嵌入

• create - 嵌入

事件

• get_stream_events - 获取流事件
• get_workflow_events - 获取工作流事件

文件

• upload - 上传文件
• list - 列出文件
• retrieve - 检索文件
• delete - 删除文件
• download - 下载文件
• get_signed_url - 获取签名URL

Fim

• complete - Fim完成
• stream - 流式Fim完成

微调作业

• list - 获取微调作业
• create - 创建微调作业
• get - 获取微调作业
• cancel - 取消微调作业
• start - 开始微调作业

模型

• list - 列出模型
• retrieve - 检索模型
• delete - 删除模型
• update - 更新微调后的模型
• archive - 归档微调后的模型
• unarchive - 解档微调后的模型

OCR

• process - OCR

工作流

• get_workflows - 获取工作流
• get_workflow_registrations - 获取工作流注册信息
• execute_workflow - 执行工作流
• execute_workflow_registration - 执行工作流注册 :warning: 已弃用
• get_workflow - 获取工作流
• update_workflow - 更新工作流
• get_workflow_registration - 获取工作流注册信息
• archive_workflow - 归档工作流
• unarchive_workflow - 取消归档工作流

工作流.部署

• list_deployments - 列出部署
• get_deployment - 获取部署

工作流.事件

• get_stream_events - 获取流事件
• get_workflow_events - 获取工作流事件

工作流.执行

• get_workflow_execution - 获取工作流执行
• get_workflow_execution_history - 获取工作流执行历史
• signal_workflow_execution - 发送工作流执行信号
• query_workflow_execution - 查询工作流执行
• terminate_workflow_execution - 终止工作流执行
• batch_terminate_workflow_executions - 批量终止工作流执行
• cancel_workflow_execution - 取消工作流执行
• batch_cancel_workflow_executions - 批量取消工作流执行
• reset_workflow - 重置工作流
• update_workflow_execution - 更新工作流执行
• get_workflow_execution_trace_otel - 获取工作流执行追踪（Otel）
• get_workflow_execution_trace_summary - 获取工作流执行追踪摘要
• get_workflow_execution_trace_events - 获取工作流执行追踪事件
• stream - 流式传输

工作流.指标

• get_workflow_metrics - 获取工作流指标

工作流.运行

• list_runs - 列出运行
• get_run - 获取运行
• get_run_history - 获取运行历史

工作流.计划

• get_schedules - 获取计划
• schedule_workflow - 计划工作流
• unschedule_workflow - 取消计划工作流

服务器发送事件流

服务器发送事件 用于从某些操作中流式传输内容。这些操作会将流暴露为 生成器, 可以使用简单的 for 循环来消费。当服务器不再有事件可发送并关闭底层连接时，循环将终止。

该流也是一个 上下文管理器, 可以与 with 语句一起使用，并在退出上下文时自动关闭底层连接。

from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

    })

    with res as event_stream:
        for event in event_stream:
            # 处理事件
            print(event, flush=True)

分页

此 SDK 中的一些端点支持分页。要使用分页，您可以像往常一样调用 SDK 方法，但返回的响应对象将具有一个 Next 方法，可以调用该方法来获取下一批结果。如果 Next 的返回值为 None, 则表示没有更多页面可供获取。

以下是一个此类分页调用的示例：

from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.workflows.get_workflows(active_only=False, include_shared=True, limit=50)

    while res is not None:
        # 处理项目

        res = res.next()

文件上传

某些 SDK 方法接受文件对象作为请求体或多部分请求的一部分。通常建议以流的形式上传文件，而不是将整个文件内容读入内存。这样做可以避免过度消耗内存，并防止在处理非常大的文件时出现内存不足错误。下面的示例展示了如何将文件流附加到请求中。

[!提示]

对于处理文件上传的端点，也可以使用字节数组。然而，对于大文件，建议使用流。

from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.transcriptions.complete(model="Model X", diarize=False)

    # 处理响应
    print(res)

重试

本 SDK 中的部分端点支持重试。如果您在未进行任何配置的情况下使用该 SDK，它将回退到 API 提供的默认重试策略。不过，您可以在单个操作级别或在整个 SDK 范围内覆盖默认重试策略。

要为单个 API 调用更改默认重试策略，只需向调用提供一个 RetryConfig 对象：

from mistralai.client import Mistral
from mistralai.client.utils import BackoffStrategy、RetryConfig
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

    },
        RetryConfig("backoff", BackoffStrategy(1, 50, 1.1, 100), False))

    with res as event_stream:
        for event in event_stream:
            # 处理事件
            print(event, flush=True)

如果您希望覆盖所有支持重试的操作的默认重试策略，可以在初始化 SDK 时使用 retry_config 可选参数：

from mistralai.client import Mistral
from mistralai.client.utils import BackoffStrategy、RetryConfig
import os


with Mistral(
    retry_config=RetryConfig("backoff", BackoffStrategy(1, 50, 1.1, 100), False),
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

    })

    with res as event_stream:
        for event in event_stream:
            # 处理事件
            print(event, flush=True)

错误处理

MistralError 是所有 HTTP 错误响应的基类。它具有以下属性：

属性	类型	描述
err.message	str	错误消息
err.status_code	int	HTTP 响应的状态码，例如 404
err.headers	httpx.Headers	HTTP 响应头
err.body	str	HTTP 正文。如果没有返回正文，则可能为空字符串。
err.raw_response	httpx.Response	原始 HTTP 响应
err.data		可选。某些错误可能包含结构化数据。请参阅错误类。

示例

from mistralai.client import Mistral、errors
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:
    res = None
    try:

        res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

        })

        with res as event_stream:
            for event in event_stream:
                # 处理事件
                print(event, flush=True)


    except errors.MistralError as e:
        # HTTP 错误响应的基类
        print(e.message)
        print(e.status_code)
        print(e.body)
        print(e.headers)
        print(e.raw_response)

        # 根据方法的不同，可能会抛出不同的错误
        if isinstance(e, errors.HTTPValidationError):
            print(e.data.detail)  # Optional[List[models.ValidationError]]

错误类

主要错误：

• MistralError：HTTP 错误响应的基类。

较常见的错误（7 种）

网络错误：

• httpx.RequestError：请求错误的基类。
  • httpx.ConnectError：HTTP 客户端无法向服务器发出请求。
  • httpx.TimeoutException：HTTP 请求超时。

继承自 MistralError：

• HTTPValidationError：验证错误。状态码为 422。适用于 168 种方法中的 103 种。*
• ObservabilityError：错误请求——请求参数或数据无效。适用于 168 种方法中的 41 种。*
• ResponseValidationError：响应数据与预期的 Pydantic 模型类型不匹配。可通过 cause 属性访问 Pydantic 验证错误。

* 请查看 方法文档 以了解错误是否适用。

服务器选择

按名称选择服务器

您可以通过在初始化 SDK 客户端实例时将服务器名称传递给 server: str 可选参数，来全局覆盖默认服务器。所选服务器随后将作为使用它的操作的默认服务器。下表列出了与可用服务器关联的名称：

名称	服务器	描述
eu	https://api.mistral.ai	欧盟生产服务器

示例

from mistralai.client import Mistral
import os


with Mistral(
    server="eu",
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

    })

    with res as event_stream:
        for event in event_stream:
            # 处理事件
            print(event, flush=True)

按客户端覆盖服务器 URL

默认服务器也可以通过在初始化 SDK 客户端实例时将 URL 传递给 server_url: str 可选参数来全局覆盖。例如：

from mistralai.client import Mistral
import os


with Mistral(
    server_url="https://api.mistral.ai",
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

    })

    with res as event_stream:
        for event in event_stream:
            # 处理事件
            print(event, flush=True)

自定义 HTTP 客户端

Python SDK 使用 httpx HTTP 库进行 API 调用。为了提供一种便捷的方式来配置超时、Cookie、代理、自定义头部以及其他底层配置，您可以使用自己的 HTTP 客户端实例来初始化 SDK 客户端。 根据您使用的是同步版还是异步版 SDK，您可以分别传入 HttpClient 或 AsyncHttpClient 的实例，它们是 Protocol 的实现，确保客户端具备执行 API 调用所需的方法。 这使您能够将客户端封装在自己的自定义逻辑中，例如添加自定义头部、日志记录或错误处理；或者您也可以直接传入 httpx.Client 或 httpx.AsyncClient 的实例。

例如，您可以为该 SDK 发出的每个请求指定一个头部，如下所示：

from mistralai.client import Mistral
import httpx

http_client = httpx.Client(headers={"x-custom-header": "someValue"})
s = Mistral(client=http_client)

或者，您可以将客户端封装在自己的自定义逻辑中：

from mistralai.client import Mistral
from mistralai.client.httpclient import AsyncHttpClient
import httpx

class CustomClient(AsyncHttpClient):
    client: AsyncHttpClient

    def __init__(self, client: AsyncHttpClient):
        self.client = client

    async def send(
        self,
        request: httpx.Request,
        *,
        stream: bool = False,
        auth: Union[
            httpx._types.AuthTypes, httpx._client.UseClientDefault, None
        ] = httpx.USE_CLIENT_DEFAULT,
        follow_redirects: Union[
            bool, httpx._client.UseClientDefault
        ] = httpx.USE_CLIENT_DEFAULT,
    ) -> httpx.Response:
        request.headers["Client-Level-Header"] = "added by client"

        return await self.client.send(
            request, stream=stream, auth=auth, follow_redirects=follow_redirects
        )

    def build_request(
        self,
        method: str,
        url: httpx._types.URLTypes,
        *,
        content: Optional[httpx._types.RequestContent] = None,
        data: Optional[httpx._types.RequestData] = None,
        files: Optional[httpx._types.RequestFiles] = None,
        json: Optional[Any] = None,
        params: Optional[httpx._types.QueryParamTypes] = None,
        headers: Optional[httpx._types.HeaderTypes] = None,
        cookies: Optional[httpx._types.CookieTypes] = None,
        timeout: Union[
            httpx._types.TimeoutTypes, httpx._client.UseClientDefault
        ] = httpx.USE_CLIENT_DEFAULT,
        extensions: Optional[httpx._types.RequestExtensions] = None,
    ) -> httpx.Request:
        return self.client.build_request(
            method,
            url,
            content=content,
            data=data,
            files=files,
            json=json,
            params=params,
            headers=headers,
            cookies=cookies,
            timeout=timeout,
            extensions=extensions,
        )

s = Mistral(async_client=CustomClient(httpx.AsyncClient()))

认证

每客户端安全方案

该 SDK 全局支持以下安全方案：

名称	类型	方案	环境变量
api_key	HTTP	HTTP Bearer	MISTRAL_API_KEY

要通过 API 进行认证，在初始化 SDK 客户端实例时必须设置 api_key 参数。例如：

from mistralai.client import Mistral
import os


with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    res = mistral.audio.speech.complete(input="<value>", stream=False, additional_properties={

    })

    with res as event_stream:
        for event in event_stream:
            # 处理事件
            print(event, flush=True)

资源管理

Mistral 类实现了上下文管理协议，并注册了一个终结器函数，用于关闭其内部使用的同步和异步 HTTPX 客户端。这将关闭 HTTP 连接，释放内存，并释放 SDK 占用的其他资源。对于仅执行少量 SDK 方法调用的短生命周期 Python 程序和笔记本而言，资源管理可能不是主要关注点。然而，在长期运行的程序中，通过 上下文管理器 创建单个 SDK 实例并在整个应用程序中重复使用它会更有益。

from mistralai.client import Mistral
import os
def main():

    with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:
        # 应用程序的其余部分在这里...


# 或者在使用异步时：
async def amain():

    async with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:
        # 应用程序的其余部分在这里...

调试

您可以配置 SDK 以输出 SDK 请求和响应的调试日志。

您可以直接将自己的日志记录器类传递给 SDK。

from mistralai.client import Mistral
import logging

logging.basicConfig(level=logging.DEBUG)
s = Mistral(debug_logger=logging.getLogger("mistralai.client"))

您还可以通过将环境变量 MISTRAL_DEBUG 设置为 true 来启用默认的调试日志记录器。

IDE 支持

PyCharm

通常，SDK 可以很好地与大多数 IDE 无缝配合。然而，在使用 PyCharm 时，通过安装一个额外的插件，您可以享受与 Pydantic 更好的集成。

• PyCharm Pydantic 插件

开发

贡献

虽然我们重视对本 SDK 的开源贡献，但此库是由程序自动生成的。任何手动添加到内部文件的更改都将在下一次生成时被覆盖。 我们期待您的反馈。欢迎随时提交 PR 或包含概念验证的议题，我们将尽力将其纳入未来的版本中。

Mistral Python Client 快速上手指南

环境准备

在开始之前，请确保满足以下要求：

• Python 版本：Python 3.10 或更高版本。
• API Key：您需要一个 Mistral AI API Key。
  1. 访问 Mistral 控制台 注册账号并获取 Key。
  2. 将 Key 设置为环境变量（以 zsh 为例）：

     echo 'export MISTRAL_API_KEY=[your_key_here]' >> ~/.zshenv
     source ~/.zshenv
     

• 包管理器：推荐使用 uv（速度更快），也支持 pip 或 poetry。

注意：如果您计划使用国内网络环境，建议在安装命令中指定清华或阿里镜像源以提升下载速度（见安装步骤）。

安装步骤

您可以选择以下任意一种方式安装 SDK：

方式一：使用 uv（推荐）

uv 是现代化的快速 Python 包管理器。

uv add mistralai

国内加速方案：

uv add mistralai --index-url https://pypi.tuna.tsinghua.edu.cn/simple

方式二：使用 pip

pip install mistralai

国内加速方案：

pip install mistralai -i https://pypi.tuna.tsinghua.edu.cn/simple

方式三：使用 Poetry

poetry add mistralai

可选：安装 Agents 额外依赖

如果您需要使用 Agent 相关功能，请安装额外依赖：

pip install "mistralai[agents]"

基本使用

以下示例展示如何初始化客户端并调用聊天补全接口。SDK 同时支持同步和异步调用。

同步调用示例

from mistralai.client import Mistral
import os

# 初始化客户端
with Mistral(
    api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:

    # 调用聊天接口
    res = mistral.chat.complete(
        model="mistral-large-latest", 
        messages=[
            {
                "role": "user",
                "content": "Who is the best French painter? Answer in one short sentence.",
            },
        ], 
        stream=False, 
        response_format={
            "type": "text",
        }
    )

    # 处理响应
    print(res)

异步调用示例

import asyncio
from mistralai.client import Mistral
import os

async def main():
    async with Mistral(
        api_key=os.getenv("MISTRAL_API_KEY", ""),
    ) as mistral:

        res = await mistral.chat.complete_async(
            model="mistral-large-latest", 
            messages=[
                {
                    "role": "user",
                    "content": "Who is the best French painter? Answer in one short sentence.",
                },
            ], 
            stream=False, 
            response_format={
                "type": "text",
            }
        )

        print(res)

asyncio.run(main())

其他常用功能简述

• 文件上传：使用 mistral.files.upload 方法。
• Embeddings：使用 mistral.embeddings.create 方法生成向量嵌入。
• 流式输出：将 stream 参数设为 True 即可启用服务器发送事件（SSE）流式传输。