defmodule LangChain.ChatModels.ChatMistralAI do use Ecto.Schema require Logger import Ecto.Changeset alias __MODULE__ alias LangChain.Config alias LangChain.ChatModels.ChatModel alias LangChain.ChatModels.ChatOpenAI alias LangChain.Function alias LangChain.Message alias LangChain.Message.ContentPart alias LangChain.Message.ToolCall alias LangChain.Message.ToolResult alias LangChain.TokenUsage alias LangChain.MessageDelta alias LangChain.LangChainError alias LangChain.Utils alias LangChain.Callbacks @behaviour ChatModel @current_config_version 1 @receive_timeout 60_000 @default_endpoint "https://api.mistral.ai/v1/chat/completions" @primary_key false embedded_schema do field :endpoint, :string, default: @default_endpoint # The version/model of the Mistral API to use. field :model, :string field :api_key, :string, redact: true # Sampling temperature, 0..1 for Mistral field :temperature, :float, default: 0.9 # top_p param to shape token selection field :top_p, :float, default: 1.0 # Duration in milliseconds for the response to be received. field :receive_timeout, :integer, default: @receive_timeout # Maximum tokens to generate in the response field :max_tokens, :integer # Some Mistral deployments allow a "safe_prompt" option field :safe_prompt, :boolean, default: false # Optional random seed for reproducible outputs. field :random_seed, :integer # Whether to stream partial/delta responses field :stream, :boolean, default: false # For choosing a specific tool call (like forcing a function execution). field :tool_choice, :map # JSON Schema to validate the output format (for structured JSON output) field :json_schema, :map # Whether to force a JSON response format field :json_response, :boolean, default: false # A list of callback handlers field :callbacks, {:array, :map}, default: [] # Whether to allow parallel tool calls. Default is set to `true` according to the API field :parallel_tool_calls, :boolean, default: true # For help with debugging. It outputs the RAW Req response received and the # RAW Elixir map being submitted to the API. field :verbose_api, :boolean, default: false # Number of retries on closed-connection errors (stale pool). The initial # request always runs; this controls additional attempts only. field :retry_count, :integer, default: 2 end @type t :: %ChatMistralAI{} @create_fields [ :endpoint, :model, :api_key, :temperature, :top_p, :receive_timeout, :max_tokens, :safe_prompt, :random_seed, :stream, :tool_choice, :json_schema, :json_response, :parallel_tool_calls, :verbose_api, :retry_count ] @required_fields [ :model ] @spec get_api_key(t) :: String.t() defp get_api_key(%ChatMistralAI{api_key: api_key}) do # If no API key is set, fall back to the globally configured Mistral API key api_key || Config.resolve(:mistral_api_key) end @spec new(attrs :: map()) :: {:ok, t} | {:error, Ecto.Changeset.t()} def new(%{} = attrs \\ %{}) do %ChatMistralAI{} |> cast(attrs, @create_fields) |> common_validation() |> apply_action(:insert) end @spec new!(attrs :: map()) :: t() | no_return() def new!(attrs \\ %{}) do case new(attrs) do {:ok, chain} -> chain {:error, changeset} -> raise LangChainError, changeset end end defp common_validation(changeset) do changeset |> validate_required(@required_fields) end @doc """ Formats this struct plus the given messages and tools as a request payload. """ @spec for_api(t(), [Message.t()], ChatModel.tools()) :: %{atom() => any()} def for_api(%ChatMistralAI{} = mistral, messages, tools) do %{ model: mistral.model, temperature: mistral.temperature, top_p: mistral.top_p, safe_prompt: mistral.safe_prompt, stream: mistral.stream, # a single ToolResult can expand into multiple tool messages for Mistral messages: messages |> Enum.reduce([], fn m, acc -> case for_api(mistral, m) do %{} = data -> [data | acc] data when is_list(data) -> Enum.reverse(data) ++ acc end end) |> Enum.reverse() } |> Utils.conditionally_add_to_map(:random_seed, mistral.random_seed) |> Utils.conditionally_add_to_map(:max_tokens, mistral.max_tokens) |> Utils.conditionally_add_to_map(:tools, get_tools_for_api(mistral, tools)) |> Utils.conditionally_add_to_map(:tool_choice, get_tool_choice(mistral)) |> Utils.conditionally_add_to_map(:response_format, set_response_format(mistral)) |> Utils.conditionally_add_to_map(:parallel_tool_calls, mistral.parallel_tool_calls) end # Creates the response_format field for JSON output when json_response is true. # If json_schema is provided, it will be included in the response format. # # For Mistral, the format is as follows: # https://docs.mistral.ai/capabilities/structured-output/custom_structured_output/ # { # "type": "json_schema", # "json_schema": { # "schema": { ... }, # "name": "output", # "strict": true # } # } @spec set_response_format(t()) :: map() | nil defp set_response_format(%ChatMistralAI{json_response: true, json_schema: schema}) when is_map(schema) and map_size(schema) > 0 do # The schema should already be in the correct format schema end defp set_response_format(%ChatMistralAI{json_response: true}) do # For Mistral, when no schema is provided, we use json_object type %{ "type" => "json_object" } end defp set_response_format(%ChatMistralAI{}) do nil end # Add a more complete function to map tools. This mirrors ChatOpenAI approach. defp get_tools_for_api(%__MODULE__{} = _model, nil), do: [] defp get_tools_for_api(%__MODULE__{} = model, tools) when is_list(tools) do Enum.map(tools, fn %Function{} = function -> %{"type" => "function", "function" => for_api(model, function)} end) end defp get_tool_choice(%ChatMistralAI{ tool_choice: %{"type" => "function", "function" => %{"name" => name}} = _tool_choice }) when is_binary(name) and byte_size(name) > 0, do: %{"type" => "function", "function" => %{"name" => name}} defp get_tool_choice(%ChatMistralAI{tool_choice: %{"type" => type} = _tool_choice}) when is_binary(type) and byte_size(type) > 0, do: type defp get_tool_choice(%ChatMistralAI{}), do: nil @doc """ Converts a LangChain Message-based structure into the expected map of data for Mistral. We also include any `tool_calls` stored on the message. """ @spec for_api( struct(), Message.t() | ContentPart.t() | ToolCall.t() | ToolResult.t() | Function.t() ) :: %{String.t() => any()} | [%{String.t() => any()}] def for_api(%_{} = model, %Message{content: content} = msg) when is_binary(content) do role = get_message_role(model, msg.role) %{ "role" => role, "content" => msg.content } |> Utils.conditionally_add_to_map("name", msg.name) |> Utils.conditionally_add_to_map( "tool_calls", Enum.map(msg.tool_calls || [], &for_api(model, &1)) ) end def for_api(%_{} = model, %Message{role: :assistant, tool_calls: tool_calls} = msg) when is_list(tool_calls) do content = case msg.content do content when is_binary(content) -> content content when is_list(content) -> ContentPart.parts_to_string(content) nil -> nil end %{ "role" => :assistant, "content" => content } |> Utils.conditionally_add_to_map("tool_calls", Enum.map(tool_calls, &for_api(model, &1))) end def for_api(%_{} = _model, %Message{role: :user, content: content} = msg) when is_list(content) do # A user message can hold an array of ContentParts %{ "role" => msg.role, "content" => ContentPart.parts_to_string(content) } |> Utils.conditionally_add_to_map("name", msg.name) end # Handle messages with ContentPart content for non-user roles def for_api(%_{} = model, %Message{content: content} = msg) when is_list(content) do role = get_message_role(model, msg.role) %{ "role" => role, "content" => ContentPart.parts_to_string(content) } |> Utils.conditionally_add_to_map("name", msg.name) |> Utils.conditionally_add_to_map( "tool_calls", Enum.map(msg.tool_calls || [], &for_api(model, &1)) ) end # Handle ContentPart structures def for_api(%_{} = _model, %ContentPart{type: :text, content: content}) do content end # ToolResult => stand-alone message with "role: :tool" def for_api(%_{} = model, %ToolResult{type: :function} = result) do # a ToolResult becomes a stand-alone %Message{role: :tool} response. %{ "role" => :tool, "tool_call_id" => result.tool_call_id, "content" => content_for_tool_result(model, result.content) } end def for_api(%_{} = model, %Message{role: :tool, tool_results: tool_results} = _msg) when is_list(tool_results) do # ToolResults turn into a list of tool messages for Mistral Enum.map(tool_results, fn result -> %{ "role" => :tool, "tool_call_id" => result.tool_call_id, "content" => content_for_tool_result(model, result.content) } end) end # ToolCall => "function" style request def for_api(%_{} = _model, %ToolCall{type: :function} = fun) do %{ "id" => fun.call_id, "type" => "function", "function" => %{ "name" => fun.name, "arguments" => Jason.encode!(fun.arguments) } } end def for_api(_model, %Function{} = fun) do %{ "name" => fun.name, "description" => fun.description, "parameters" => fun.parameters_schema || %{} } end # Implementation only: more straightforward approach for Mistral defp get_message_role(%ChatMistralAI{}, role), do: role # Convert content to a format suitable for Mistral tool results. # Mistral expects a string, not a list of ContentParts. defp content_for_tool_result(_model, content) when is_list(content) do ContentPart.parts_to_string(content) end defp content_for_tool_result(_model, content) when is_binary(content), do: content defp content_for_tool_result(_model, nil), do: "" @doc """ Calls the Mistral API passing the ChatMistralAI struct plus either a simple string prompt or a list of messages as the prompt. Optionally pass in a list of tools. """ @impl ChatModel def call(%__MODULE__{} = mistralai, prompt, tools) when is_binary(prompt) and is_list(tools) do messages = [ Message.new_system!(), Message.new_user!(prompt) ] call(mistralai, messages, tools) end def call(%__MODULE__{} = mistralai, messages, tools) when is_list(messages) and is_list(tools) do metadata = %{ model: mistralai.model, message_count: length(messages), tools_count: length(tools) } LangChain.Telemetry.span([:langchain, :llm, :call], metadata, fn -> try do # Track the prompt being sent LangChain.Telemetry.llm_prompt( %{system_time: System.system_time()}, %{model: mistralai.model, messages: messages} ) case do_api_request(mistralai, messages, tools) do {:error, reason} -> {:error, reason} parsed_data -> # Track the response being received LangChain.Telemetry.llm_response( %{system_time: System.system_time()}, %{model: mistralai.model, response: parsed_data} ) {:ok, parsed_data} end rescue err in LangChainError -> {:error, err} end end) end # Make the API request. If `stream: true`, we handle partial chunk deltas; # otherwise, we parse a single complete body. @doc false @spec do_api_request(t(), [Message.t()], ChatModel.tools(), integer() | nil) :: list() | struct() | {:error, LangChainError.t()} def do_api_request(openai, messages, tools, retry_count \\ nil) def do_api_request(_mistralai, _messages, _tools, 0) do raise LangChainError, "Retries exceeded. Connection failed." end def do_api_request( %__MODULE__{stream: false} = mistralai, messages, tools, retry_count ) do retry_count = retry_count || mistralai.retry_count + 1 raw_data = for_api(mistralai, messages, tools) req = Req.new( url: mistralai.endpoint, json: raw_data, auth: {:bearer, get_api_key(mistralai)}, headers: [ {"api-key", get_api_key(mistralai)} ], receive_timeout: mistralai.receive_timeout, # Disable Req-level retry to prevent compounding with LangChain's own # :closed retry. See https://github.com/brainlid/langchain/issues/503 retry: false ) req |> Req.post() |> case do {:ok, %Req.Response{body: data} = response} -> Callbacks.fire(mistralai.callbacks, :on_llm_response_headers, [response.headers]) Callbacks.fire(mistralai.callbacks, :on_llm_token_usage, [ get_token_usage(data) ]) case do_process_response(mistralai, data) do {:error, %LangChainError{} = reason} -> {:error, reason} result -> # Track non-streaming response completion LangChain.Telemetry.emit_event( [:langchain, :llm, :response, :non_streaming], %{system_time: System.system_time()}, %{ model: mistralai.model, response_size: byte_size(inspect(result)) } ) Callbacks.fire(mistralai.callbacks, :on_llm_new_message, [result]) result end {:error, %Req.TransportError{reason: :timeout} = err} -> {:error, LangChainError.exception(type: "timeout", message: "Request timed out", original: err)} {:error, %Req.TransportError{reason: :closed}} -> Logger.debug(fn -> "Connection closed: retry count = #{inspect(retry_count)}" end) do_api_request(mistralai, messages, tools, retry_count - 1) other -> Logger.warning(fn -> "Unexpected and unhandled API response! #{inspect(other)}" end) other end end def do_api_request( %__MODULE__{stream: true} = mistralai, messages, tools, retry_count ) do retry_count = retry_count || mistralai.retry_count + 1 raw_data = for_api(mistralai, messages, tools) req = Req.new( url: mistralai.endpoint, json: raw_data, auth: {:bearer, get_api_key(mistralai)}, headers: [ {"api-key", get_api_key(mistralai)} ], receive_timeout: mistralai.receive_timeout ) req |> Req.post( into: Utils.handle_stream_fn( mistralai, # Mistral's streaming API is mostly compatible with OpenAI's, # so we can reuse the same decoder &ChatOpenAI.decode_stream/1, &do_process_response(mistralai, &1) ) ) |> case do {:ok, %Req.Response{body: data} = response} -> Callbacks.fire(mistralai.callbacks, :on_llm_response_headers, [response.headers]) data {:error, %Req.TransportError{reason: :timeout} = err} -> {:error, LangChainError.exception(type: "timeout", message: "Request timed out", original: err)} {:error, %Req.TransportError{reason: :closed}} -> Logger.debug(fn -> "Connection closed: retry count = #{inspect(retry_count)}" end) do_api_request(mistralai, messages, tools, retry_count - 1) other -> Logger.warning(fn -> "Unhandled and unexpected response from streamed call. #{inspect(other)}" end) {:error, LangChainError.exception( type: "unexpected_response", message: "Unexpected", original: other )} end end # Parse final or partial responses to produce the appropriate LangChain structure. @doc false @spec do_process_response(t(), data :: any()) :: :skip | :ok | TokenUsage.t() | Message.t() | [Message.t() | MessageDelta.t() | TokenUsage.t() | {:error, LangChainError.t()}] | MessageDelta.t() | [MessageDelta.t()] | {:error, LangChainError.t()} # The last chunk of the response contains both the final delta in the "choices" key, # and the token usage in the "usage" key def do_process_response(model, %{"choices" => choices, "usage" => %{} = _usage} = data) do case get_token_usage(data) do %TokenUsage{} = token_usage -> Callbacks.fire(model.callbacks, :on_llm_token_usage, [token_usage]) :ok nil -> :ok end Enum.map(choices, &do_process_response(model, &1)) end def do_process_response(_model, %{"choices" => []}), do: :skip def do_process_response(model, %{"choices" => choices}) when is_list(choices) do Enum.map(choices, &do_process_response(model, &1)) end def do_process_response(_model, %{"choices" => _not_a_list} = data) do Logger.warning(""" Mistral returned a response with a "choices" key that is not a list. data: #{inspect(data)} """) :skip end # Partial 'delta' format: look for any embedded "tool_calls" def do_process_response( model, %{ "delta" => delta_body, "finish_reason" => finish, "index" => index } ) do status = case finish do nil -> :incomplete "stop" -> :complete "length" -> :length "model_length" -> :length "tool_calls" -> :complete other -> Logger.warning("Unsupported finish_reason in delta message. Reason: #{inspect(other)}") nil end # If partial chunk references some tool calls tool_calls = case delta_body do %{"tool_calls" => calls} when is_list(calls) -> Enum.map(calls, &do_process_response(model, &1)) _ -> nil end # Validate that tool_calls is not empty when finish_reason is "tool_calls" # Mistral API sometimes returns finish_reason="tool_calls" in streaming deltas # but without actual tool_calls data, which causes message ordering errors if finish == "tool_calls" and (tool_calls == nil or tool_calls == []) do error_msg = "Mistral API returned finish_reason='tool_calls' in delta but tool_calls is empty. " <> "Delta content: #{inspect(delta_body["content"])}, index: #{inspect(index)}" Logger.warning(error_msg) {:error, LangChainError.exception( type: "invalid_tool_calls", message: error_msg, original: %{"delta" => delta_body, "finish_reason" => finish, "index" => index} )} else role = case delta_body do %{"role" => role} -> role # Mistral doesn't include a `role` key in the delta. # Defaulting to `:assistant`. seems like it makes sense. _ -> "assistant" end # Convert Mistral's content format to ContentPart structs content = process_mistral_content(delta_body["content"]) # Adjust index to prevent thinking and text content from merging. # Thinking is always at index 0. Text content is offset by 1 to avoid collision. # Similar to DeepSeek: thinking at index 0, text starts at index 1. adjusted_index = case content do %ContentPart{type: :thinking} -> 0 _ -> (index || 0) + 1 end data = delta_body |> Map.put("role", role) |> Map.put("index", adjusted_index) |> Map.put("status", status) |> Map.put("tool_calls", tool_calls) |> Map.put("content", content) case MessageDelta.new(data) do {:ok, message} -> message {:error, %Ecto.Changeset{} = changeset} -> {:error, LangChainError.exception(changeset)} end end end # Complete message with tool calls: def do_process_response( model, %{"finish_reason" => finish_reason, "message" => %{"tool_calls" => calls} = message} = data ) when finish_reason in ["tool_calls", "stop"] do tool_calls = if is_list(calls), do: Enum.map(calls, &do_process_response(model, &1)), else: [] # Validate that tool_calls is not empty when finish_reason is "tool_calls" # Mistral API sometimes returns finish_reason="tool_calls" with empty or # malformed tool_calls, which causes message ordering errors when sent back if finish_reason == "tool_calls" and Enum.empty?(tool_calls) do error_msg = "Mistral API returned finish_reason='tool_calls' but tool_calls is empty. " <> "Content: #{inspect(message["content"])}" Logger.warning(error_msg) {:error, LangChainError.exception( type: "invalid_tool_calls", message: error_msg, original: data )} else case Message.new(%{ "role" => "assistant", "content" => message["content"], "complete" => true, "index" => data["index"], "tool_calls" => tool_calls }) do {:ok, msg} -> msg {:error, %Ecto.Changeset{} = changeset} -> {:error, LangChainError.exception(changeset)} end end end # Complete message without tool calls (e.g., Azure Foundry responses that omit the "tool_calls" key) def do_process_response( _model, %{"finish_reason" => finish_reason, "message" => message} = data ) when finish_reason in ["stop", "length", "model_length"] do status = case finish_reason do "stop" -> :complete "length" -> :length "model_length" -> :length end case Message.new(%{ "role" => message["role"] || "assistant", "content" => message["content"], "status" => status, "index" => data["index"] }) do {:ok, msg} -> msg {:error, %Ecto.Changeset{} = changeset} -> {:error, LangChainError.exception(changeset)} end end # Tool call from a complete message def do_process_response( _model, %{ "function" => %{"arguments" => args, "name" => name}, "id" => call_id, "index" => _maybe_index } = _data ) do case ToolCall.new(%{ type: :function, status: :complete, name: name, arguments: args, call_id: call_id }) do {:ok, %ToolCall{} = call} -> call {:error, %Ecto.Changeset{} = changeset} -> {:error, LangChainError.exception(changeset)} end end def do_process_response(_model, %{ "function" => %{"arguments" => args, "name" => name}, "id" => call_id, "type" => "function" }) do case ToolCall.new(%{ type: :function, status: :complete, name: name, arguments: args, call_id: call_id }) do {:ok, %ToolCall{} = call} -> call {:error, %Ecto.Changeset{} = changeset} -> {:error, LangChainError.exception(changeset)} end end def do_process_response(_model, %{"error" => %{"message" => reason}} = response) do {:error, LangChainError.exception(message: reason, original: response)} end # Handle Mistral's error format: %{"object" => "error", "message" => "...", "type" => "...", "code" => "..."} def do_process_response( _model, %{"object" => "error", "message" => reason, "type" => type} = response ) do {:error, LangChainError.exception(type: type, message: reason, original: response)} end def do_process_response(_model, {:error, %Jason.DecodeError{} = response}) do error_message = "Received invalid JSON: #{inspect(response)}" {:error, LangChainError.exception(type: "invalid_json", message: error_message, original: response)} end def do_process_response(_model, other) do {:error, LangChainError.exception( type: "unexpected_response", message: "Unexpected response", original: other )} end # Process Mistral's content format for thinking blocks and text in list format. # Mistral can return: # - Thinking: [%{"type" => "thinking", "thinking" => [%{"text" => "...", "type" => "text"}]}] # - Text: [%{"type" => "text", "text" => "..."}] defp process_mistral_content(nil), do: nil defp process_mistral_content(content) when is_binary(content), do: content defp process_mistral_content([%{"type" => "thinking", "thinking" => thinking_list} | _]) when is_list(thinking_list) do # Extract text from thinking array and convert to ContentPart thinking_text = thinking_list |> Enum.filter(&match?(%{"type" => "text"}, &1)) |> Enum.map(&Map.get(&1, "text", "")) |> Enum.join("") ContentPart.thinking!(thinking_text) end defp process_mistral_content([%{"type" => "text", "text" => text} | _]) do ContentPart.text!(text) end # For any other content format, pass through unchanged defp process_mistral_content(content), do: content defp get_token_usage(%{"usage" => usage} = _response_body) do # extract out the reported response token usage # # https://platform.mistralai.com/docs/api-reference/chat/object#chat/object-usage TokenUsage.new!(%{ input: Map.get(usage, "prompt_tokens"), output: Map.get(usage, "completion_tokens"), raw: usage }) end defp get_token_usage(_response_body), do: nil @doc """ Determine if an error should be retried. If `true`, a fallback LLM may be used. If `false`, the error is understood to be more fundamental with the request rather than a service issue and it should not be retried or fallback to another service. """ @impl ChatModel @spec retry_on_fallback?(LangChainError.t()) :: boolean() def retry_on_fallback?(%LangChainError{type: "rate_limited"}), do: true def retry_on_fallback?(%LangChainError{type: "rate_limit_exceeded"}), do: true def retry_on_fallback?(%LangChainError{type: "timeout"}), do: true def retry_on_fallback?(%LangChainError{type: "too_many_requests"}), do: true def retry_on_fallback?(%LangChainError{type: "unreachable_backend"}), do: true def retry_on_fallback?(%LangChainError{type: "server_error"}), do: true def retry_on_fallback?(%LangChainError{type: "service_unavailable"}), do: true def retry_on_fallback?(_), do: false @doc """ Generate a config map that can later restore the model's configuration. """ @impl ChatModel @spec serialize_config(t()) :: %{String.t() => any()} def serialize_config(%ChatMistralAI{} = model) do Utils.to_serializable_map( model, [ :endpoint, :model, :temperature, :top_p, :receive_timeout, :max_tokens, :safe_prompt, :random_seed, :stream, :json_schema, :json_response, :verbose_api ], @current_config_version ) end @doc """ Restores the model from the config map. """ @impl ChatModel def restore_from_map(%{"version" => 1} = data) do ChatMistralAI.new(data) end end