core package

Submodules

core.basics module

class core.basics.DesignGoal(text: str | None = None, images: List[str] | None = None, extra: Dict[str, Any] | None = None)

Bases: object

Represents a design goal, which can be defined by either text, images, or both.

A design goal encapsulates the user’s input, which can be in the form of a textual description, one or more images, or a combination of both. Images can be provided as paths to individual image files or as a path to a folder containing multiple images.

Parameters:

• text (Optional[str]) – A textual description of the design goal. Defaults to None.

• images (Optional[list[str]]) – A list of image file paths or a single folder path containing images. Defaults to None.

• extra (Optional[Dict[str, Any]]) – Additional information related to the design goal, such as original user input or decomposed part list, etc. Defaults to an empty dictionary.

Raises:

ValueError – If neither text nor images is provided.

text

The textual description of the design goal.

Type:

Optional[str]

images

A list of image file paths or a single folder path.

Type:

Optional[list[str]]

extra

Additional information related to the design goal, such as original user input or decomposed part list, etc.

Type:

Dict[str, Any]

classmethod from_dict(data: Dict[str, Any]) → DesignGoal

Create a DesignGoal object from a dictionary.

Parameters:

data (Dict[str, Any]) – A dictionary containing the design goal data.

Returns:

A DesignGoal object.

Return type:

DesignGoal

classmethod from_json(json_str: str) → DesignGoal

Create a DesignGoal object from a JSON string.

Parameters:

json_str (str) – A JSON string containing the design goal data.

Returns:

A DesignGoal object.

Return type:

DesignGoal

to_dict() → Dict[str, Any]

Convert the DesignGoal object to a dictionary.

Returns:

A dictionary representation of the DesignGoal object.

Return type:

Dict[str, Any]

to_json() → str

Convert the DesignGoal object to a JSON string.

Returns:

A JSON string representation of the DesignGoal object.

Return type:

str

class core.basics.PromptBuilder

Bases: object

A utility class for constructing prompts with text and images.

This class is designed to build a list of messages that can be used as input for models that accept multi-modal prompts (e.g., text and images). Messages can include system prompts, user prompts with text, and user prompts with images.

messages

A list of messages, where each message is a dictionary containing a role (“system” or “user”) and content (text or image data).

Type:

list

add_images(image_data: list[str] | str, msg_index: int | None = None) → int

Add images to the messages. If msg_index is provided, the images will be appended to the existing message at that index to form a multi-content message.

Accepts a list of image paths or a single image path. Each image is converted to a base64-encoded string and added as a user message with image content.

Parameters:

• image_data (list[str] | str) – A list of image paths or a single image path.

• msg_index (Optional[int]) – The index of the message in the messages list. Useful when appending to an existing message.

Returns:

The index of the message in the messages list, or -1 if no valid images were found.

Return type:

msg_index (int)

add_system_message(content)

Add a system prompt to the messages.

Parameters:

content (str) – The content of the system prompt.

add_text(content: str, msg_index: int | None = None) → int

Add a user message with text content to the messages. If msg_index is provided, the text will be appended to the existing message at that index to form a multi-content message.

Parameters:

• content (str) – The text content of the user message.

• msg_index (Optional[int]) – The index of the message in the messages list. Useful when appending to an existing message.

Returns:

The index of the message in the messages list.

Return type:

msg_index (int)

add_user_message(content)

Add a user prompt with text content to the messages.

Parameters:

content (str) – The text content of the user prompt.

core.embeddings module

class core.embeddings.Embeddings(api_key: str, api_base_url: str, model_name: str, org_id: str | None = None, **model_kwargs)

Bases: ABC

embed(texts: List[SupportStr]) → List[List[float]]

Generate embeddings for texts using unified HTTP request helper.

Parameters:

texts (List[SupportStr]) – A list of SupportStr objects to generate embeddings for.

Returns:

A list of embeddings, where each embedding is a list of floats.

Return type:

List[List[float]]

embed_documents(texts: List[SupportStr]) → List[List[float]]

Generate embeddings for multiple documents using HTTP API.

Parameters:

texts (List[SupportStr]) – A list of SupportStr objects to generate embeddings for.

Returns:

A list of embeddings, where each embedding is a list of floats.

Return type:

List[List[float]]

embed_query(text: SupportStr) → List[float]

Generate embedding for a single query text using HTTP API.

Parameters:

text (SupportStr) – Query text to generate embedding for.

Returns:

Embedding vector for the query text.

Return type:

List[float]

core.model module

class core.model.MultiModalModel(api_key: str, api_base_url: str, model_name: str, org_id: str | None = None, **model_kwargs: Dict[str, Any])

Bases: ABC

Language model base class with tool use support.

api_key

API key for authentication

Type:

str

api_base_url

Base URL for API endpoint

Type:

str

model_name

Name of the language model

