Tools and Utilities

Overview

LangGraph tools for human interaction, code execution, and knowledge search.

Python Code Execution

Tool for running python code.

class openchatbi.tool.run_python_code.PythonCodeInput(*, reasoning: str, code: str)[source]

Bases: BaseModel

reasoning: str

code: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Human Interaction

Tool for asking human clarification when information is ambiguous.

class openchatbi.tool.ask_human.AskHuman(*, question: str, options: list[str])[source]

Bases: BaseModel

Ask user for clarification when data is missing or ambiguous.

Use this tool ONLY when you are STRONGLY certain that information is ambiguous or missing. First try to solve the question with available user input before calling this tool.

question: str

options: list[str]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Memory Management

class openchatbi.tool.memory.UserProfile(*, name: str | None = None, language: str | None = None, timezone: str | None = None, jargon: str | None = None)[source]

Bases: BaseModel

Represents the full representation of a user.

name: str | None

language: str | None

timezone: str | None

jargon: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

openchatbi.tool.memory.get_sync_memory_store() → SqliteStore | None[source]

async openchatbi.tool.memory.get_async_memory_store() → AsyncSqliteStore | None[source]: Get or create the async memory store.

async openchatbi.tool.memory.cleanup_async_memory_store() → None[source]: Cleanup async memory store resources.

async openchatbi.tool.memory.setup_async_memory_store() → Any[source]: Setup async memory store for langmem.

openchatbi.tool.memory.fix_schema_for_openai(schema: dict) → None[source]

openchatbi.tool.memory.get_memory_manager() → Any[source]

class openchatbi.tool.memory.StructuredToolWithRequired(orig_tool: StructuredTool)[source]

Bases: StructuredTool

__init__(orig_tool: StructuredTool)[source]

Initialize the tool.

Raises:: TypeError – If args_schema is not a subclass of pydantic BaseModel or dict.

property tool_call_schema: Any

Get the schema for tool calls, excluding injected arguments.

Returns:: The schema that should be used for tool calls from language models.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'ignore', 'protected_namespaces': ()}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

openchatbi.tool.memory.get_memory_tools(llm: BaseChatModel, sync_mode: bool = False, store: Any | None = None) → list[StructuredTool] | None[source]

async openchatbi.tool.memory.get_async_memory_tools(llm: BaseChatModel) → list[StructuredTool][source]: Get memory tools configured with async store.

Knowledge Search

Tools for searching knowledge bases and schema information.

class openchatbi.tool.search_knowledge.SearchInput(*, reasoning: str, query_list: list[str], knowledge_bases: list[str], with_table_list: bool)[source]

Bases: BaseModel

Input schema for knowledge search tool.

reasoning: str

query_list: list[str]

knowledge_bases: list[str]

with_table_list: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class openchatbi.tool.search_knowledge.ShowSchemaInput(*, reasoning: str, tables: list[str])[source]

Bases: BaseModel

Input schema for show schema tool.

reasoning: str

tables: list[str]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Anomaly Detection

Time series anomaly detection tool. Compares recent points against a forecasting baseline and produces a multi-factor anomaly score. Shares the time series forecasting service health check with timeseries_forecast — when the forecasting service is unavailable, both tools are excluded.

Tool for time series anomaly detection (thin LangChain wrapper).

class openchatbi.tool.anomaly_detection.AnomalyDetectionInput(*, reasoning: str, input_data: list[float | int | dict[str, Any]], detection_range: Annotated[int | None, Ge(ge=2)] = None, frequency: str = 'hourly', stride: Annotated[int | None, Ge(ge=1), Le(le=200)] = None, target_column: str = 'value', input_length: int | None = None, drop_weight: Annotated[float, Ge(ge=0.0), Le(le=3.0)] = 1.0, rise_weight: Annotated[float, Ge(ge=0.0), Le(le=3.0)] = 1.0)[source]

Bases: BaseModel

Input schema for anomaly detection tool.

reasoning: str

input_data: list[float | int | dict[str, Any]]

detection_range: int | None

frequency: str

stride: int | None

target_column: str

input_length: int | None

drop_weight: float

rise_weight: float

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Anomaly Drill-Down (Adtributor)

Multi-dimensional root cause analysis based on the Microsoft Adtributor algorithm. Accepts a 1D melted table (list of rows with dimension_name / element_value and baseline/actual values) and returns the dimensions and elements that best explain an anomaly, with a business-friendly narrative.

Bases: BaseModel

A single row of data for the adtributor drilldown analysis.

dimension_name: str

element_value: str | int | float

predict: float | None

real: float | None

predict_numerator: float | None

predict_denominator: float | None

real_numerator: float | None

real_denominator: float | None

proportion: float | None

base_proportion: float | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class openchatbi.tool.adtributor_tool.AdtributorToolInput(*, reasoning: str, data: list[DrilldownRow], derived: bool, issue_type: str = 'drop', tep: float = 0.7, teep: float = 0.02, k: int = 1)[source]

Bases: BaseModel

Input schema for the adtributor drilldown tool.

reasoning: str

data: list[DrilldownRow]

derived: bool

issue_type: str

tep: float

teep: float

k: int

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Data Analysis Agent

The data analysis agent is a specialized sub-agent that orchestrates complex analysis tasks (like anomaly detection, root cause drill-down, and trend forecasting).

See the dedicated Data Analysis Agent section for full details on the agent workflows, tool dependencies, and the underlying algorithms.