Modelos

Productos

Precios

Docs

Blog

Sobre nosotros

Contacto

🎉Hy4-preview está disponible en SiliconFlow. Pruébalo AHORA.

Volver a los blogs

Llamada a funciones con DeepSeek V4 Pro: un tutorial de Python para agentes que utilizan herramientas

17 sept 2026

Índice de contenidos

Llamada a funciones de DeepSeek V4 Pro: un tutorial de Python para agentes que utilizan herramientas

La llamada a funciones de DeepSeek V4 Pro permite que un modelo solicite acciones externas mientras su aplicación controla lo que realmente se ejecuta. Este tutorial de Python crea un agente que verifica el estado de un pedido, valida los argumentos generados, ejecuta una función aprobada, devuelve el resultado y se detiene de manera segura si el flujo de trabajo no finaliza dentro de sus límites.

A partir de septiembre de 2026, la cadena del modelo de SiliconFlow utilizada aquí es deepseek-ai/DeepSeek-V4-Pro-0813.

Lo que creará: un agente que busca el estado de un pedido

El agente gestiona una solicitud como:

¿Dónde está el pedido ORD-123456? El correo electrónico es alex@example.com.

La solicitud pasa por cuatro etapas:

Su aplicación envía el mensaje del usuario y la definición de una herramienta de estado de pedido a DeepSeek V4 Pro.
El modelo devuelve una solicitud estructurada que contiene el nombre de la herramienta y los argumentos JSON.
Python valida los argumentos y ejecuta la función aprobada.
El resultado de la herramienta regresa al modelo, el cual produce la respuesta orientada al cliente.

El modelo no se conecta directamente a la base de datos de pedidos ni ejecuta Python. Su aplicación sigue siendo responsable de la autorización, validación, ejecución de herramientas y gestión de errores.

DeepSeek V4 Pro function calling workflow showing a user request, the model tool request, argument validation, tool execution, and the final customer answer

Utilice estos ajustes para el despliegue en SiliconFlow:

Ajuste	Valor
URL base	https://api.siliconflow.com/v1
ID del modelo	deepseek-ai/DeepSeek-V4-Pro-0813
API	Chat Completions
Tipo de herramienta	Function

El despliegue de DeepSeek-V4-Pro-0813 admite herramientas a través de la API Serverless de SiliconFlow. La disponibilidad y los identificadores del modelo pueden cambiar, así que confirme la cadena del modelo en la Biblioteca de Modelos antes de pasar una integración a producción. La biblioteca de Python de OpenAI actual requiere Python 3.10 o posterior. Instálela o actualícela con:

python -m pip install --upgrade

python -m pip install --upgrade

python -m pip install --upgrade

Cree una clave de API de SiliconFlow y luego almacénela en una variable de entorno en lugar de colocarla dentro del script.

Para macOS o Linux:

export SILICONFLOW_API_KEY="your-api-key"

export SILICONFLOW_API_KEY="your-api-key"

export SILICONFLOW_API_KEY="your-api-key"

Para Windows PowerShell:

$env:SILICONFLOW_API_KEY="your-api-key"

$env:SILICONFLOW_API_KEY="your-api-key"

$env:SILICONFLOW_API_KEY="your-api-key"

Definir el esquema de la herramienta y la función de Python

El esquema de una herramienta describe la función que el modelo puede solicitar. Especifica el nombre de la función, su propósito, los campos aceptados y los argumentos requeridos.

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": (
                "Look up the current status of an order after matching "
                "the order ID and customer email address."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": (
                            "Order ID in the format ORD-123456."
                        ),
                    },
                    "email": {
                        "type": "string",
                        "description": (
                            "Email address used to place the order."
                        ),
                    },
                },
                "required": ["order_id", "email"],
                "additionalProperties": False,
            },
        },
    }
]

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": (
                "Look up the current status of an order after matching "
                "the order ID and customer email address."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": (
                            "Order ID in the format ORD-123456."
                        ),
                    },
                    "email": {
                        "type": "string",
                        "description": (
                            "Email address used to place the order."
                        ),
                    },
                },
                "required": ["order_id", "email"],
                "additionalProperties": False,
            },
        },
    }
]

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": (
                "Look up the current status of an order after matching "
                "the order ID and customer email address."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": (
                            "Order ID in the format ORD-123456."
                        ),
                    },
                    "email": {
                        "type": "string",
                        "description": (
                            "Email address used to place the order."
                        ),
                    },
                },
                "required": ["order_id", "email"],
                "additionalProperties": False,
            },
        },
    }
]

