fabricatio_core.rust

Attributes

`CONFIG`
`GENERIC_BLOCK_TYPE`
`ROUTER`
`TEMPLATE_MANAGER`
`generic_parser`
`json_parser`
`logger`
`python_parser`
`router_usage`
`snippet_parser`

Classes

`CodeBlockParser`
`CodeSnippet`	Represents a code snippet extracted from text.
`CodeSnippetParser`
`Config`	Configuration structure containing all system components.
`ContentBlockParser`
`DebugConfig`
`DeploymentConfig`	Configuration for a specific deployment.
`EmbeddingConfig`	Embedding configuration structure.
`EmitterConfig`	Pymitter configuration structure.
`Event`
`GeneralConfig`	General configuration structure for application-wide settings.
`GenericBlockParser`
`JsonParser`
`LLMConfig`	Configuration for Language Learning Models (LLMs) like OpenAI's GPT.
`Logger`
`ProviderConfig`	Configuration for a specific provider.
`RerankerConfig`	Reranker configuration structure.
`Router`
`RouterUsage`
`RoutingConfig`	Routing configuration structure for controlling request dispatching behavior.
`SecretStr`
`TemplateConfig`	Template configuration structure.
`TemplateManager`	Python bindings for the TemplateManager struct.
`TemplateManagerConfig`
`TextCapturer`
`ProviderType`	Enum representing supported LLM provider types.
`TaskStatus`	Create a collection of name/value pairs.

Functions

`blake3_hash`(→ str)	Calculates a BLAKE3 hash of the given content.
`detect_language`(→ str)	Detects the language of a given string and returns its full native name.
`extra_satisfied`(→ bool)	Checks if a specific extra (optional dependency) of a Python package is satisfied.
`extras_satisfied`(→ bool)	Checks if all specified extras (optional dependencies) of a Python package are satisfied.
`is_arabic`(→ bool)	Checks if the given string is written in Arabic.
`is_chinese`(→ bool)	Checks if the given string is written in Simplified Chinese.
`is_dutch`(→ bool)	Checks if the given string is written in Dutch.
`is_english`(→ bool)	Checks if the given string is written in English.
`is_french`(→ bool)	Checks if the given string is written in French.
`is_german`(→ bool)	Checks if the given string is written in German.
`is_hindi`(→ bool)	Checks if the given string is written in Hindi.
`is_installed`(→ bool)	Checks if a Python package is installed.
`is_italian`(→ bool)	Checks if the given string is written in Italian.
`is_japanese`(→ bool)	Checks if the given string is written in Japanese.
`is_korean`(→ bool)	Checks if the given string is written in Korean.
`is_likely_text`(→ bool)	Determines if a file is likely text content.
`is_portuguese`(→ bool)	Checks if the given string is written in Portuguese.
`is_russian`(→ bool)	Checks if the given string is written in Russian.
`is_swedish`(→ bool)	Checks if the given string is written in Swedish.
`is_turkish`(→ bool)	Checks if the given string is written in Turkish.
`is_vietnamese`(→ bool)	Checks if the given string is written in Vietnamese.
`list_installed`(→ list[str])	Lists all installed Python packages.
`split_into_chunks`(→ list[str])	Splits a string into chunks based on maximum size and overlapping rate.
`split_sentence_bounds`(→ list[str])	Splits a string into sentences using Unicode sentence boundaries.
`split_word_bounds`(→ list[str])	Splits a string into words using Unicode word boundaries.
`tokens_of`(→ int)	Counts the number of tokens in a text string.
`word_count`(→ int)	Counts the number of words in a string.

Package Contents

fabricatio_core.rust.CONFIG: Config

fabricatio_core.rust.GENERIC_BLOCK_TYPE: str

fabricatio_core.rust.ROUTER: Router

fabricatio_core.rust.TEMPLATE_MANAGER: TemplateManager

fabricatio_core.rust.generic_parser: GenericBlockParser

fabricatio_core.rust.json_parser: JsonParser

fabricatio_core.rust.logger: Logger

fabricatio_core.rust.python_parser: CodeBlockParser

fabricatio_core.rust.router_usage: RouterUsage

fabricatio_core.rust.snippet_parser: CodeSnippetParser

class fabricatio_core.rust.CodeBlockParser

property language: str

static with_language(language: str) → CodeBlockParser

Create a new CodeBlockParser instance.

Parameters:: language (Option<&str>) – The programming language of the code block. Capture all kinds of code block if it set to None.
Returns:: A new CodeBlockParser instance.
Return type:: PyResult<Self>

static capture_python() → CodeBlockParser

capture(text: str) → str | None

Capture the first code block match in the text.

Returns the captured code block content or None if no match is found.

capture_all(text: str) → list[str]

Capture all code block matches in the text.

Returns a vector of captured code block contents.

