nndeploy.api_llm.llm_chat¶

Functions

call_llm(params, messages, *[, ...])

Call an OpenAI-compatible Chat Completions endpoint.

Classes

`LLMChatNode`(name[, inputs, outputs])	LLM chat node (OpenAI-compatible) Input: str / list[messages] / dict({"messages":[...]}) Output: str (model generated text)
`LLMChatNodeCreator`()
`LLMParams`([api_base, api_key_env, model, ...])

class nndeploy.api_llm.llm_chat.LLMParams(api_base: str = 'https://api.openai.com', api_key_env: str = 'OPENAI_API_KEY', model: str = 'gpt-4o-mini', temperature: float = 0.2, max_tokens: int = 512, system_prompt: str = 'You are a helpful assistant.', timeout_s: float = 60.0, retry: int = 2, retry_backoff_s: float = 1.2, stream: bool = False, response_json_mode: bool = False, json_schema: Optional[dict] = None)[源代码]¶

基类：object

api_base: str = 'https://api.openai.com'¶

api_key_env: str = 'OPENAI_API_KEY'¶

model: str = 'gpt-4o-mini'¶

temperature: float = 0.2¶

max_tokens: int = 512¶

system_prompt: str = 'You are a helpful assistant.'¶

timeout_s: float = 60.0¶

retry: int = 2¶

retry_backoff_s: float = 1.2¶

stream: bool = False¶

response_json_mode: bool = False¶

json_schema: Optional[dict] = None¶

__init__(api_base: str = 'https://api.openai.com', api_key_env: str = 'OPENAI_API_KEY', model: str = 'gpt-4o-mini', temperature: float = 0.2, max_tokens: int = 512, system_prompt: str = 'You are a helpful assistant.', timeout_s: float = 60.0, retry: int = 2, retry_backoff_s: float = 1.2, stream: bool = False, response_json_mode: bool = False, json_schema: Optional[dict] = None) → None¶

nndeploy.api_llm.llm_chat.call_llm(params: LLMParams, messages: list, *, api_key_override: str = None, node_for_interrupt: Optional[Node] = None, emit_usage_to_log: bool = True, stream_print: bool = False, proxies: Optional[dict] = None, verify: Optional[Union[bool, str]] = None) → str[源代码]¶: Call an OpenAI-compatible Chat Completions endpoint. - Supports non-stream and stream modes. - Respects 429 Retry-After; uses exponential backoff with jitter. - Logs usage if available. Returns the final concatenated content as a string.

class nndeploy.api_llm.llm_chat.LLMChatNode(name, inputs: list[nndeploy.dag.edge.Edge] = None, outputs: list[nndeploy.dag.edge.Edge] = None)[源代码]¶

基类：Node

LLM chat node (OpenAI-compatible) Input: str / list[messages] / dict({“messages”:[…]}) Output: str (model generated text)

__init__(name, inputs: list[nndeploy.dag.edge.Edge] = None, outputs: list[nndeploy.dag.edge.Edge] = None)[源代码]¶

Initialize the node.

参数:

name – The name of the node
inputs – List of input edges or None
outputs – List of output edges or None

run()[源代码]¶: Execute the node’s main computation.

serialize()[源代码]¶: Serialize the node to a JSON string.

deserialize(target: str)[源代码]¶: Deserialize the node from a JSON string.

add_dropdown_param(dropdown_param: str, dropdown_values: list)¶: Add a dropdown parameter.

add_io_param(io_param: str)¶: Add an IO parameter.

add_required_param(required_param: str)¶: Add a required parameter.

add_resource_with_state(key: str, edge: Edge)¶: Add stateful resources using key and edge.

add_resource_without_state(key: str, value)¶: Add stateless resources using key-value pairs.

add_ui_param(ui_param: str)¶: Add a UI parameter.

check_inputs(inputs) → bool¶: Validate the input data.

check_outputs(outputs_name) → bool¶: Validate the output names.

clear_dropdown_params()¶: Clear the list of dropdown parameters.

clear_io_params()¶: Clear the list of IO parameters.

clear_required_params()¶: Clear the list of required parameters.

clear_ui_params()¶: Clear the list of UI parameters.

create_internal_output_edge(self: nndeploy._nndeploy_internal.dag.Node, name: str) → nndeploy::dag::Edge¶

create_resource_with_state(key: str) → Edge¶: Create stateful resources and return an edge pointer.