additionalProperties: false le indica al modelo que no agregue campos fuera del esquema. Esto no reemplaza la validación en tiempo de ejecución, ya que los argumentos generados por el modelo aún podrían estar mal formados o no ser adecuados para la aplicación.

Un diccionario puede actuar como sistema de pedidos durante las pruebas:

MOCK_ORDERS = {
    "ORD-123456": {
        "email": "alex@example.com",
        "status": "in_transit",
        "estimated_delivery": "2026-09-11",
    }
}
def get_order_status(order_id, email):
    order = MOCK_ORDERS.get(order_id)
    if (
        order is None
        or order["email"].casefold() != email.casefold()
    ):
        return {
            "ok": False,
            "error": "order_not_found",
            "message": "No matching order was found.",
        }
    return {
        "ok": True,
        "order_id": order_id,
        "status": order["status"],
        "estimated_delivery": order[
            "estimated_delivery"
        ],
    }

MOCK_ORDERS = {
    "ORD-123456": {
        "email": "alex@example.com",
        "status": "in_transit",
        "estimated_delivery": "2026-09-11",
    }
}
def get_order_status(order_id, email):
    order = MOCK_ORDERS.get(order_id)
    if (
        order is None
        or order["email"].casefold() != email.casefold()
    ):
        return {
            "ok": False,
            "error": "order_not_found",
            "message": "No matching order was found.",
        }
    return {
        "ok": True,
        "order_id": order_id,
        "status": order["status"],
        "estimated_delivery": order[
            "estimated_delivery"
        ],
    }

MOCK_ORDERS = {
    "ORD-123456": {
        "email": "alex@example.com",
        "status": "in_transit",
        "estimated_delivery": "2026-09-11",
    }
}
def get_order_status(order_id, email):
    order = MOCK_ORDERS.get(order_id)
    if (
        order is None
        or order["email"].casefold() != email.casefold()
    ):
        return {
            "ok": False,
            "error": "order_not_found",
            "message": "No matching order was found.",
        }
    return {
        "ok": True,
        "order_id": order_id,
        "status": order["status"],
        "estimated_delivery": order[
            "estimated_delivery"
        ],
    }

La función devuelve la misma respuesta para un pedido desconocido que para una discrepancia de correo electrónico. Esto evita que los emisores de llamadas utilicen diferentes direcciones de correo electrónico para averiguar si existe un ID de pedido.

Para una aplicación real, reemplace el diccionario con una función de servicio con un alcance bien definido. Un ID de cliente autenticado también representa un límite de autorización más sólido que una dirección de correo electrónico proporcionada en un prompt.

Structure of the get_order_status tool schema with required order_id and email fields alongside a mock order lookup function

Enviar la solicitud y validar los argumentos de la herramienta

Inicialice el cliente con la URL base de SiliconFlow:

import os
from openai import OpenAI
client = OpenAI(
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url="https://api.siliconflow.com/v1",
    timeout=30.0,
    max_retries=0,
)

import os
from openai import OpenAI
client = OpenAI(
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url="https://api.siliconflow.com/v1",
    timeout=30.0,
    max_retries=0,
)

import os
from openai import OpenAI
client = OpenAI(
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url="https://api.siliconflow.com/v1",
    timeout=30.0,
    max_retries=0,
)

El tiempo de espera, el número de reintentos y el límite de salida utilizados en este tutorial son configuraciones de ejemplo y no requisitos de la plataforma. Ajústelos según la latencia de servicio esperada y la longitud de la respuesta.

Envíe el mensaje del usuario junto con el esquema de la herramienta:

response = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Pro-0813",
    messages=[
        {
            "role": "system",
            "content": (
                "You help customers check orders. Use "
                "get_order_status when both an order ID and email "
                "are available. Ask for any missing field. Never "
                "invent order data."
            ),
        },
        {
            "role": "user",
            "content": (
                "Where is order ORD-123456? "
                "The email is alex@example.com."
            ),
        },
    ],
    tools=TOOLS,
    temperature=0,
    max_tokens=800,
    stream=False,
)
assistant_message = response.choices[0].message

response = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Pro-0813",
    messages=[
        {
            "role": "system",
            "content": (
                "You help customers check orders. Use "
                "get_order_status when both an order ID and email "
                "are available. Ask for any missing field. Never "
                "invent order data."
            ),
        },
        {
            "role": "user",
            "content": (
                "Where is order ORD-123456? "
                "The email is alex@example.com."
            ),
        },
    ],
    tools=TOOLS,
    temperature=0,
    max_tokens=800,
    stream=False,
)
assistant_message = response.choices[0].message

response = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Pro-0813",
    messages=[
        {
            "role": "system",
            "content": (
                "You help customers check orders. Use "
                "get_order_status when both an order ID and email "
                "are available. Ask for any missing field. Never "
                "invent order data."
            ),
        },
        {
            "role": "user",
            "content": (
                "Where is order ORD-123456? "
                "The email is alex@example.com."
            ),
        },
    ],
    tools=TOOLS,
    temperature=0,
    max_tokens=800,
    stream=False,
)
assistant_message = response.choices[0].message

La API de Chat Completions devuelve los argumentos de la función como una cadena con formato JSON dentro de tool_calls. Analice la cadena y valide su contenido antes de la ejecución:

import json
import re
def validate_order_arguments(raw_arguments):
    try:
        data = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Tool arguments were not valid JSON."
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(
            "Tool arguments must be a JSON object."
        )
    required = {"order_id", "email"}
    missing = required - data.keys()
    unexpected = data.keys() - required
    if missing:
        raise ValueError(
            "Missing required field(s): "
            + ", ".join(sorted(missing))
        )
    if unexpected:
        raise ValueError(
            "Unexpected field(s): "
            + ", ".join(sorted(unexpected))
        )
    order_id = data["order_id"]
    email = data["email"]
    if not isinstance(order_id, str) or not re.fullmatch(
        r"ORD-\d{6}", order_id
    ):
        raise ValueError(
            "order_id must match ORD-123456."
        )
    if not isinstance(email, str) or not re.fullmatch(
        r"[^\s@]+@[^\s@]+\.[^\s@]+", email
    ):
        raise ValueError(
            "email must match a basic email address format."
        )
    return {
        "order_id": order_id,
        "email"

import json
import re
def validate_order_arguments(raw_arguments):
    try:
        data = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Tool arguments were not valid JSON."
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(
            "Tool arguments must be a JSON object."
        )
    required = {"order_id", "email"}
    missing = required - data.keys()
    unexpected = data.keys() - required
    if missing:
        raise ValueError(
            "Missing required field(s): "
            + ", ".join(sorted(missing))
        )
    if unexpected:
        raise ValueError(
            "Unexpected field(s): "
            + ", ".join(sorted(unexpected))
        )
    order_id = data["order_id"]
    email = data["email"]
    if not isinstance(order_id, str) or not re.fullmatch(
        r"ORD-\d{6}", order_id
    ):
        raise ValueError(
            "order_id must match ORD-123456."
        )
    if not isinstance(email, str) or not re.fullmatch(
        r"[^\s@]+@[^\s@]+\.[^\s@]+", email
    ):
        raise ValueError(
            "email must match a basic email address format."
        )
    return {
        "order_id": order_id,
        "email"

import json
import re
def validate_order_arguments(raw_arguments):
    try:
        data = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Tool arguments were not valid JSON."
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(
            "Tool arguments must be a JSON object."
        )
    required = {"order_id", "email"}
    missing = required - data.keys()
    unexpected = data.keys() - required
    if missing:
        raise ValueError(
            "Missing required field(s): "
            + ", ".join(sorted(missing))
        )
    if unexpected:
        raise ValueError(
            "Unexpected field(s): "
            + ", ".join(sorted(unexpected))
        )
    order_id = data["order_id"]
    email = data["email"]
    if not isinstance(order_id, str) or not re.fullmatch(
        r"ORD-\d{6}", order_id
    ):
        raise ValueError(
            "order_id must match ORD-123456."
        )
    if not isinstance(email, str) or not re.fullmatch(
        r"[^\s@]+@[^\s@]+\.[^\s@]+", email
    ):
        raise ValueError(
            "email must match a basic email address format."
        )
    return {
        "order_id": order_id,
        "email"

El patrón de correo electrónico realiza únicamente una comprobación básica de formato. No demuestra que la dirección exista o pertenezca a quien la solicita. La autenticación y la autorización deben gestionarse de forma independiente.

Argument validation flow that parses the JSON string, checks required and unexpected fields, and verifies the order ID and email formats

Ejecutar la herramienta y devolver su resultado al modelo

Utilice una lista de permitidos para decidir qué función de Python se puede ejecutar:

def execute_tool_call(tool_call):
    name = tool_call.function.name
    if name != "get_order_status":
        return {
            "ok": False,
            "error": "unknown_tool",
        }
    try:
        arguments = validate_order_arguments(
            tool_call.function.arguments
        )
        return get_order_status(**arguments)
    except ValueError as exc:
        return {
            "ok": False,
            "error": "invalid_arguments",
            "message": str(exc),
        }

def execute_tool_call(tool_call):
    name = tool_call.function.name
    if name != "get_order_status":
        return {
            "ok": False,
            "error": "unknown_tool",
        }
    try:
        arguments = validate_order_arguments(
            tool_call.function.arguments
        )
        return get_order_status(**arguments)
    except ValueError as exc:
        return {
            "ok": False,
            "error": "invalid_arguments",
            "message": str(exc),
        }

def execute_tool_call(tool_call):
    name = tool_call.function.name
    if name != "get_order_status":
        return {
            "ok": False,
            "error": "unknown_tool",
        }
    try:
        arguments = validate_order_arguments(
            tool_call.function.arguments
        )
        return get_order_status(**arguments)
    except ValueError as exc:
        return {
            "ok": False,
            "error": "invalid_arguments",
            "message": str(exc),
        }

No utilice eval() para ejecutar nombres de funciones o argumentos generados por el modelo. Analizar JSON y distribuirlos a través de una lista de permitidos mantiene explícita la superficie invocable.

Tras la ejecución, conserve el mensaje del asistente y añada un mensaje de herramienta:

messages.append(
    assistant_message.model_dump(exclude_none=True)
)
tool_result = execute_tool_call(
    assistant_message.tool_calls[0]
)
messages.append(
    {
        "role": "tool",
        "tool_call_id": (
            assistant_message.tool_calls[0].id
        ),
        "content": json.dumps(tool_result),
    }
)

messages.append(
    assistant_message.model_dump(exclude_none=True)
)
tool_result = execute_tool_call(
    assistant_message.tool_calls[0]
)
messages.append(
    {
        "role": "tool",
        "tool_call_id": (
            assistant_message.tool_calls[0].id
        ),
        "content": json.dumps(tool_result),
    }
)

messages.append(
    assistant_message.model_dump(exclude_none=True)
)
tool_result = execute_tool_call(
    assistant_message.tool_calls[0]
)
messages.append(
    {
        "role": "tool",
        "tool_call_id": (
            assistant_message.tool_calls[0].id
        ),
        "content": json.dumps(tool_result),
    }
)

Cada resultado debe utilizar el tool_call_id de su solicitud correspondiente. Así, el modelo podrá vincular los datos del pedido devueltos con la función que solicitó. Envíe la conversación actualizada de nuevo a DeepSeek V4 Pro. Si el modelo dispone de suficiente información, devolverá un mensaje de asistente normal en lugar de otra solicitud de herramienta.

Ejemplo completo en Python con gestión de errores y límites de turnos

El bucle completo del agente DeepSeek que se muestra a continuación añade reintentos de API, registro de errores internos, validación de argumentos, lista de permitidos para nombres de herramientas, límites de turnos del agente y protección para no superar el número de mensajes documentado.

import json
import logging
import os
import re
import time
from openai import (
    APIConnectionError,
    APIStatusError,
    OpenAI,
    RateLimitError,
)
MODEL = "deepseek-ai/DeepSeek-V4-Pro-0813"
MAX_AGENT_TURNS = 4
MAX_MESSAGES = 10
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = OpenAI(
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url="https://api.siliconflow.com/v1",
    timeout=30.0,
    max_retries=0,
)
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": (
                "Look up the current status of an order after "
                "matching the order ID and customer email address."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": (
                            "Order ID in the format ORD-123456."
                        ),
                    },
                    "email": {
                        "type": "string",
                        "description": (
                            "Email address used to place the order."
                        ),
                    },
                },
                "required": ["order_id", "email"],
                "additionalProperties": False,
            },
        },
    }
]
# Tutorial-only data. Replace it with a narrowly scoped
# service function connected to your order system.
MOCK_ORDERS = {
    "ORD-123456": {
        "email": "alex@example.com",
        "status": "in_transit",
        "estimated_delivery": "2026-09-11",
    }
}
def validate_order_arguments(raw_arguments):
    try:
        data = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Tool arguments were not valid JSON."
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(
            "Tool arguments must be a JSON object."
        )
    required = {"order_id", "email"}
    missing = required - data.keys()
    unexpected = data.keys() - required
    if missing:
        raise ValueError(
            "Missing required field(s): "
            + ", ".join(sorted(missing))
        )
    if unexpected:
        raise ValueError(
            "Unexpected field(s): "
            + ", ".join(sorted(unexpected))
        )
    order_id = data["order_id"]
    email = data["email"]
    if not isinstance(order_id, str) or not re.fullmatch(
        r"ORD-\d{6}", order_id
    ):
        raise ValueError(
            "order_id must match ORD-123456."
        )
    if not isinstance(email, str) or not re.fullmatch(
        r"[^\s@]+@[^\s@]+\.[^\s@]+", email
    ):
        raise ValueError(
            "email must match a basic email address format."
        )
    return {
        "order_id": order_id,
        "email": email.casefold(),
    }