class fabricatio_core.rust.CodeSnippet

Represents a code snippet extracted from text.

Contains its source code, programming language, and the target file path for writing.

property source: str: The source code content of the snippet.

property language: str: The programming language of the snippet.

property write_to: pathlib.Path: The file path where the snippet should be written.

write(parent_dirs: bool = True) → None

Writes the code snippet to its designated file path.

Parameters:: parent_dirs – Whether to create parent directories if they don’t exist.
Returns:: PyResult<()> indicating success.

class fabricatio_core.rust.CodeSnippetParser

property left_sep: str

property right_sep: str

static with_separators(left_sep: str = '>>>>>', right_sep: str = '<<<<<') → CodeSnippetParser

Create a new CodeSnippetParser instance.

Parameters:

left_sep (&str) – The left separator marking the start of the snippet.
right_sep (&str) – The right separator marking the end of the snippet.

Returns:

A new CodeSnippetParser instance.

Return type:

PyResult<Self>

static default() → CodeSnippetParser

parse(text: str) → list[CodeSnippet]

Parse text into path-content pairs.

Captures all snippet matches from the text and groups them into pairs, where each pair consists of a path and its corresponding content.

Returns:: A vector of tuples containing the path and content for each matched snippet.
Return type:: Vec<(PathBuf, String)>

class fabricatio_core.rust.Config

Configuration structure containing all system components.

property embedding: EmbeddingConfig: Embedding configuration parameters.

property reranker: RerankerConfig: Reranker configuration parameters.

property llm: LLMConfig: Language Learning Model settings with validation rules.

property debug: DebugConfig: Debug settings containing log level and verbosity.

property templates: TemplateConfig: Template paths/names for various operations.

property template_manager: TemplateManagerConfig: Template loading and management settings.

property routing: RoutingConfig: Request routing and load balancing settings.

property general: GeneralConfig: Global behavior configuration options.

property emitter: EmitterConfig: Event emission control settings.

load(name: str, config_cls: Type[_T]) → _T: Load configuration data for a given section name and instantiate a Python class.

class fabricatio_core.rust.ContentBlockParser

property left_delimiter: str

property right_delimiter: str

static with_delimiters(left_delimiter: str, right_delimiter: str | None = None) → ContentBlockParser

Create a new ContentBlockParser instance.

Parameters:

left_delimiter (&str) – The left delimiter marking the start of the content.
right_delimiter (Option<&str>) – The right delimiter marking the end of the content. Defaults to left_delimiter if not provided.

Returns:

A new ContentBlockParser instance.

Return type:

PyResult<Self>

capture(text: str) → str | None

Capture the first content block match in the text.

Returns the captured content or None if no match is found.

capture_all(text: str) → list[str]

Capture all content block matches in the text.

Returns a vector of captured contents.

class fabricatio_core.rust.DebugConfig

property log_level: str

property log_dir: pathlib.Path | None

property rotation: str | None

class fabricatio_core.rust.DeploymentConfig

Configuration for a specific deployment.

Defines the identity, grouping, and rate limits for a deployed service instance.

property id: str: Unique identifier for the deployment.

property group: str: Name of the route group this deployment belongs to.

property tpm: int | None: Optional quota limit for tokens per minute (TPM).

property rpm: int | None: Optional quota limit for requests per minute (RPM).

class fabricatio_core.rust.EmbeddingConfig

Embedding configuration structure.

property send_to: str | None

property no_cache: bool | None

property ndim: int | None

class fabricatio_core.rust.EmitterConfig

Pymitter configuration structure.

Contains settings for controlling event emission and listener behavior

property delimiter: str: The delimiter used to separate the event name into segments.

class fabricatio_core.rust.Event

property segments: list[str]

static instantiate_from(event: List[str] | str | Event) → Event

Creates an Event from various input types.

Parameters:: event – A string, list of strings, or another Event instance.
Returns:: A new Event instance with segments extracted from the input.

static quick_instantiate(event: List[str] | str | Event) → Event

Creates an Event with wildcard and pending status appended.

Parameters:: event – A string, list of strings, or another Event instance.
Returns:: A new Event instance with “*” and “Pending” segments appended.

derive(event: List[str] | str | Event) → Event

Derives a new event by appending segments from another event.

Parameters:: event – A string, list of strings, or another Event instance to append.
Returns:: A new Event with the combined segments.

collapse() → str

Collapses the event segments into a single delimited string.

Returns:: A string with segments joined by the configured delimiter.

fork() → Event

Creates a copy of the event.

Returns:: A clone of this Event instance.

push(segment: TaskStatus | str) → Event

Pushes a segment onto the event.

Parameters:: segment – A TaskStatus enum or string to append.
Returns:: A mutable reference to this Event instance.

push_wildcard() → Event

Appends a wildcard segment to the event.

Returns:: A mutable reference to this Event instance.