default_param(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Status¶

deinit()¶: Deinitialize the node and clean up resources.

forward(*args, **kwargs)¶

Overloaded function.

forward(self: nndeploy._nndeploy_internal.dag.Node, inputs: list[nndeploy::dag::Edge]) -> list[nndeploy::dag::Edge]
forward(self: nndeploy._nndeploy_internal.dag.Node) -> list[nndeploy::dag::Edge]
forward(self: nndeploy._nndeploy_internal.dag.Node, input: nndeploy::dag::Edge) -> list[nndeploy::dag::Edge]

get_all_input()¶: Get all inputs of the node.

get_all_output()¶: Get all outputs of the node.

get_completed_size() → int¶: Get the completed size of the node.

get_composite_node(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy::dag::CompositeNode¶

get_constructed() → bool¶: Check if the node has been constructed.

get_debug_flag() → bool¶: Check if debug mode is enabled.

get_desc() → str¶: Get the description of the node.

get_developer() → str¶: Get the developer information of the node.

get_device_type() → DeviceType¶: Get the device type of the node.

get_dropdown_params(self: nndeploy._nndeploy_internal.dag.Node) → dict[str, list[str]]¶

get_external_param(key: str)¶: Get external parameters by key.

get_graph()¶: Get the parent graph of this node.

get_graph_flag() → bool¶: Check if the graph flag is set.

get_initialized() → bool¶: Check if the node has been initialized.

get_input(index: int = 0)¶: Get input at the specified index.

get_input_count(self: nndeploy._nndeploy_internal.dag.Node) → int¶

get_input_data(index: int = 0)¶: Get input data at the specified index.

get_input_index(self: nndeploy._nndeploy_internal.dag.Node, name: str) → int¶

get_input_name(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → str¶

get_input_names(self: nndeploy._nndeploy_internal.dag.Node) → list[str]¶

get_input_type_info() → EdgeTypeInfo¶: Get the input type information of the node.

get_io_params() → list¶: Get the list of IO parameters.

get_io_type()¶: Get the I/O type of the node.

get_key() → str¶: Get the unique key of the node.

get_loop_count() → int¶: Get the loop count of the node.

get_memory_size() → int¶: Get the memory size required by the node.

get_name() → str¶: Get the name of the node.

get_node_type()¶: Get the node type.

get_output(index: int = 0)¶: Get output at the specified index.

get_output_count(self: nndeploy._nndeploy_internal.dag.Node) → int¶

get_output_index(self: nndeploy._nndeploy_internal.dag.Node, name: str) → int¶

get_output_name(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → str¶

get_output_names(self: nndeploy._nndeploy_internal.dag.Node) → list[str]¶

get_output_type_info() → EdgeTypeInfo¶: Get the output type information of the node.

get_parallel_type() → ParallelType¶: Get the parallel execution type of the node.

get_param()¶: Get the parameters of the node.

get_real_outputs_name()¶: Get the actual output names.

get_required_params() → list¶: Get the list of required parameters.

get_resource_with_state(key: str) → Edge¶: Get stateful resource edge by key.

get_resource_without_state(key: str)¶: Get stateless resources by key.

get_run_size() → int¶: Get the run size of the node.

get_run_status() → RunStatus¶: Get the run status of the node.

get_source() → str¶: Get the source code repository information of the node.

get_stream()¶: Get the execution stream of the node.

get_time_profile_flag() → bool¶: Check if time profiling is enabled.

get_trace_flag() → bool¶: Check if tracing is enabled.

get_ui_params() → list¶: Get the list of UI parameters.

get_version() → str¶: Get the version of the node.

init()¶: Initialize the node.

is_dynamic_input(self: nndeploy._nndeploy_internal.dag.Node) → bool¶

is_dynamic_output(self: nndeploy._nndeploy_internal.dag.Node) → bool¶

is_inputs_changed(self: nndeploy._nndeploy_internal.dag.Node, inputs: list[nndeploy::dag::Edge]) → bool¶

is_running() → bool¶: Check if the node is currently running.

load_file(path: str)¶: Load the node configuration from a file.

remove_dropdown_param(dropdown_param: str)¶: Remove a dropdown parameter.

remove_io_param(io_param: str)¶: Remove an IO parameter.

remove_required_param(required_param: str)¶: Remove a required parameter.

remove_ui_param(ui_param: str)¶: Remove a UI parameter.

save_file(path: str)¶: Save the node configuration to a file.

set_composite_node(self: nndeploy._nndeploy_internal.dag.Node, composite_node: nndeploy::dag::CompositeNode) → nndeploy._nndeploy_internal.base.Status¶

set_debug_flag(flag: bool)¶: Enable or disable debug mode for the node.

set_desc(desc: str)¶: Set the description of the node.

set_developer(developer: str)¶: Set the developer information for the node.

set_device_type(device_type: DeviceType)¶: Set the device type for node execution.

set_dropdown_params(dropdown_params: dict)¶: Set the list of dropdown parameters.

set_dynamic_input(self: nndeploy._nndeploy_internal.dag.Node, is_dynamic_input: bool) → None¶

set_dynamic_output(self: nndeploy._nndeploy_internal.dag.Node, is_dynamic_output: bool) → None¶

set_external_param(key: str, external_param)¶: Set external parameters using key-value pairs.

set_graph(graph)¶: Set the parent graph for this node.

set_graph_flag(flag: bool)¶: Set the graph flag for the node.

set_initialized_flag(flag: bool)¶: Set the initialization flag for the node.

set_inner_flag(flag: bool)¶: Set the inner flag for the node.

set_input(input, index: int = -1)¶: Set input at the specified index.

set_input_name(self: nndeploy._nndeploy_internal.dag.Node, name: str, index: int = 0) → nndeploy._nndeploy_internal.base.Status¶

set_input_names(self: nndeploy._nndeploy_internal.dag.Node, names: list[str]) → nndeploy._nndeploy_internal.base.Status¶

set_input_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, input: nndeploy::dag::Edge, index: int = -1) → nndeploy._nndeploy_internal.base.Status¶

set_input_type(input_type: type, desc: str = '')¶

Set the input type for the node.

参数:: input_type – The input type
返回:: Status code

set_input_type_info(input_type_info: EdgeTypeInfo, desc: str = '')¶: Set the input type information for the node.

set_inputs(inputs)¶: Set all inputs for the node.

set_inputs_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, inputs: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.base.Status¶

set_io_params(io_params: list)¶: Set the list of IO parameters.

set_io_type(io_type)¶: Set the I/O type for the node.

set_iter_input(input, index: int = -1)¶: Set iterative input at the specified index.

set_key(key: str)¶: Set the unique key for the node.

set_loop_count(loop_count: int)¶: Set the loop count for the node.

set_memory(buffer: Buffer)¶: Set the memory buffer for the node.

set_name(name: str)¶: Set the name of the node.

set_node_type(node_type)¶: Set the node type.

set_output(output, index: int = -1)¶: Set output at the specified index.

set_output_data(output, index: int = 0, is_external: bool = True)¶: Set output data at the specified index.

set_output_name(self: nndeploy._nndeploy_internal.dag.Node, name: str, index: int = 0) → nndeploy._nndeploy_internal.base.Status¶

set_output_names(self: nndeploy._nndeploy_internal.dag.Node, names: list[str]) → nndeploy._nndeploy_internal.base.Status¶

set_output_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, output: nndeploy::dag::Edge, index: int = -1) → nndeploy._nndeploy_internal.base.Status¶

set_output_type(output_type: type, desc: str = '')¶: Set the output type for the node.

set_output_type_info(output_type_info: EdgeTypeInfo, desc: str = '')¶: Set the output type information for the node.

set_outputs(outputs)¶: Set all outputs for the node.

set_outputs_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, outputs: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.base.Status¶

set_parallel_type(parallel_type: ParallelType)¶: Set the parallel execution type for the node.

set_param(param)¶: Set the parameters for the node.

set_required_params(required_params: list)¶: Set the list of required parameters.

set_running_flag(flag: bool)¶: Set the running flag for the node.

set_source(source: str)¶: Set the source code repository information for the node.

set_stream(stream)¶: Set the execution stream for the node.

set_time_profile_flag(flag: bool)¶: Enable or disable time profiling for the node.

set_trace_flag(flag: bool)¶: Enable or disable tracing for the node.

set_ui_params(ui_params: list)¶: Set the list of UI parameters.

set_version(version: str)¶: Set the version of the node.

synchronize() → bool¶: Synchronize the node execution.

to_static_graph(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Status¶

update_input()¶: Update the input data for the node.

class nndeploy.api_llm.llm_chat.LLMChatNodeCreator[源代码]¶

基类：NodeCreator

__init__(self: nndeploy._nndeploy_internal.dag.NodeCreator) → None[源代码]¶

create_node(name: str, inputs: list[nndeploy.dag.edge.Edge], outputs: list[nndeploy.dag.edge.Edge])[源代码]¶

Create a new node instance.

This method must be implemented by subclasses.

参数:

name – The name of the node
inputs – List of input edges
outputs – List of output edges

返回:

The created node instance

create_node_shared_ptr(self: nndeploy._nndeploy_internal.dag.NodeCreator, arg0: str, arg1: list[nndeploy::dag::Edge], arg2: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.dag.Node¶