def get_order_status(order_id, email):
    order = MOCK_ORDERS.get(order_id)
    if (
        order is None
        or order["email"].casefold() != email.casefold()
    ):
        return {
            "ok": False,
            "error": "order_not_found",
            "message": "No matching order was found.",
        }
    return {
        "ok": True,
        "order_id": order_id,
        "status": order["status"],
        "estimated_delivery": order[
            "estimated_delivery"
        ],
    }
def execute_tool_call(tool_call):
    name = tool_call.function.name
    if name != "get_order_status":
        return {
            "ok": False,
            "error": "unknown_tool",
        }
    try:
        arguments = validate_order_arguments(
            tool_call.function.arguments
        )
        return get_order_status(**arguments)
    except ValueError as exc:
        return {
            "ok": False,
            "error": "invalid_arguments",
            "message": str(exc),
        }
    except Exception:
        logger.exception(
            "Order status tool execution failed"
        )
        return {
            "ok": False,
            "error": "tool_execution_failed",
            "message": (
                "The order service is temporarily unavailable."
            ),
        }
def create_completion(messages, max_attempts=3):
    if not 1 <= len(messages) <= MAX_MESSAGES:
        raise RuntimeError(
            f"Message count must be between 1 "
            f"and {MAX_MESSAGES}."
        )
    for attempt in range(max_attempts):
        try:
            return client.chat.completions.create(
                model=MODEL,
                messages=messages,
                tools=TOOLS,
                temperature=0,
                max_tokens=800,
                stream=False,
            )
        except (RateLimitError, APIConnectionError):
            if attempt == max_attempts - 1:
                raise
        except APIStatusError as exc:
            if exc.status_code not in {
                500,
                502,
                503,
                504,
            }:
                raise
            if attempt == max_attempts - 1:
                raise
        time.sleep(2**attempt)
    raise RuntimeError(
        "Completion failed without returning or raising."
    )