push_pending() → Event

Appends a Pending status segment to the event.

Returns:: A mutable reference to this Event instance.

push_running() → Event

Appends a Running status segment to the event.

Returns:: A mutable reference to this Event instance.

push_finished() → Event

Appends a Finished status segment to the event.

Returns:: A mutable reference to this Event instance.

push_failed() → Event

Appends a Failed status segment to the event.

Returns:: A mutable reference to this Event instance.

push_cancelled() → Event

Appends a Cancelled status segment to the event.

Returns:: A mutable reference to this Event instance.

pop() → str | None

Removes and returns the last segment.

Returns:: The last segment if present, None otherwise.

clear() → Event

Clears all segments from the event.

Returns:: A mutable reference to this Event instance.

concat(event: List[str] | str | Event) → Event

Concatenates another event’s segments onto this event.

Parameters:: event – A string, list of strings, or another Event instance to append.
Returns:: A mutable reference to this Event instance with combined segments.

__hash__() → int

Computes the hash of the collapsed event string.

Returns:: The hash value as a u64.

__richcmp__(other: Any, op: int) → bool

Compares this event with another value for equality.

Parameters:

other – Another Event instance or string to compare against.
op – The comparison operation (Eq or Ne).

Returns:

True if the comparison holds, False otherwise.

class fabricatio_core.rust.GeneralConfig

General configuration structure for application-wide settings.

property use_json_repair: bool: Whether to automatically repair malformed JSON.

class fabricatio_core.rust.GenericBlockParser

property block_type: str

static with_block_type(block_type: str = 'String') → GenericBlockParser

Create a new GenericBlockParser instance.

Parameters:: block_type (&str) – The type identifier of the generic block.
Returns:: A new GenericBlockParser instance.
Return type:: PyResult<Self>

static capture_generic_string() → GenericBlockParser

capture(text: str) → str | None

Capture the first generic block match in the text.

Returns the captured block content or None if no match is found.

capture_all(text: str) → list[str]

Capture all generic block matches in the text.

Returns a vector of captured block contents.

class fabricatio_core.rust.JsonParser

static with_pattern(pattern: str) → JsonParser

Creates a JsonParser with a custom regex pattern.

Parameters:: pattern – The regex pattern to use for capturing JSON.
Returns:: A new JsonParser instance.

static with_capturer(capturer: TextCapturer) → JsonParser

Creates a JsonParser with an existing TextCapturer.

Parameters:: capturer – The TextCapturer to use.
Returns:: A new JsonParser instance.

static capture_json_codeblock() → JsonParser

capture(text: str, fix: bool = True) → str | None

Captures and optionally repairs the first JSON match in text.

Parameters:

text – The text to search within.
fix – Whether to attempt JSON repair on the captured content.

Returns:

The captured text or None if no match is found.

capture_all(text: str, fix: bool = True) → list[str]

Captures and optionally repairs all JSON matches in text.

Parameters:

text – The text to search within.
fix – Whether to attempt JSON repair on each captured content.

Returns:

A list of captured JSON strings.

convert(text: str, fix: bool = True) → Any | None

Converts captured text to a Python object.

Parameters:

text – The text to parse as JSON.
fix – Whether to attempt JSON repair before parsing.

Returns:

The parsed Python object or None if conversion fails.

convert_all(text: str, fix: bool = True) → list[Any]

Converts all captured JSON strings to Python objects.

Parameters:

text – The text to search within.
fix – Whether to attempt JSON repair before parsing.

Returns:

A list of parsed Python objects.

validate_list(text: str, elements_type: Type[_T] | None = None, length: int | None = None, fix: bool = True) → List[_T] | None

Validates that the text parses to a list with optional constraints.

Parameters:

text – The text to parse as JSON.
elements_type – Optional type to check all elements against.
length – Optional exact length requirement.
fix – Whether to attempt JSON repair before parsing.

Returns:

The validated list or None if validation fails.

validate_list_str(text: str, length: int | None = None, fix: bool = True) → List[str] | None

Validates that the text parses to a Vec<String> with optional length constraint.

This is a typed convenience wrapper over deserialize that avoids Python GIL interaction since it operates on pure Rust types.

Parameters:

text – The text to parse as JSON.
length – Optional exact length requirement.
fix – Whether to attempt JSON repair before parsing.

Returns:

The validated Vec<String> or None if deserialization or length check fails.

validate_set(text: str, elements_type: Type[_T] | None = None, length: int | None = None, fix: bool = True) → Set[_T] | None

Validates that the text parses to a set with optional constraints.

Parameters:

text – The text to parse as JSON.
elements_type – Optional type to check all elements against.
length – Optional exact length requirement.
fix – Whether to attempt JSON repair before parsing.

Returns:

The validated set or None if validation fails.