Type:

str

org_id

Organization ID

Type:

Optional[str]

model_kwargs

Additional model parameters

Type:

Dict[str, Any]

stream

Whether to use streaming mode

Type:

bool

client

OpenAI client instance

Type:

openai.OpenAI

class StreamState(content: str = '', reasoning_content: str = '', stream_tool_calls: Dict[int, Dict[str, Any]] = <factory>, tool_responses: List[Dict[str, Any]] = <factory>)

Bases: object

content: str = ''

reasoning_content: str = ''

stream_tool_calls: Dict[int, Dict[str, Any]]

tool_responses: List[Dict[str, Any]]

get_response_with_tools(messages, tool_calls, tools, result, stream=False)

query(prompt: str | None = None, system_prompt: str | None = None, stream: bool = False, prompt_builder: PromptBuilder | None = None, messages: List[ChatCompletionMessage] | None = None, tools: ToolRegistry | None = None) → Dict[str, Any]

Query the language model with support for tool use.

Parameters:

• prompt (Optional[str]) – User input prompt

• system_prompt (Optional[str]) – System prompt

• stream (bool) – Whether to use streaming mode

• prompt_builder (Optional[PromptBuilder]) – PromptBuilder instance

• messages (Optional[List[ChatCompletionMessage]]) – List of messages

• tools (Optional[ToolRegistry]) – Tool registry instance

Returns:

Dictionary containing:

• content: Main response content

• reasoning_content: Reasoning steps (if available)

• tool_responses: Tool execution results (if tools used)

• formatted_response: Formatted final response

• tool_chain: Chain of tool calls and responses (if tools used)

Return type:

Dict[str, Any]

Raises:

ValueError – If no response is received from the model

core.rerank module

class core.rerank.Reranker(api_key: str, api_base_url: str, model_name: str, **model_kwargs)

Bases: ABC

rerank(query: SupportStr, documents: List[SupportStr], top_n: int = 4, return_documents: bool = False) → List[Dict]

Rerank a list of documents based on a query.

Parameters:

• query (str) – The query to rerank documents against.

• documents (List[str]) – List of documents to rerank.

• top_n (int, optional) – Number of top documents to return. Defaults to 4.

• return_documents (bool, optional) – Whether to return the full documents or just scores. Defaults to False.

Returns:

List of reranked documents or scores.

Return type:

List[Dict]

core.types module

class core.types.SupportStr(*args, **kwargs)

Bases: Protocol

A protocol for objects that support conversion to a string.

This protocol ensures that any implementing class has a __str__ method, allowing the object to be represented as a string.

Example

This can be implemented in a class as follows:

``` class MyClass:

def __str__(self) -> str:

return “MyClass string representation”

```

Returns:

The string representation of the object when __str__ is called.

Return type:

str

core.utils module

class core.utils.ANSIStrippingFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

A logging formatter that removes ANSI escape codes from log messages.

This formatter is particularly useful when logging to files, where ANSI escape codes (used for terminal coloring) are not needed and can clutter the log output.

ansi_escape = re.compile('\\x1B(?:[@-Z\\\\-_]|\\[[0-?]*[ -/]*[@-~])')

format(record)

Format the log record, removing any ANSI escape codes from the message.

Parameters:

record (logging.LogRecord) – The log record to format.

Returns:

The formatted log message with ANSI escape codes removed.

Return type:

str

core.utils.colorstring(message: Any, color: str | None = 'green', bold: bool = False) → str

Returns a colored string using either ANSI escape codes or blessed terminal capabilities.

Parameters:

• message – The message to be colored. Can be of any type (e.g., str, int, float, bool).

• color – The color to apply. Supported colors: ‘black’, ‘red’, ‘green’, ‘yellow’, ‘blue’, ‘magenta’, ‘cyan’, ‘white’, and their bright variants: ‘bright_black’, ‘bright_red’, ‘bright_green’, ‘bright_yellow’, ‘bright_blue’, ‘bright_magenta’, ‘bright_cyan’, ‘bright_white’.

• bold – If True, applies bold styling to the text (only applicable when use_ansi=True).

Returns:

A string with the specified color and styling.

core.utils.cprint(message: Any, color: str | None = 'green', **kwargs) → None

Prints a colored string using blessed terminal capabilities.

Parameters:

• message – The message to be colored. Can be of any type (e.g., str, int, float, bool).

• color – The color to apply. Supported colors: ‘black’, ‘red’, ‘green’, ‘yellow’, ‘blue’, ‘magenta’, ‘cyan’, ‘white’.

• kwargs – Additional keyword arguments to pass to the print function (e.g., end, sep, file, flush).

core.utils.extract_section_markdown(text: str, heading: str) → str

Extracts content from a markdown text under the specified heading.

core.utils.find_files_with_extensions(directory_path: str, extensions: str | Iterable[str], return_all: bool = False) → str | List[str] | None