def run_order_agent(
    user_message,
    max_turns=MAX_AGENT_TURNS,
):
    messages = [
        {
            "role": "system",
            "content": (
                "You help customers check orders. Use "
                "get_order_status when both an order ID and "
                "email are available. Ask for any missing "
                "field. Never invent order data."
            ),
        },
        {
            "role": "user",
            "content": user_message,
        },
    ]
    for _ in range(max_turns):
        response = create_completion(messages)
        assistant_message = response.choices[0].message
        # Preserve the complete assistant message, including
        # its tool calls and call IDs.
        messages.append(
            assistant_message.model_dump(
                exclude_none=True
            )
        )
        if not assistant_message.tool_calls:
            return (
                assistant_message.content
                or "The model returned no response."
            )
        if (
            len(messages)
            + len(assistant_message.tool_calls)
            > MAX_MESSAGES
        ):
            raise RuntimeError(
                "The agent cannot return all tool results "
                "without exceeding the API message limit."
            )
        for tool_call in assistant_message.tool_calls:
            tool_result = execute_tool_call(tool_call)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(tool_result),
                }
            )
    raise RuntimeError(
        "The agent reached its turn limit before "
        "producing a final answer."
    )
if __name__ == "__main__":
    try:
        answer = run_order_agent(
            "Where is order ORD-123456? "
            "The email is alex@example.com."
        )
        print(answer)
    except (
        RateLimitError,
        APIConnectionError,
        APIStatusError,
    ) as exc:
        print(
            f"SiliconFlow API request failed: {exc}"
        )
    except RuntimeError as exc:
        print(f"Agent stopped: {exc}")