validate_dict(text: str, key_type: Type[_K] | None = None, value_type: Type[_V] | None = None, length: int | None = None, fix: bool = True) → Dict[_K, _V] | None

Validates that the text parses to a dictionary with optional constraints.

Parameters:

text – The text to parse as JSON.
key_type – Optional type to check all keys against.
value_type – Optional type to check all values against.
length – Optional exact length requirement.
fix – Whether to attempt JSON repair before parsing.

Returns:

The validated dictionary or None if validation fails.

validate_dict_str_str(text: str, length: int | None = None, fix: bool = True) → Dict[str, str] | None

Validates that the text parses to a HashMap<String, String> with optional length constraint.

This is a typed convenience wrapper over deserialize that avoids Python GIL interaction since it operates on pure Rust types.

Parameters:

text – The text to parse as JSON.
length – Optional exact length requirement.
fix – Whether to attempt JSON repair before parsing.

Returns:

The validated HashMap<String, String> or None if deserialization or length check fails.

class fabricatio_core.rust.LLMConfig

Configuration for Language Learning Models (LLMs) like OpenAI’s GPT.

This structure contains all parameters needed to configure and interact with LLM services. All fields are optional to allow partial configuration from different sources.

property send_to: str | None

property no_cache: bool | None

property temperature: float | None

property top_p: float | None

property stream: bool

property max_completion_tokens: int | None

property presence_penalty: float | None

property frequency_penalty: float | None

class fabricatio_core.rust.Logger

info(msg: str) → None

debug(msg: str) → None

error(msg: str) → None

warn(msg: str) → None

trace(msg: str) → None

class fabricatio_core.rust.ProviderConfig

Configuration for a specific provider.

Contains the necessary details to connect to and authenticate with a service provider.

property ptype: ProviderType: The type of the provider (e.g., OpenAI, Anthropic).

property name: str | None: Optional name identifier for the provider instance.

property key: SecretStr | None: Optional authentication key for the provider API.

property base_url: str | None: Optional URL endpoint for the provider’s API. Must be a valid URL if provided.

class fabricatio_core.rust.RerankerConfig

Reranker configuration structure.

property send_to: str | None

property no_cache: bool | None

class fabricatio_core.rust.Router

completion(send_to: str, message: str, stream: bool = False, top_p: float | None = None, temperature: float | None = None, max_completion_tokens: int | None = None, presence_penalty: float | None = None, frequency_penalty: float | None = None, no_cache: bool = False) → Awaitable[str]

Sends a completion request to the specified group and returns the full response.

Note: Although a ‘stream’ argument exists for protocol compatibility, this implementation always aggregates the full response before returning. It does not yield chunks asynchronously.

Parameters:

send_to (str) – The router group name.
message (str) – The user prompt content.
stream (bool) – Logical flag for compatibility. No performance difference. Defaults to False.
top_p (Optional[float]) – Nucleus sampling parameter. Defaults to 1.0 if None.
temperature (Optional[float]) – Controls randomness. Defaults to 0.7 if None.
max_completion_tokens (Optional[int]) – Maximum tokens to generate. Defaults to 2048 if None.
presence_penalty (Optional[float]) – Penalizes new tokens based on presence. Defaults to 0.0 if None.
frequency_penalty (Optional[float]) – Penalizes new tokens based on frequency. Defaults to 0.0 if None.
no_cache (bool) – Whether to bypass the cache for this request. Defaults to False.

Returns:

The complete aggregated response content.

Return type:

str

completion_batch(send_to: str, messages: Sequence[str], stream: bool = False, top_p: float | None = None, temperature: float | None = None, max_completion_tokens: int | None = None, presence_penalty: float | None = None, frequency_penalty: float | None = None, no_cache: bool = False) → Awaitable[List[str | None]]

Sends a batch of completion requests to the specified group and returns all responses.

Note

Although a ‘stream’ argument exists for protocol compatibility, this implementation always aggregates the full response before returning. It does not yield chunks asynchronously.

Parameters:

send_to (RouteGroupName) – The router group name to route the completion requests.
messages (List[str]) – A list of user prompt contents.
stream (bool) – Logical flag for compatibility. No performance difference. Defaults to False.
top_p (Optional[float]) – Nucleus sampling parameter. Defaults to 1.0 if None.
temperature (Optional[float]) – Controls randomness. Defaults to 0.7 if None.
max_completion_tokens (Optional[int]) – Maximum tokens to generate. Defaults to 2048 if None.
presence_penalty (Optional[float]) – Penalizes new tokens based on presence. Defaults to 0.0 if None.
frequency_penalty (Optional[float]) – Penalizes new tokens based on frequency. Defaults to 0.0 if None.
no_cache (bool) – Whether to bypass the cache for each request. Defaults to False.

Returns:

A list of complete aggregated response contents. Failed requests return None.

Return type:

List[str | None]

embedding(send_to: str, texts: Sequence[str], ndim: int, no_cache: bool = False) → Awaitable[List[List[float]]]

Sends an embedding request to the specified group.

Parameters:

send_to (str) – The router group name to route the embedding request.
texts (List[str]) – A list of text strings to generate embeddings for.
ndim (int) – The dimensionality of the output embeddings. Must match between search and store.
no_cache (bool) – Whether to bypass the cache for this request. Defaults to False.

Returns:

A list of embedding vectors corresponding to the input texts.

Return type:

List[List[float]]

rerank(send_to: str, query: str, documents: Sequence[str], no_cache: bool = False) → Awaitable[List[Tuple[int, float]]]

Sends a reranking request to the specified group.

Parameters:

send_to (str) – The router group name to route the reranking request.
query (str) – The query text to rank documents against.
documents (List[str]) – A list of document texts to rerank.
no_cache (bool) – Whether to bypass the cache for this request. Defaults to False.

Returns:

A list of (document_index, score) pairs sorted by relevance descending.

Return type:

List[Tuple[int, float]]

add_provider(provider_type: ProviderType, name: str | None = None, api_key: SecretStr | None = None, endpoint: str | None = None) → None

Adds a provider to the router.

This method registers a new provider with the completion, embedding, and reranker routers.

Parameters:

provider_type (ProviderType) – The type of the provider (e.g., OpenAI, Anthropic).
name (Optional[str]) – Optional custom name for the provider.
api_key (Optional[SecretStr]) – Optional API key for authentication.
endpoint (Optional[str]) – Optional custom API endpoint URL.

Returns:

This is an asynchronous operation that modifies the router state.

Return type:

None

add_completion_model(group: str, model_identifier: str, rpm: int | None = None, tpm: int | None = None) → None

Adds a completion model to the specified group.

Registers a new model identifier within a specific routing group for completion tasks.

Parameters:

group (str) – The target router group name.
model_identifier (str) – The unique identifier of the model to be added.
rpm (Optional[int]) – Optional requests per minute limit.
tpm (Optional[int]) – Optional tokens per minute limit.

Returns:

This is an asynchronous operation that modifies the router state.

Return type:

None

add_embedding_model(group: str, model_identifier: str, rpm: int | None = None, tpm: int | None = None) → None

Adds an embedding model to the specified group.

Registers a new model identifier within a specific routing group for embedding tasks.

Parameters:

group (str) – The target router group name.
model_identifier (str) – The unique identifier of the model to be added.
rpm (Optional[Quota]) – Optional requests per minute limit.
tpm (Optional[Quota]) – Optional tokens per minute limit.

Returns:

This is an asynchronous operation that modifies the router state.

Return type:

None

add_reranker_model(group: str, model_identifier: str, rpm: int | None = None, tpm: int | None = None) → None

Adds a reranker model to the specified group.

Registers a new model identifier within a specific routing group for reranking tasks.

Parameters:

group (str) – The target router group name.
model_identifier (str) – The unique identifier of the model to be added.
rpm (Optional[Quota]) – Optional requests per minute limit.
tpm (Optional[Quota]) – Optional tokens per minute limit.

Returns:

This is an asynchronous operation that modifies the router state.

Return type:

None

add_or_update_dummy_completion_model(group: str, model_identifier: str, responses: Sequence[str]) → None

add_or_update_dummy_embedding_model(group: str, model_identifier: str, embeddings: Sequence[Sequence[Sequence[float]]]) → None

add_or_update_dummy_reranker_model(group: str, model_identifier: str, rankings: Sequence[Sequence[tuple[int, float]]]) → None

class fabricatio_core.rust.RouterUsage