Find files with the specified extensions in the given directory. If return_all is False (default), returns the first matching file based on priority. If return_all is True, returns a list of all matching files, sorted by priority.

Parameters:

• directory_path (str) – Path to the directory to search.

• extensions (Union[str, List[str]]) – A single extension or a list of extensions.

• return_all (bool) – If True, return all matching files; otherwise, return the first match.

Returns:

A single file path, a list of file paths, or None if no files are found.

Return type:

Union[str, List[str], None]

core.utils.get_image_paths(path: str | List[str]) → List[str]

Get image file paths from a specified folder, a single image file, or a list of image paths.

Parameters: path (Union[str, List[str]]): The path to the folder, the single image file, or a list of image paths.

Returns: List[str]: A list of image file paths.

Raises: ValueError: If any path does not exist or is not a valid image file or folder of images.

core.utils.image_to_base64(image: Image | str, quality: int = 85, max_resolution: tuple = (448, 448), img_format: str = 'WEBP') → str

Convert the image to a base64 encoded string.

Parameters:

• image – PIL Image object or the path to the image file.

• quality – Compression quality (0-100) for WebP format. Higher values mean better quality but larger size.

• max_resolution – Optional maximum resolution (width, height) to fit the image within while preserving aspect ratio.

• img_format – Image format to use for encoding. Default is “WEBP”.

Returns:

Base64 encoded string of the image.

core.utils.is_base64_encoded(s: str) → bool

Check if a string is Base64 encoded. :param s: The string to check. :return: True if the string is Base64 encoded, False otherwise.

core.utils.load_config(config_path: str, config_name: str | None = None) → dict

Load a YAML configuration file or a specific configuration from a folder or file.

Parameters:

• config_path (str) – Path to the YAML configuration file or folder containing configuration files.

• config_name (Optional[str]) – Name of the target configuration file (if config_path is a folder) or the key within the YAML file (if config_path is a file). If omitted and config_path is a file, the entire file is loaded.

Returns:

Dictionary containing the configuration data.

Return type:

dict

Raises:

• FileNotFoundError – If the specified config_path does not exist.

• yaml.YAMLError – If the YAML file is malformed or cannot be parsed.

• ValueError – If the config_name is not found in the configuration.

core.utils.load_prompts(prompts_path: str, which_model: str) → dict

Load prompts from a YAML file and return prompts for a specific model.

Parameters:

• prompts_path (str) – Path to the YAML file containing prompts.

• which_model (str) – Key specifying which model’s prompts to load.

Returns:

Dictionary containing prompts for the specified model.

Return type:

dict

Raises:

KeyError – If the specified which_model key is not found in the YAML file.

core.utils.make_http_request(base_url: str, endpoint: str, api_key: str, payload: Dict, method: str = 'POST', timeout: float = 30.0, max_retries: int = 3, retry_delay: float = 1.0, retry_status_codes: List[int] = [429, 500, 502, 503, 504]) → Dict

Enhanced HTTP request helper with retry mechanism.

Parameters:

• base_url – Base URL for the API

• endpoint – API endpoint path

• api_key – API key for authentication

• payload – Request payload

• method – HTTP method (default: POST)

• timeout – Request timeout in seconds

• max_retries – Maximum number of retry attempts

• retry_delay – Initial delay between retries in seconds (exponential backoff)

• retry_status_codes – HTTP status codes that should trigger a retry

Returns:

Parsed JSON response

Return type:

Dict

Raises:

RuntimeError – After all retries failed or for non-retryable errors

core.utils.parse_design_goal(design_goal_input: str) → str

Parse the design goal input, which can be either a JSON file or plain text. If it’s a JSON file, extract the ‘text’ field.

Parameters:

design_goal_input (str) – Path to a JSON file or plain text.

Returns:

The design goal text.

Return type:

str

core.utils.parse_json_response(response: str) → Dict[str, Any]

Parse JSON response from VLM, handling potential errors

core.utils.recover_stream_tool_calls(stream_tool_calls: Dict) → List[ChatCompletionMessageToolCall]

core.utils.setup_logging(log_level: str = 'INFO', log_file: str | None = None, log_format: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s') → None

Configure the logging system.

This function sets up logging with the following behavior: - Console logs retain ANSI escape codes for colored output. - File logs (if specified) have ANSI escape codes removed for cleaner output. - Mutes httpx INFO logs while keeping the global log_level at INFO.

Parameters:

• log_level (str) – The logging level (e.g., “INFO”, “DEBUG”). Defaults to “INFO”.

• log_file (Optional[str]) – Path to the log file. If None, no file logging is performed.

• log_format (str) – The format string for log messages. Defaults to “%(asctime)s - %(name)s - %(levelname)s - %(message)s”.

Returns:

None

Module contents