import json
import logging
import os
import re
import time
from openai import (
    APIConnectionError,
    APIStatusError,
    OpenAI,
    RateLimitError,
)
MODEL = "deepseek-ai/DeepSeek-V4-Pro-0813"
MAX_AGENT_TURNS = 4
MAX_MESSAGES = 10
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = OpenAI(
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url="https://api.siliconflow.com/v1",
    timeout=30.0,
    max_retries=0,
)
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": (
                "Look up the current status of an order after "
                "matching the order ID and customer email address."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": (
                            "Order ID in the format ORD-123456."
                        ),
                    },
                    "email": {
                        "type": "string",
                        "description": (
                            "Email address used to place the order."
                        ),
                    },
                },
                "required": ["order_id", "email"],
                "additionalProperties": False,
            },
        },
    }
]
# Tutorial-only data. Replace it with a narrowly scoped
# service function connected to your order system.
MOCK_ORDERS = {
    "ORD-123456": {
        "email": "alex@example.com",
        "status": "in_transit",
        "estimated_delivery": "2026-09-11",
    }
}
def validate_order_arguments(raw_arguments):
    try:
        data = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Tool arguments were not valid JSON."
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(
            "Tool arguments must be a JSON object."
        )
    required = {"order_id", "email"}
    missing = required - data.keys()
    unexpected = data.keys() - required
    if missing:
        raise ValueError(
            "Missing required field(s): "
            + ", ".join(sorted(missing))
        )
    if unexpected:
        raise ValueError(
            "Unexpected field(s): "
            + ", ".join(sorted(unexpected))
        )
    order_id = data["order_id"]
    email = data["email"]
    if not isinstance(order_id, str) or not re.fullmatch(
        r"ORD-\d{6}", order_id
    ):
        raise ValueError(
            "order_id must match ORD-123456."
        )
    if not isinstance(email, str) or not re.fullmatch(
        r"[^\s@]+@[^\s@]+\.[^\s@]+", email
    ):
        raise ValueError(
            "email must match a basic email address format."
        )
    return {
        "order_id": order_id,
        "email": email.casefold(),
    }
def get_order_status(order_id, email):
    order = MOCK_ORDERS.get(order_id)
    if (
        order is None
        or order["email"].casefold() != email.casefold()
    ):
        return {
            "ok": False,
            "error": "order_not_found",
            "message": "No matching order was found.",
        }
    return {
        "ok": True,
        "order_id": order_id,
        "status": order["status"],
        "estimated_delivery": order[
            "estimated_delivery"
        ],
    }
def execute_tool_call(tool_call):
    name = tool_call.function.name
    if name != "get_order_status":
        return {
            "ok": False,
            "error": "unknown_tool",
        }
    try:
        arguments = validate_order_arguments(
            tool_call.function.arguments
        )
        return get_order_status(**arguments)
    except ValueError as exc:
        return {
            "ok": False,
            "error": "invalid_arguments",
            "message": str(exc),
        }
    except Exception:
        logger.exception(
            "Order status tool execution failed"
        )
        return {
            "ok": False,
            "error": "tool_execution_failed",
            "message": (
                "The order service is temporarily unavailable."
            ),
        }
def create_completion(messages, max_attempts=3):
    if not 1 <= len(messages) <= MAX_MESSAGES:
        raise RuntimeError(
            f"Message count must be between 1 "
            f"and {MAX_MESSAGES}."
        )
    for attempt in range(max_attempts):
        try:
            return client.chat.completions.create(
                model=MODEL,
                messages=messages,
                tools=TOOLS,
                temperature=0,
                max_tokens=800,
                stream=False,
            )
        except (RateLimitError, APIConnectionError):
            if attempt == max_attempts - 1:
                raise
        except APIStatusError as exc:
            if exc.status_code not in {
                500,
                502,
                503,
                504,
            }:
                raise
            if attempt == max_attempts - 1:
                raise
        time.sleep(2**attempt)
    raise RuntimeError(
        "Completion failed without returning or raising."
    )
def run_order_agent(
    user_message,
    max_turns=MAX_AGENT_TURNS,
):
    messages = [
        {
            "role": "system",
            "content": (
                "You help customers check orders. Use "
                "get_order_status when both an order ID and "
                "email are available. Ask for any missing "
                "field. Never invent order data."
            ),
        },
        {
            "role": "user",
            "content": user_message,
        },
    ]
    for _ in range(max_turns):
        response = create_completion(messages)
        assistant_message = response.choices[0].message
        # Preserve the complete assistant message, including
        # its tool calls and call IDs.
        messages.append(
            assistant_message.model_dump(
                exclude_none=True
            )
        )
        if not assistant_message.tool_calls:
            return (
                assistant_message.content
                or "The model returned no response."
            )
        if (
            len(messages)
            + len(assistant_message.tool_calls)
            > MAX_MESSAGES
        ):
            raise RuntimeError(
                "The agent cannot return all tool results "
                "without exceeding the API message limit."
            )
        for tool_call in assistant_message.tool_calls:
            tool_result = execute_tool_call(tool_call)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(tool_result),
                }
            )
    raise RuntimeError(
        "The agent reached its turn limit before "
        "producing a final answer."
    )