ask(question: str, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[str]
ask(question: List[str], send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[str]]
ask(question: str | List[str], send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[str | List[str]]

mapping_strings(requirement: str, k: int | None, max_validations: int, default: Dict[str, str] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[Dict[str, str] | None]
mapping_strings(requirement: List[str], k: int | None, max_validations: int, default: Dict[str, str] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[Dict[str, str] | None]]
mapping_strings(requirement: str | List[str], k: int | None, max_validations: int, default: Dict[str, str] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[Dict[str, str] | None | List[Dict[str, str] | None]]

listing_strings(requirement: str, k: int | None, max_validations: int, default: List[str] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[str] | None]
listing_strings(requirement: List[str], k: int | None, max_validations: int, default: List[str] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[List[str] | None]]
listing_strings(requirement: str | List[str], k: int | None, max_validations: int, default: List[str] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[str] | None | List[List[str] | None]]

generic_string(requirement: str, max_validations: int, default: str | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[str | None]
generic_string(requirement: List[str], max_validations: int, default: str | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[str | None]]
generic_string(requirement: str | List[str], max_validations: int, default: str | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[str | None | List[str | None]]

code_string(requirement: str, code_language: str | None, max_validations: int, default: str | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[str | None]
code_string(requirement: List[str], code_language: str | None, max_validations: int, default: str | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[str | None]]
code_string(requirement: str | List[str], code_language: str | None, max_validations: int, default: str | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[str | None | List[str | None]]

code_snippets(requirement: str, code_language: str | None, max_validations: int, default: List[CodeSnippet] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[CodeSnippet] | None]
code_snippets(requirement: List[str], code_language: str | None, max_validations: int, default: List[CodeSnippet] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[List[CodeSnippet] | None]]
code_snippets(requirement: str | List[str], code_language: str | None, max_validations: int, default: List[CodeSnippet] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[CodeSnippet] | None | List[List[CodeSnippet] | None]]

judging(requirement: str, max_validations: int, default: bool | None, affirm_case: str, deny_case: str, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[bool | None]
judging(requirement: List[str], max_validations: int, default: bool | None, affirm_case: str, deny_case: str, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[bool | None]]
judging(requirement: str | List[str], max_validations: int, default: bool | None, affirm_case: str, deny_case: str, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[bool | None | List[bool | None]]

choosing(requirement: str, valid_names: List[str], k: int | None, max_validations: int, default: List[int] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[int] | None]
choosing(requirement: List[str], valid_names: List[str], k: int | None, max_validations: int, default: List[int] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[List[int] | None]]
choosing(requirement: str | List[str], valid_names: List[str], k: int | None, max_validations: int, default: List[int] | None, send_to: str, stream: bool, top_p: float | None, temperature: float | None, max_completion_tokens: int | None, presence_penalty: float | None, frequency_penalty: float | None, no_cache: bool) → Awaitable[List[int] | None | List[List[int] | None]]

class fabricatio_core.rust.RoutingConfig

Routing configuration structure for controlling request dispatching behavior.

Manages the list of available providers and their corresponding deployments to handle load balancing and request routing.

property providers: list[ProviderConfig]: List of configured providers available for routing.

property reranker_deployments: list[DeploymentConfig]: List of configured reranker model deployments associated with the providers.

property embedding_deployments: list[DeploymentConfig]: List of configured embedding model deployments associated with the providers.

property completion_deployments: list[DeploymentConfig]: List of configured completion model deployments associated with the providers.

property cache_database_path: pathlib.Path | None: Path to the cache database file.

class fabricatio_core.rust.SecretStr

get_secret_value() → str

class fabricatio_core.rust.TemplateConfig

Template configuration structure.

property mapping_template: str

property task_briefing_template: str: The name of the task briefing template which will be used to brief a task.

property dependencies_template: str: The name of the dependencies template which will be used to manage dependencies.

property make_choice_template: str: The name of the make choice template which will be used to make a choice.

property make_judgment_template: str: The name of the make judgment template which will be used to make a judgment.

property code_string_template: str: The name of the code string template which will be used to generate a code string.

property code_snippet_template: str: The name of the code snippet template which will be used to generate a code snippet.

property generic_string_template: str: The name of the generic string template which will be used to review a string.

property co_validation_template: str: The name of the co-validation template which will be used to co-validate a string.

property liststr_template: str: The name of the liststr template which will be used to display a list of strings.

property pathstr_template: str: The name of the pathstr template which will be used to acquire a path of strings.

property create_json_obj_template: str: The name of the create json object template which will be used to create a json object.

class fabricatio_core.rust.TemplateManager

Python bindings for the TemplateManager struct.

property templates_stores: list[pathlib.Path]

property template_count: int: The count of templates currently registered.

add_store(source: str | os.PathLike | pathlib.Path, rediscovery: bool = False) → TemplateManager

Adds a template directory to the list of template directories.

Parameters:

source – The path to the template directory.
rediscovery – Whether to immediately discover templates (default: False).

Returns:

A mutable reference to self for method chaining.

add_stores(sources: Sequence[str | os.PathLike | pathlib.Path], rediscovery: bool = False) → TemplateManager

Adds multiple template directories to the list.

Parameters:

sources – A list of paths to template directories.
rediscovery – Whether to immediately discover templates (default: False).

Returns:

A mutable reference to self for method chaining.

discover_templates() → TemplateManager

Discovers and registers all templates from the configured directories.

Returns:: A mutable reference to self for method chaining.

render_template(name: str, data: Dict[str, Any]) → str
render_template(name: str, data: List[Dict[str, Any]]) → List[str]

render_template_raw(template: str, data: Dict[str, Any]) → str
render_template_raw(template: str, data: List[Dict[str, Any]]) → List[str]

class fabricatio_core.rust.TemplateManagerConfig

property template_stores: list[pathlib.Path]: The directory containing the templates.

property active_loading: bool: Whether to enable active loading of templates.

property template_suffix: str: The suffix of the templates.

class fabricatio_core.rust.TextCapturer

cap1(text: str) → str | None

Captures the first match and extracts group 1.

Parameters:: text – The text to search within.
Returns:: The first captured group if a match is found.

cap1_all(text: str) → list[str]

Captures all matches and extracts group 1 from each.

Parameters:: text – The text to search within.
Returns:: A list of first captured groups from all matches.

cap2(text: str) → tuple[str, str] | None

Captures the first match and extracts groups 1 and 2.

Parameters:: text – The text to search within.
Returns:: A tuple of (group1, group2) if a match is found.

cap2_all(text: str) → list[tuple[str, str]]

Captures all matches and extracts groups 1 and 2 from each.

Parameters:: text – The text to search within.
Returns:: A list of (group1, group2) tuples from all matches.

cap3(text: str) → tuple[str, str, str] | None

Captures the first match and extracts groups 1, 2, and 3.

Parameters:: text – The text to search within.
Returns:: A tuple of (group1, group2, group3) if a match is found.

cap3_all(text: str) → list[tuple[str, str, str]]

Captures all matches and extracts groups 1, 2, and 3 from each.

Parameters:: text – The text to search within.
Returns:: A list of (group1, group2, group3) tuples from all matches.

static with_pattern(pattern: str) → TextCapturer

Creates a TextCapturer with a custom regex pattern.

Parameters:: pattern – The regex pattern to use.
Returns:: A new TextCapturer instance.

static capture_snippet(l_sep: str = '>>>>>', r_sep: str = '<<<<<') → TextCapturer

Creates a TextCapturer for capturing code snippets with separators.

Parameters:

l_sep – The left separator (default: “>>>>>”).
r_sep – The right separator (default: “<<<<<”).

Returns:

A new TextCapturer instance configured for snippets.

static capture_code_block(language: str = '.*?') → TextCapturer

Capture a code block of the given language.

Parameters:

language (Option<&str>) – The programming language of the code block.
None. (Capture all kinds of code block if it set to)

Returns:

An instance of TextCapturer configured to capture code blocks.

Return type:

PyResult<Self>

static capture_generic_block(language: str = 'String') → TextCapturer

Capture a generic block of the given language.

Parameters:: language (&str) – The language or identifier of the generic block.
Returns:: An instance of TextCapturer configured to capture generic blocks.
Return type:: PyResult<Self>

static capture_content(left_delimiter: str, right_delimiter: str | None = None) → TextCapturer

Capture content between delimiters.

Parameters:

left_delimiter (&str) – The left delimiter marking the start of the content.
right_delimiter (Option<&str>) – The right delimiter marking the end of the content.

Returns:

An instance of TextCapturer configured to capture content between delimiters.

Return type:

PyResult<Self>

Note

If right_delimiter is not provided, it defaults to left_delimiter.

class fabricatio_core.rust.ProviderType

Bases: enum.Enum

Enum representing supported LLM provider types.

Use this enum to specify which type of provider to create when calling [create_provider].

# Variants

OpenAI - Official OpenAI API provider. Requires an API key, falls back to OPENAI_API_KEY environment variable.
OpenAICompatible - Any OpenAI API-compatible provider (Azure OpenAI, LocalAI, custom endpoints). Requires name, API key, and endpoint URL.
Dummy - A provider that doesn’t make real HTTP calls. Useful for testing and development.

OpenAI = Ellipsis: Official OpenAI API provider.

OpenAICompatible = Ellipsis: OpenAI API-compatible provider (Azure, LocalAI, custom endpoints).

Dummy = Ellipsis: Dummy provider for testing (does not make real HTTP calls).

class fabricatio_core.rust.TaskStatus

Bases: enum.Enum

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

attribute access:
```
>>> Color.RED
<Color.RED: 1>
```
value lookup:
```
>>> Color(1)
<Color.RED: 1>
```
name lookup:
```
>>> Color['RED']
<Color.RED: 1>
```

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Pending = Ellipsis

Running = Ellipsis

Finished = Ellipsis

Failed = Ellipsis

Cancelled = Ellipsis

__setstate__(state: bytes) → None

__getstate__() → bytes

__getnewargs__() → tuple[int]

fabricatio_core.rust.blake3_hash(content: bytes) → str

Calculates a BLAKE3 hash of the given content.

Parameters:: content – The byte content to hash.
Returns:: A hexadecimal string representation of the BLAKE3 hash.

fabricatio_core.rust.detect_language(string: str) → str

Detects the language of a given string and returns its full native name.

This function uses the whichlang library to detect the primary language of the input text and converts it to a human-readable string.

Parameters:: string – The input text string to analyze.
Returns:: A string containing the detected language name in its native script.

fabricatio_core.rust.extra_satisfied(pkg_name: str, extra_name: str) → bool

Checks if a specific extra (optional dependency) of a Python package is satisfied.

Parameters:

pkg_name – The name of the package.
extra_name – The name of the extra/optional dependency.

Returns:

True if the extra is satisfied, False otherwise.

fabricatio_core.rust.extras_satisfied(pkg_name: str, extras: Sequence[str]) → bool

Checks if all specified extras (optional dependencies) of a Python package are satisfied.

Parameters:

pkg_name – The name of the package.
extras – A list of extra/optional dependency names to check.

Returns:

True if all extras are satisfied, False otherwise.

fabricatio_core.rust.is_arabic(string: str) → bool

Checks if the given string is written in Arabic.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Arabic, False otherwise.

fabricatio_core.rust.is_chinese(string: str) → bool

Checks if the given string is written in Simplified Chinese.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Mandarin Chinese, False otherwise.

fabricatio_core.rust.is_dutch(string: str) → bool

Checks if the given string is written in Dutch.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Dutch, False otherwise.

fabricatio_core.rust.is_english(string: str) → bool

Checks if the given string is written in English.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is English, False otherwise.

fabricatio_core.rust.is_french(string: str) → bool

Checks if the given string is written in French.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is French, False otherwise.

fabricatio_core.rust.is_german(string: str) → bool

Checks if the given string is written in German.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is German, False otherwise.

fabricatio_core.rust.is_hindi(string: str) → bool

Checks if the given string is written in Hindi.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Hindi, False otherwise.

fabricatio_core.rust.is_installed(pkg_name: str) → bool

Checks if a Python package is installed.

Parameters:: pkg_name – The name of the package to check.
Returns:: True if the package is installed, False otherwise.

fabricatio_core.rust.is_italian(string: str) → bool

Checks if the given string is written in Italian.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Italian, False otherwise.

fabricatio_core.rust.is_japanese(string: str) → bool

Checks if the given string is written in Japanese.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Japanese, False otherwise.

fabricatio_core.rust.is_korean(string: str) → bool

Checks if the given string is written in Korean.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Korean, False otherwise.

fabricatio_core.rust.is_likely_text(path: str | os.PathLike | pathlib.Path) → bool

Determines if a file is likely text content.

This function uses heuristics to determine if a file contains text. Directories and non-existent paths return False.

Parameters:: path – The path to the file to check.
Returns:: True if the file appears to be text, False otherwise.

fabricatio_core.rust.is_portuguese(string: str) → bool

Checks if the given string is written in Portuguese.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Portuguese, False otherwise.

fabricatio_core.rust.is_russian(string: str) → bool

Checks if the given string is written in Russian.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Russian, False otherwise.

fabricatio_core.rust.is_swedish(string: str) → bool

Checks if the given string is written in Swedish.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Swedish, False otherwise.

fabricatio_core.rust.is_turkish(string: str) → bool

Checks if the given string is written in Turkish.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Turkish, False otherwise.

fabricatio_core.rust.is_vietnamese(string: str) → bool

Checks if the given string is written in Vietnamese.

Parameters:: string – The input text string to check.
Returns:: True if the detected language is Vietnamese, False otherwise.

fabricatio_core.rust.list_installed() → list[str]

Lists all installed Python packages.

Returns:: A list of names of all installed packages.

fabricatio_core.rust.split_into_chunks(string: str, max_chunk_size: int, max_overlapping_rate: float = 0.3) → list[str]

Splits a string into chunks based on maximum size and overlapping rate.

The function prioritizes splitting at sentence boundaries. If a sentence exceeds the maximum chunk size, it will be split at word boundaries.

Parameters:

string – The input string to be split.
max_chunk_size – The maximum number of words allowed in a chunk.
max_overlapping_rate – The rate of overlapping between consecutive chunks (default: 0.3).

Returns:

A list of chunk strings.

fabricatio_core.rust.split_sentence_bounds(string: str) → list[str]

Splits a string into sentences using Unicode sentence boundaries.

This function uses Unicode segmentation to properly identify sentence boundaries across different writing systems.

Parameters:: string – The input string to split.
Returns:: A list of sentence strings.

fabricatio_core.rust.split_word_bounds(string: str) → list[str]

Splits a string into words using Unicode word boundaries.

This function uses Unicode segmentation to properly handle words in various languages, not just whitespace-separated tokens.

Parameters:: string – The input string to split.
Returns:: A list of word strings.

fabricatio_core.rust.tokens_of(text: str) → int

Counts the number of tokens in a text string.

This function uses the thryd library’s token counting mechanism.

Parameters:: text – The input text to count tokens in.
Returns:: The number of tokens in the text.

fabricatio_core.rust.word_count(string: str) → int

Counts the number of words in a string.

This function splits the string by word boundaries and counts non-empty tokens.

Parameters:: string – The input string.
Returns:: The number of words in the string.