模型

產品

定價

文檔

部落格

關於

聯繫

🎉 Hy4-preview可在 SiliconFlow 上使用。現在就試試看。

返回部落格

DeepSeek V4 Pro Function Calling: A Python Tutorial for Tool-Using Agents

2026年9月17日

DeepSeek V4 Pro function calling lets a model request external actions while your application controls what actually runs. This Python tutorial builds an agent that checks an order status, validates the generated arguments, executes an approved function, returns the result, and stops safely if the workflow does not finish within its limits.

As of September 2026, the SiliconFlow model string used here is deepseek-ai/DeepSeek-V4-Pro-0813.

What You Will Build: An Agent That Looks Up Order Status

The agent handles a request such as:

Where is order ORD-123456? The email is alex@example.com.

The request moves through four stages:

Your application sends the user message and an order-status tool definition to DeepSeek V4 Pro.
The model returns a structured request containing the tool name and JSON arguments.
Python validates the arguments and runs the approved function.
The tool result goes back to the model, which produces the customer-facing answer.

The model does not connect directly to the order database or execute Python. Your application remains responsible for authorization, validation, tool execution, and error handling.

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

Use these settings for the SiliconFlow deployment:

Setting	Value
Base URL	https://api.siliconflow.com/v1
Model ID	deepseek-ai/DeepSeek-V4-Pro-0813
API	Chat Completions
Tool type	Function

The DeepSeek-V4-Pro-0813 deployment supports tools through SiliconFlow’s serverless API. Model availability and identifiers can change, so confirm the model string in the Model Library before moving an integration into production. The current OpenAI Python library requires Python 3.10 or later. Install or update it with:

python -m pip install --upgrade

python -m pip install --upgrade

python -m pip install --upgrade

Create a SiliconFlow API key, then store it in an environment variable instead of placing it inside the script.

For macOS or Linux:

export SILICONFLOW_API_KEY="your-api-key"

export SILICONFLOW_API_KEY="your-api-key"

export SILICONFLOW_API_KEY="your-api-key"

For Windows PowerShell:

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

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

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

Define the Tool Schema and Python Function

A tool schema describes the function that the model may request. It specifies the function name, purpose, accepted fields, and required arguments.

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 tells the model not to add fields outside the schema. It does not replace runtime validation, because model-generated arguments may still be malformed or unsuitable for the application.

A dictionary can act as the order system while testing:

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"],
    }

The function returns the same response for an unknown order and an email mismatch. That prevents callers from using different email addresses to discover whether an order ID exists.

For a real application, replace the dictionary with a narrowly scoped service function. An authenticated customer ID is also a stronger authorization boundary than an email address supplied in a prompt.

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

Send the Request and Validate the Tool Arguments

Initialize the client with the SiliconFlow base URL:

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,
)

The timeout, retry count, and output limit used in this tutorial are example settings rather than platform requirements. Adjust them for the expected service latency and response length.

Send the user message together with the tool schema:

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

The Chat Completions API returns function arguments as a JSON-formatted string inside tool_calls. Parse the string and validate its contents before execution:

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"

The email pattern performs only a basic format check. It does not prove that the address exists or belongs to the requester. Authentication and authorization must be handled separately.

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

Execute the Tool and Return Its Result to the Model

Use an allowlist to decide which Python function may run:

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),
        }

Do not use eval() to execute model-generated function names or arguments. Parsing JSON and dispatching through an allowlist keeps the callable surface explicit.

After execution, preserve the assistant message and append a tool message:

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),
    }
)

Each result must use the tool_call_id from its corresponding request. The model can then connect the returned order data to the function it requested. Send the updated conversation back to DeepSeek V4 Pro. If the model has enough information, it will return a normal assistant message rather than another tool request.

Complete Python Example With Error Handling and Turn Limits

The complete DeepSeek agent loop below adds API retries, internal error logging, argument validation, tool-name allowlisting, agent-turn limits, and protection against exceeding the documented message count.

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, and max_tokens are application-level choices in this example. A production service should tune them according to request size, latency targets, tool cost, and the consequences of repeated execution.

Read-only lookups are relatively low risk. Tools that cancel orders, issue refunds, change customer details, send messages, or create financial commitments should require explicit authorization and, where appropriate, human confirmation.

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

Common Questions About DeepSeek V4 Pro Function Calling

Q1. Can I Use DeepSeek V4 Pro Tool Calling Without the OpenAI SDK?

Yes. You can send the same messages and tools fields directly to the SiliconFlow Chat Completions REST endpoint with an HTTP client such as requests. The OpenAI SDK mainly provides convenient request construction, response objects, retries, and exception classes.

Q2. How Should the Agent Resume After Asking for Missing Information?

Store the existing conversation, append the user’s next answer as a new role: "user" message, and call the model again. Recheck message and token limits before sending the expanded history rather than starting an unrelated order lookup.

Q3. Should Tool Results Stay in the Conversation History?

Yes, while the model still needs them to complete the active task. Retain the assistant tool request and matching tool result together. Remove or summarize older results when appropriate, and avoid keeping sensitive order data longer than the application requires.

Q4. How Should I Test the Agent Before Connecting a Real Order System?

Use mock records and cover successful lookups, unknown orders, mismatched emails, malformed JSON, missing fields, unexpected fields, unknown tools, repeated calls, API failures, and turn-limit exhaustion. Confirm that invalid requests never reach the real service function.