if __name__ == "__main__":
    try:
        answer = run_order_agent(
            "Where is order ORD-123456? "
            "The email is alex@example.com."
        )
        print(answer)
    except (
        RateLimitError,
        APIConnectionError,
        APIStatusError,
    ) as exc:
        print(
            f"SiliconFlow API request failed: {exc}"
        )
    except RuntimeError as exc:
        print(f"Agent stopped: {exc}")

import json
import logging
import os
import re
import time
from openai import (
    APIConnectionError,
    APIStatusError,
    OpenAI,
    RateLimitError,
)
MODEL = "deepseek-ai/DeepSeek-V4-Pro-0813"
MAX_AGENT_TURNS = 4
MAX_MESSAGES = 10
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = OpenAI(
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url="https://api.siliconflow.com/v1",
    timeout=30.0,
    max_retries=0,
)
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": (
                "Look up the current status of an order after "
                "matching the order ID and customer email address."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": (
                            "Order ID in the format ORD-123456."
                        ),
                    },
                    "email": {
                        "type": "string",
                        "description": (
                            "Email address used to place the order."
                        ),
                    },
                },
                "required": ["order_id", "email"],
                "additionalProperties": False,
            },
        },
    }
]
# Tutorial-only data. Replace it with a narrowly scoped
# service function connected to your order system.
MOCK_ORDERS = {
    "ORD-123456": {
        "email": "alex@example.com",
        "status": "in_transit",
        "estimated_delivery": "2026-09-11",
    }
}
def validate_order_arguments(raw_arguments):
    try:
        data = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "Tool arguments were not valid JSON."
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(
            "Tool arguments must be a JSON object."
        )
    required = {"order_id", "email"}
    missing = required - data.keys()
    unexpected = data.keys() - required
    if missing:
        raise ValueError(
            "Missing required field(s): "
            + ", ".join(sorted(missing))
        )
    if unexpected:
        raise ValueError(
            "Unexpected field(s): "
            + ", ".join(sorted(unexpected))
        )
    order_id = data["order_id"]
    email = data["email"]
    if not isinstance(order_id, str) or not re.fullmatch(
        r"ORD-\d{6}", order_id
    ):
        raise ValueError(
            "order_id must match ORD-123456."
        )
    if not isinstance(email, str) or not re.fullmatch(
        r"[^\s@]+@[^\s@]+\.[^\s@]+", email
    ):
        raise ValueError(
            "email must match a basic email address format."
        )
    return {
        "order_id": order_id,
        "email": email.casefold(),
    }
def get_order_status(order_id, email):
    order = MOCK_ORDERS.get(order_id)
    if (
        order is None
        or order["email"].casefold() != email.casefold()
    ):
        return {
            "ok": False,
            "error": "order_not_found",
            "message": "No matching order was found.",
        }
    return {
        "ok": True,
        "order_id": order_id,
        "status": order["status"],
        "estimated_delivery": order[
            "estimated_delivery"
        ],
    }
def execute_tool_call(tool_call):
    name = tool_call.function.name
    if name != "get_order_status":
        return {
            "ok": False,
            "error": "unknown_tool",
        }
    try:
        arguments = validate_order_arguments(
            tool_call.function.arguments
        )
        return get_order_status(**arguments)
    except ValueError as exc:
        return {
            "ok": False,
            "error": "invalid_arguments",
            "message": str(exc),
        }
    except Exception:
        logger.exception(
            "Order status tool execution failed"
        )
        return {
            "ok": False,
            "error": "tool_execution_failed",
            "message": (
                "The order service is temporarily unavailable."
            ),
        }
def create_completion(messages, max_attempts=3):
    if not 1 <= len(messages) <= MAX_MESSAGES:
        raise RuntimeError(
            f"Message count must be between 1 "
            f"and {MAX_MESSAGES}."
        )
    for attempt in range(max_attempts):
        try:
            return client.chat.completions.create(
                model=MODEL,
                messages=messages,
                tools=TOOLS,
                temperature=0,
                max_tokens=800,
                stream=False,
            )
        except (RateLimitError, APIConnectionError):
            if attempt == max_attempts - 1:
                raise
        except APIStatusError as exc:
            if exc.status_code not in {
                500,
                502,
                503,
                504,
            }:
                raise
            if attempt == max_attempts - 1:
                raise
        time.sleep(2**attempt)
    raise RuntimeError(
        "Completion failed without returning or raising."
    )
def run_order_agent(
    user_message,
    max_turns=MAX_AGENT_TURNS,
):
    messages = [
        {
            "role": "system",
            "content": (
                "You help customers check orders. Use "
                "get_order_status when both an order ID and "
                "email are available. Ask for any missing "
                "field. Never invent order data."
            ),
        },
        {
            "role": "user",
            "content": user_message,
        },
    ]
    for _ in range(max_turns):
        response = create_completion(messages)
        assistant_message = response.choices[0].message
        # Preserve the complete assistant message, including
        # its tool calls and call IDs.
        messages.append(
            assistant_message.model_dump(
                exclude_none=True
            )
        )
        if not assistant_message.tool_calls:
            return (
                assistant_message.content
                or "The model returned no response."
            )
        if (
            len(messages)
            + len(assistant_message.tool_calls)
            > MAX_MESSAGES
        ):
            raise RuntimeError(
                "The agent cannot return all tool results "
                "without exceeding the API message limit."
            )
        for tool_call in assistant_message.tool_calls:
            tool_result = execute_tool_call(tool_call)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(tool_result),
                }
            )
    raise RuntimeError(
        "The agent reached its turn limit before "
        "producing a final answer."
    )
if __name__ == "__main__":
    try:
        answer = run_order_agent(
            "Where is order ORD-123456? "
            "The email is alex@example.com."
        )
        print(answer)
    except (
        RateLimitError,
        APIConnectionError,
        APIStatusError,
    ) as exc:
        print(
            f"SiliconFlow API request failed: {exc}"
        )
    except RuntimeError as exc:
        print(f"Agent stopped: {exc}")

MAX_AGENT_TURNS, MAX_MESSAGES, max_attempts, timeout y max_tokens son opciones a nivel de aplicación en este ejemplo. Un servicio en producción debería ajustarlos de acuerdo con el tamaño de la solicitud, los objetivos de latencia, el coste de la herramienta y las consecuencias de una ejecución repetida.

Las búsquedas de solo lectura conllevan un riesgo relativamente bajo. Las herramientas que cancelan pedidos, emiten reembolsos, cambian datos de clientes, envían mensajes o crean compromisos financieros deberían requerir una autorización explícita y, cuando sea apropiado, confirmación humana.

Complete DeepSeek agent loop with API retries, tool-name allowlisting, agent-turn limits, and message-count protection

Preguntas frecuentes sobre la llamada a funciones de DeepSeek V4 Pro

P1. ¿Puedo utilizar la llamada a herramientas de DeepSeek V4 Pro sin el SDK de OpenAI?

Sí. Puede enviar los mismos campos de mensajes y herramientas directamente al punto de conexión REST de Chat Completions de SiliconFlow con un cliente HTTP como requests. El SDK de OpenAI proporciona principalmente comodidad para la construcción de solicitudes, objetos de respuesta, reintentos y clases de excepciones.

P2. ¿Cómo debe reanudar el agente después de solicitar la información que falta?

Guarde la conversación existente, añada la siguiente respuesta del usuario como un nuevo mensaje con el rol: "user", y vuelva a llamar al modelo. Vuelva a comprobar los límites de mensajes y tokens antes de enviar el historial ampliado, en lugar de iniciar una búsqueda de pedido no relacionada.

P3. ¿Deben permanecer los resultados de la herramienta en el historial de conversación?

Sí, mientras el modelo los necesite para completar la tarea activa. Conserve juntos la solicitud de la herramienta del asistente y el resultado correspondiente de la misma. Elimine o resuma los resultados más antiguos cuando proceda, y evite guardar datos confidenciales de pedidos durante más tiempo del que requiera la aplicación.

P4. ¿Cómo debo probar el agente antes de conectarlo a un sistema de pedidos real?

Utilice registros ficticios (mock) y cubra búsquedas correctas, pedidos desconocidos, correos electrónicos que no coinciden, JSON mal formado, campos ausentes, campos inesperados, herramientas desconocidas, llamadas repetidas, fallos de la API y agotamiento del límite de turnos. Confirme que las solicitudes no válidas nunca lleguen a la función de servicio real.