dataknobs-bots Complete API Reference¶
Complete auto-generated API documentation from source code docstrings.
💡 Also see: - Curated API Guide - Hand-crafted tutorials and examples - Package Overview - Introduction and getting started - Source Code - View on GitHub
dataknobs_bots ¶
DataKnobs Bots - Configuration-driven AI agents.
Modules:
| Name | Description |
|---|---|
api |
FastAPI integration components for dataknobs_bots. |
artifacts |
Artifact management for conversational workflows. |
bot |
Bot core components. |
config |
Configuration utilities for DynaBot. |
context |
Context management for conversational workflows. |
generators |
Deterministic content generators for structured output production. |
knowledge |
Knowledge base implementations for DynaBot. |
memory |
Memory implementations for DynaBot. |
middleware |
Middleware components for bot request/response lifecycle. |
providers |
Provider creation utilities for dataknobs-bots. |
reasoning |
Reasoning strategies for DynaBot. |
registry |
Registry module for bot registration storage and management. |
review |
Review system for validating artifacts. |
rubrics |
Rubric-based evaluation system for structured content assessment. |
testing |
Testing utilities for dataknobs-bots. |
tools |
Tools for DynaBot. |
utils |
Utility functions and helpers for the dataknobs_bots package. |
Classes:
| Name | Description |
|---|---|
BotContext |
Runtime context for bot execution. |
BotManager |
Manages multiple DynaBot instances for multi-tenancy. |
BotRegistry |
Multi-tenant bot registry with caching and environment support. |
DynaBot |
Configuration-driven chatbot leveraging the DataKnobs ecosystem. |
UndoResult |
Result of an undo operation. |
ConfigDraftManager |
File-based draft manager for interactive config creation. |
ConfigTemplate |
A reusable DynaBot configuration template. |
ConfigTemplateRegistry |
Registry for managing and applying configuration templates. |
ConfigValidator |
Pluggable validation engine for DynaBot configurations. |
DraftMetadata |
Metadata for a configuration draft. |
DynaBotConfigBuilder |
Fluent builder for DynaBot configurations. |
DynaBotConfigSchema |
Queryable registry of valid DynaBot configuration options. |
TemplateVariable |
Definition of a template variable. |
ToolCatalog |
Registry mapping tool names to class paths and default configuration. |
ToolEntry |
Metadata for a tool in the catalog. |
ValidationResult |
Result of validating a configuration. |
RAGKnowledgeBase |
RAG knowledge base using dataknobs-xization for chunking and vector search. |
BufferMemory |
Simple buffer memory keeping last N messages. |
CompositeMemory |
Combines multiple memory strategies into one. |
Memory |
Abstract base class for memory implementations. |
SummaryMemory |
Memory that summarizes older messages to maintain long context windows. |
VectorMemory |
Vector-based semantic memory using dataknobs-data vector stores. |
CostTrackingMiddleware |
Middleware for tracking LLM API costs and usage. |
LoggingMiddleware |
Middleware for tracking conversation interactions. |
Middleware |
Base class for bot middleware. |
ReActReasoning |
ReAct (Reasoning + Acting) strategy. |
ReasoningStrategy |
Abstract base class for reasoning strategies. |
SimpleReasoning |
Simple reasoning strategy that makes direct LLM calls. |
StrategyCapabilities |
Declares what a reasoning strategy manages autonomously. |
StrategyRegistry |
Registry mapping strategy names to their factories. |
BotTestHarness |
High-level test helper for ALL DynaBot behavioral tests. |
CaptureReplay |
Loads a capture JSON file and creates pre-loaded EchoProviders. |
TurnResult |
Result of a single |
WizardConfigBuilder |
Fluent builder for wizard configuration dicts. |
AddKBResourceTool |
Tool for adding a resource to the knowledge base resource list. |
CheckKnowledgeSourceTool |
Tool for verifying a knowledge source directory exists and has content. |
GetTemplateDetailsTool |
Tool for getting detailed information about a template. |
IngestKnowledgeBaseTool |
Tool for writing the KB ingestion manifest and finalizing KB config. |
KnowledgeSearchTool |
Tool for searching the knowledge base. |
ListAvailableToolsTool |
Tool for listing tools available to configure for a bot. |
ListKBResourcesTool |
Tool for listing currently tracked knowledge base resources. |
ListTemplatesTool |
Tool for listing available configuration templates. |
PreviewConfigTool |
Tool for previewing the configuration being built. |
RemoveKBResourceTool |
Tool for removing a resource from the knowledge base resource list. |
SaveConfigTool |
Tool for saving/finalizing the configuration. |
Functions:
| Name | Description |
|---|---|
normalize_wizard_state |
Normalize wizard metadata to canonical structure. |
create_default_catalog |
Create a new ToolCatalog pre-populated with built-in tools. |
create_knowledge_base_from_config |
Create knowledge base from configuration. |
create_memory_from_config |
Create memory instance from configuration. |
create_reasoning_from_config |
Create reasoning strategy from configuration. |
register_strategy |
Register a custom reasoning strategy. |
inject_providers |
Inject LLM providers into a DynaBot instance for testing. |
Attributes:
| Name | Type | Description |
|---|---|---|
default_catalog |
ToolCatalog
|
Module-level singleton catalog pre-populated with built-in tools. |
Attributes¶
default_catalog
module-attribute
¶
Module-level singleton catalog pre-populated with built-in tools.
Classes¶
BotContext
dataclass
¶
BotContext(
conversation_id: str,
client_id: str,
user_id: str | None = None,
session_metadata: dict[str, Any] = dict(),
request_metadata: dict[str, Any] = dict(),
)
Runtime context for bot execution.
Supports dict-like access for dynamic attributes via request_metadata.
Use context["key"] or context.get("key") for dynamic data.
Attributes:
| Name | Type | Description |
|---|---|---|
conversation_id |
str
|
Unique identifier for the conversation |
client_id |
str
|
Identifier for the client/tenant |
user_id |
str | None
|
Optional user identifier |
session_metadata |
dict[str, Any]
|
Metadata for the session |
request_metadata |
dict[str, Any]
|
Metadata for the current request (also used for dict-like access) |
Methods:
| Name | Description |
|---|---|
__getitem__ |
Get item from request_metadata using dict-like access. |
__setitem__ |
Set item in request_metadata using dict-like access. |
__contains__ |
Check if key exists in request_metadata. |
get |
Get item from request_metadata with optional default. |
copy |
Create a copy of this context with optional field overrides. |
Functions¶
__getitem__ ¶
Get item from request_metadata using dict-like access.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Key to retrieve |
required |
Returns:
| Type | Description |
|---|---|
Any
|
Value from request_metadata |
Raises:
| Type | Description |
|---|---|
KeyError
|
If key not found in request_metadata |
Source code in packages/bots/src/dataknobs_bots/bot/context.py
__setitem__ ¶
Set item in request_metadata using dict-like access.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Key to set |
required |
value
|
Any
|
Value to store |
required |
__contains__ ¶
Check if key exists in request_metadata.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Key to check |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if key exists in request_metadata |
get ¶
Get item from request_metadata with optional default.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Key to retrieve |
required |
default
|
Any
|
Default value if key not found |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
Value from request_metadata or default |
Source code in packages/bots/src/dataknobs_bots/bot/context.py
copy ¶
Create a copy of this context with optional field overrides.
Creates shallow copies of session_metadata and request_metadata dicts to avoid mutation issues between the original and copy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**overrides
|
Any
|
Field values to override in the copy |
{}
|
Returns:
| Type | Description |
|---|---|
BotContext
|
New BotContext instance with copied values |
Example
ctx = BotContext(conversation_id="conv-1", client_id="client-1") ctx2 = ctx.copy(conversation_id="conv-2") ctx2.conversation_id 'conv-2'
Source code in packages/bots/src/dataknobs_bots/bot/context.py
BotManager ¶
BotManager(
config_loader: ConfigLoaderType | None = None,
environment: EnvironmentConfig | str | None = None,
env_dir: str | Path = "config/environments",
)
Manages multiple DynaBot instances for multi-tenancy.
.. deprecated::
Use :class:BotRegistry or :class:InMemoryBotRegistry instead.
BotManager handles: - Bot instance creation and caching - Client-level isolation - Configuration loading and validation - Bot lifecycle management - Environment-aware resource resolution (optional)
Each client/tenant gets its own bot instance, which can serve multiple users. The underlying DynaBot architecture ensures conversation isolation through BotContext with different conversation_ids.
Attributes:
| Name | Type | Description |
|---|---|---|
bots |
Cache of bot_id -> DynaBot instances |
|
config_loader |
Optional configuration loader (sync or async) |
|
environment_name |
str | None
|
Current environment name (if environment-aware) |
Example
# Basic usage with inline configuration
manager = BotManager()
bot = await manager.get_or_create("my-bot", config={
"llm": {"provider": "openai", "model": "gpt-4o"},
"conversation_storage": {"backend": "memory"},
})
# With environment-aware configuration
manager = BotManager(environment="production")
bot = await manager.get_or_create("my-bot", config={
"bot": {
"llm": {"$resource": "default", "type": "llm_providers"},
"conversation_storage": {"$resource": "db", "type": "databases"},
}
})
# With config loader function
def load_config(bot_id: str) -> dict:
return load_yaml(f"configs/{bot_id}.yaml")
manager = BotManager(config_loader=load_config)
bot = await manager.get_or_create("my-bot")
# List active bots
active_bots = manager.list_bots()
Initialize BotManager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config_loader
|
ConfigLoaderType | None
|
Optional configuration loader.
Can be:
- An object with a |
None
|
environment
|
EnvironmentConfig | str | None
|
Environment name or EnvironmentConfig for resource resolution. If None, environment-aware features are disabled unless an EnvironmentAwareConfig is passed to get_or_create(). If a string, loads environment config from env_dir. |
None
|
env_dir
|
str | Path
|
Directory containing environment config files. Only used if environment is a string name. |
'config/environments'
|
Methods:
| Name | Description |
|---|---|
get_or_create |
Get existing bot or create new one. |
get |
Get bot without creating if doesn't exist. |
remove |
Remove bot instance. |
reload |
Reload bot instance with fresh configuration. |
list_bots |
List all active bot IDs. |
get_bot_count |
Get count of active bots. |
clear_all |
Clear all bot instances. |
get_portable_config |
Get portable configuration for storage. |
__repr__ |
String representation. |
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
Attributes¶
environment_name
property
¶
Get current environment name, or None if not environment-aware.
environment
property
¶
Get current environment config, or None if not environment-aware.
Functions¶
get_or_create
async
¶
get_or_create(
bot_id: str,
config: dict[str, Any] | EnvironmentAwareConfig | None = None,
use_environment: bool | None = None,
config_key: str = "bot",
) -> DynaBot
Get existing bot or create new one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier (e.g., "customer-support", "sales-assistant") |
required |
config
|
dict[str, Any] | EnvironmentAwareConfig | None
|
Optional bot configuration. Can be: - dict with resolved values (traditional) - dict with $resource references (requires environment) - EnvironmentAwareConfig instance If not provided and config_loader is set, will load configuration. |
None
|
use_environment
|
bool | None
|
Whether to use environment-aware resolution. - True: Use environment for $resource resolution - False: Use config as-is (no resolution) - None (default): Auto-detect based on whether manager has an environment configured or config is EnvironmentAwareConfig |
None
|
config_key
|
str
|
Key within config containing bot configuration. Defaults to "bot". Set to None to use root config. Only used when use_environment is True. |
'bot'
|
Returns:
| Type | Description |
|---|---|
DynaBot
|
DynaBot instance |
Raises:
| Type | Description |
|---|---|
ValueError
|
If config is None and no config_loader is set |
Example
# Traditional usage (no environment resolution)
manager = BotManager()
bot = await manager.get_or_create("support-bot", config={
"llm": {"provider": "openai", "model": "gpt-4"},
"conversation_storage": {"backend": "memory"},
})
# Environment-aware usage with $resource references
manager = BotManager(environment="production")
bot = await manager.get_or_create("support-bot", config={
"bot": {
"llm": {"$resource": "default", "type": "llm_providers"},
"conversation_storage": {"$resource": "db", "type": "databases"},
}
})
# Explicit environment resolution control
bot = await manager.get_or_create(
"support-bot",
config=my_config,
use_environment=True,
config_key="bot"
)
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 | |
get
async
¶
Get bot without creating if doesn't exist.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
DynaBot | None
|
DynaBot instance if exists, None otherwise |
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
remove
async
¶
Remove bot instance.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if bot was removed, False if didn't exist |
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
reload
async
¶
Reload bot instance with fresh configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
DynaBot
|
New DynaBot instance |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no config_loader is set |
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
list_bots ¶
List all active bot IDs.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of bot identifiers |
get_bot_count ¶
Get count of active bots.
Returns:
| Type | Description |
|---|---|
int
|
Number of active bot instances |
clear_all
async
¶
Clear all bot instances.
Useful for testing or when restarting the service.
get_portable_config ¶
Get portable configuration for storage.
Extracts portable config (with $resource references intact, environment variables unresolved) suitable for storing in registries or databases.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any] | EnvironmentAwareConfig
|
Configuration to make portable. Can be dict or EnvironmentAwareConfig. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Portable configuration dictionary |
Example
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
__repr__ ¶
String representation.
Source code in packages/bots/src/dataknobs_bots/bot/manager.py
BotRegistry ¶
BotRegistry(
backend: RegistryBackend | None = None,
environment: EnvironmentConfig | str | None = None,
env_dir: str | Path = "config/environments",
cache_ttl: int = 300,
max_cache_size: int = 1000,
validate_on_register: bool = True,
config_key: str = "bot",
)
Multi-tenant bot registry with caching and environment support.
The BotRegistry manages multiple bot instances for different clients/tenants. It provides: - Pluggable storage backends via RegistryBackend protocol - Environment-aware configuration resolution - Portability validation to ensure configs work across environments - LRU-style caching with TTL for bot instances - Thread-safe access
This enables: - Multi-tenant SaaS platforms - A/B testing with different bot configurations - Horizontal scaling with stateless bot instances - Cross-environment deployment with portable configs
Attributes:
| Name | Type | Description |
|---|---|---|
backend |
RegistryBackend
|
Storage backend for configurations |
environment |
EnvironmentConfig | None
|
Environment for $resource resolution |
cache_ttl |
int
|
Time-to-live for cached bots in seconds |
max_cache_size |
int
|
Maximum number of bots to cache |
Example
from dataknobs_bots.bot import BotRegistry
from dataknobs_bots.registry import InMemoryBackend
# Create registry
registry = BotRegistry(
backend=InMemoryBackend(),
environment="production",
cache_ttl=300,
)
await registry.initialize()
# Register portable configuration
await registry.register("client-123", {
"bot": {
"llm": {"$resource": "default", "type": "llm_providers"},
}
})
# Get bot for a client
bot = await registry.get_bot("client-123")
# Use the bot
response = await bot.chat(message, context)
Initialize bot registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
backend
|
RegistryBackend | None
|
Storage backend for configurations. If None, uses InMemoryBackend. |
None
|
environment
|
EnvironmentConfig | str | None
|
Environment name or EnvironmentConfig for $resource resolution. If None, configs are used as-is without environment resolution. |
None
|
env_dir
|
str | Path
|
Directory containing environment config files. Only used if environment is a string name. |
'config/environments'
|
cache_ttl
|
int
|
Cache time-to-live in seconds (default: 300) |
300
|
max_cache_size
|
int
|
Maximum cached bots (default: 1000) |
1000
|
validate_on_register
|
bool
|
If True, validate config portability when registering (default: True) |
True
|
config_key
|
str
|
Key within config containing bot configuration. Defaults to "bot". Used during environment resolution. |
'bot'
|
Methods:
| Name | Description |
|---|---|
initialize |
Initialize the registry and backend. |
close |
Close the registry and backend. |
register |
Register or update a bot configuration. |
get_bot |
Get bot instance for a client. |
get_config |
Get stored configuration for a bot. |
get_registration |
Get full registration including metadata. |
unregister |
Remove a bot registration (hard delete). |
deactivate |
Deactivate a bot registration (soft delete). |
exists |
Check if an active bot registration exists. |
list_bots |
List all active bot IDs. |
count |
Count active bot registrations. |
get_cached_bots |
Get list of currently cached bot IDs. |
clear_cache |
Clear all cached bot instances. |
register_client |
Register or update a client's bot configuration. |
remove_client |
Remove a client from the registry. |
get_cached_clients |
Get list of currently cached client IDs. |
__repr__ |
String representation. |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
Attributes¶
environment
property
¶
Get current environment config, or None if not environment-aware.
environment_name
property
¶
Get current environment name, or None if not environment-aware.
Functions¶
initialize
async
¶
Initialize the registry and backend.
Must be called before using the registry.
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
close
async
¶
Close the registry and backend.
Closes all cached bot instances and the storage backend.
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
register
async
¶
register(
bot_id: str,
config: dict[str, Any],
status: str = "active",
skip_validation: bool = False,
) -> Registration
Register or update a bot configuration.
Stores a portable configuration in the backend. By default, validates that the configuration is portable (no resolved local values).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Unique bot identifier |
required |
config
|
dict[str, Any]
|
Bot configuration dictionary (should be portable) |
required |
status
|
str
|
Registration status (default: active) |
'active'
|
skip_validation
|
bool
|
If True, skip portability validation |
False
|
Returns:
| Type | Description |
|---|---|
Registration
|
Registration object with metadata |
Raises:
| Type | Description |
|---|---|
PortabilityError
|
If config is not portable and validation is enabled |
Example
# Register with portable config
reg = await registry.register("support-bot", {
"bot": {
"llm": {"$resource": "default", "type": "llm_providers"},
}
})
print(f"Registered at: {reg.created_at}")
# Update existing registration
reg = await registry.register("support-bot", new_config)
print(f"Updated at: {reg.updated_at}")
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
get_bot
async
¶
Get bot instance for a client.
Bots are cached for performance. If a cached bot exists and hasn't expired, it's returned. Otherwise, a new bot is created from the stored configuration with environment resolution applied.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
force_refresh
|
bool
|
If True, bypass cache and create fresh bot |
False
|
Returns:
| Type | Description |
|---|---|
DynaBot
|
DynaBot instance for the client |
Raises:
| Type | Description |
|---|---|
KeyError
|
If no registration exists for the bot_id |
ValueError
|
If bot configuration is invalid |
Example
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
get_config
async
¶
Get stored configuration for a bot.
Returns the portable configuration as stored, without environment resolution applied.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
Configuration dict if found, None otherwise |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
get_registration
async
¶
Get full registration including metadata.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
Registration | None
|
Registration if found, None otherwise |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
unregister
async
¶
Remove a bot registration (hard delete).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if removed, False if not found |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
deactivate
async
¶
Deactivate a bot registration (soft delete).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if deactivated, False if not found |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
exists
async
¶
Check if an active bot registration exists.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot_id
|
str
|
Bot identifier |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if registration exists and is active |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
list_bots
async
¶
List all active bot IDs.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of active bot identifiers |
count
async
¶
Count active bot registrations.
Returns:
| Type | Description |
|---|---|
int
|
Number of active registrations |
get_cached_bots ¶
Get list of currently cached bot IDs.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of bot IDs with cached instances |
clear_cache
async
¶
Clear all cached bot instances.
Closes each bot before removing it from cache. Does not affect stored registrations.
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
register_client
async
¶
Register or update a client's bot configuration.
.. deprecated::
Use :meth:register instead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client_id
|
str
|
Client/tenant identifier |
required |
bot_config
|
dict[str, Any]
|
Bot configuration dictionary |
required |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
remove_client
async
¶
Remove a client from the registry.
.. deprecated::
Use :meth:unregister instead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client_id
|
str
|
Client/tenant identifier |
required |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
get_cached_clients ¶
Get list of currently cached client IDs.
.. deprecated::
Use :meth:get_cached_bots instead.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of client IDs with cached bots |
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
__repr__ ¶
String representation.
Source code in packages/bots/src/dataknobs_bots/bot/registry.py
DynaBot ¶
DynaBot(
llm: AsyncLLMProvider,
prompt_builder: AsyncPromptBuilder,
conversation_storage: ConversationStorage,
tool_registry: ToolRegistry | None = None,
memory: Memory | None = None,
knowledge_base: KnowledgeBase | None = None,
kb_auto_context: bool = True,
reasoning_strategy: Any | None = None,
middleware: list[Middleware] | None = None,
system_prompt_name: str | None = None,
system_prompt_content: str | None = None,
system_prompt_rag_configs: list[dict[str, Any]] | None = None,
default_temperature: float = 0.7,
default_max_tokens: int = 1000,
context_transform: Callable[[str], str] | None = None,
max_tool_iterations: int = _DEFAULT_MAX_TOOL_ITERATIONS,
tool_timeout: float = _DEFAULT_TOOL_TIMEOUT,
tool_loop_timeout: float = _DEFAULT_TOOL_LOOP_TIMEOUT,
)
Configuration-driven chatbot leveraging the DataKnobs ecosystem.
DynaBot provides a flexible, configuration-driven bot that can be customized for different use cases through YAML/JSON configuration files.
.. versionadded:: 0.14.0
DynaBot-level tool execution loop — strategies that pass tools to the
LLM but do not execute tool_calls themselves (e.g. SimpleReasoning)
now have their tool calls executed automatically by the bot pipeline.
Attributes:
| Name | Type | Description |
|---|---|---|
llm |
LLM provider for generating responses |
|
prompt_builder |
Prompt builder for managing prompts |
|
conversation_storage |
Storage backend for conversations |
|
tool_registry |
Registry of available tools |
|
memory |
Optional memory implementation for context |
|
knowledge_base |
Optional knowledge base for RAG |
|
reasoning_strategy |
Optional reasoning strategy |
|
middleware |
list[Middleware]
|
List of middleware for request/response processing |
system_prompt_name |
Name of the system prompt template to use |
|
system_prompt_content |
Inline system prompt content (alternative to name) |
|
system_prompt_rag_configs |
RAG configurations for inline system prompts |
|
default_temperature |
Default temperature for LLM generation |
|
default_max_tokens |
Default max tokens for LLM generation |
Initialize DynaBot.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
llm
|
AsyncLLMProvider
|
LLM provider instance |
required |
prompt_builder
|
AsyncPromptBuilder
|
Prompt builder instance |
required |
conversation_storage
|
ConversationStorage
|
Conversation storage backend |
required |
tool_registry
|
ToolRegistry | None
|
Optional tool registry |
None
|
memory
|
Memory | None
|
Optional memory implementation |
None
|
knowledge_base
|
KnowledgeBase | None
|
Optional knowledge base |
None
|
kb_auto_context
|
bool
|
Whether to auto-inject KB results into messages. When False, the KB is still available for tool-based access but not automatically queried on every message. |
True
|
reasoning_strategy
|
Any | None
|
Optional reasoning strategy |
None
|
middleware
|
list[Middleware] | None
|
Optional list of Middleware instances |
None
|
system_prompt_name
|
str | None
|
Name of system prompt template (mutually exclusive with content) |
None
|
system_prompt_content
|
str | None
|
Inline system prompt content (mutually exclusive with name) |
None
|
system_prompt_rag_configs
|
list[dict[str, Any]] | None
|
RAG configurations for inline system prompts |
None
|
default_temperature
|
float
|
Default temperature (0-1) |
0.7
|
default_max_tokens
|
int
|
Default max tokens to generate |
1000
|
context_transform
|
Callable[[str], str] | None
|
Optional callable applied to each content string (KB chunks, memory context) before it is injected into the prompt. Use this to sanitize or fence external content against prompt injection. |
None
|
max_tool_iterations
|
int
|
Maximum number of tool execution rounds
before returning. When a strategy returns a response with
|
_DEFAULT_MAX_TOOL_ITERATIONS
|
tool_timeout
|
float
|
Per-tool execution timeout in seconds. If a single tool call exceeds this duration, it is cancelled and an error observation is recorded. |
_DEFAULT_TOOL_TIMEOUT
|
tool_loop_timeout
|
float
|
Wall-clock budget in seconds for the
tool execution loop (across all iterations). Checked
at the start of each iteration and before each LLM
re-call. For |
_DEFAULT_TOOL_LOOP_TIMEOUT
|
Methods:
| Name | Description |
|---|---|
register_provider |
Register an auxiliary LLM/embedding provider by role. |
get_provider |
Get a registered provider by role. |
from_config |
Create DynaBot from configuration. |
from_environment_aware_config |
Create DynaBot with environment-aware configuration. |
get_portable_config |
Extract portable configuration for storage. |
chat |
Process a chat message. |
greet |
Generate a bot-initiated greeting before the user speaks. |
stream_chat |
Stream chat response token by token. |
get_conversation |
Retrieve conversation history. |
clear_conversation |
Clear a conversation's history. |
get_wizard_state |
Get current wizard state for a conversation. |
close |
Close the bot and clean up resources. |
__aenter__ |
Async context manager entry. |
__aexit__ |
Async context manager exit - ensures cleanup. |
get_conversation_manager |
Get a cached conversation manager by conversation ID. |
undo_last_turn |
Undo the last conversational turn (user message + bot response). |
rewind_to_turn |
Rewind conversation to after the given turn number. |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 | |
Attributes¶
all_providers
property
¶
All registered providers keyed by role.
Always includes "main" (self.llm). Subsystems add
their own entries during construction. Returns a fresh dict
(snapshot) on each call.
Functions¶
register_provider ¶
Register an auxiliary LLM/embedding provider by role.
Providers registered here are included in all_providers for
observability and enumeration. The registry is a catalog — it
does not manage provider lifecycle. Each subsystem closes the
providers it created (originator-owns-lifecycle).
The "main" role is reserved for self.llm and cannot be
overwritten.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
role
|
str
|
Unique role identifier (e.g. |
required |
provider
|
AsyncLLMProvider
|
The provider instance. |
required |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
get_provider ¶
Get a registered provider by role.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
role
|
str
|
Provider role identifier. |
required |
Returns:
| Type | Description |
|---|---|
AsyncLLMProvider | None
|
The provider, or |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
from_config
async
classmethod
¶
from_config(
config: dict[str, Any],
*,
llm: AsyncLLMProvider | None = None,
middleware: list[Middleware] | None = None,
) -> DynaBot
Create DynaBot from configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary containing:
- llm: LLM configuration (provider, model, etc.).
Optional when the |
required |
llm
|
AsyncLLMProvider | None
|
Pre-built LLM provider. When provided, |
None
|
middleware
|
list[Middleware] | None
|
Pre-built middleware list. When provided, replaces any middleware defined in config. |
None
|
Returns:
| Type | Description |
|---|---|
DynaBot
|
Configured DynaBot instance |
System Prompt Formats
The system_prompt can be specified in multiple ways:
-
String: Smart detection - if the string exists as a template name in the prompt library, it's used as a template reference; otherwise it's treated as inline content.
-
Dict with name:
{"name": "template_name"}- explicit template reference - Dict with name + strict:
{"name": "template_name", "strict": true}- raises error if template doesn't exist - Dict with content:
{"content": "inline prompt text"}- inline content - Dict with content + rag_configs: inline content with RAG enhancement
Example
bot = await DynaBot.from_config(config)
# With a shared provider
shared_llm = OllamaProvider({"provider": "ollama", "model": "llama3.2"})
await shared_llm.initialize()
bot = await DynaBot.from_config(
{"conversation_storage": {"backend": "memory"}},
llm=shared_llm,
)
# With pre-built middleware
bot = await DynaBot.from_config(config, middleware=[my_middleware])
Source code in packages/bots/src/dataknobs_bots/bot/base.py
295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 | |
from_environment_aware_config
async
classmethod
¶
from_environment_aware_config(
config: EnvironmentAwareConfig | dict[str, Any],
environment: EnvironmentConfig | str | None = None,
env_dir: str | Path = "config/environments",
config_key: str = "bot",
) -> DynaBot
Create DynaBot with environment-aware configuration.
This is the recommended entry point for environment-portable bots. Resource references ($resource) are resolved against the environment config, and environment variables are substituted at instantiation time (late binding).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
EnvironmentAwareConfig | dict[str, Any]
|
EnvironmentAwareConfig instance or dict with $resource references. If dict, will be wrapped in EnvironmentAwareConfig. |
required |
environment
|
EnvironmentConfig | str | None
|
Environment name or EnvironmentConfig instance. If None, auto-detects from DATAKNOBS_ENVIRONMENT env var. Ignored if config is already an EnvironmentAwareConfig. |
None
|
env_dir
|
str | Path
|
Directory containing environment config files. Only used if environment is a string name. |
'config/environments'
|
config_key
|
str
|
Key within config containing bot configuration. Defaults to "bot". Set to None to use root config. |
'bot'
|
Returns:
| Type | Description |
|---|---|
DynaBot
|
Fully initialized DynaBot instance with resolved resources |
Example
# With portable config dict
config = {
"bot": {
"llm": {
"$resource": "default",
"type": "llm_providers",
"temperature": 0.7,
},
"conversation_storage": {
"$resource": "conversations",
"type": "databases",
},
}
}
bot = await DynaBot.from_environment_aware_config(config)
# With explicit environment
bot = await DynaBot.from_environment_aware_config(
config,
environment="production",
env_dir="configs/environments"
)
# With EnvironmentAwareConfig instance
from dataknobs_config import EnvironmentAwareConfig
env_config = EnvironmentAwareConfig.load_app("my-bot", ...)
bot = await DynaBot.from_environment_aware_config(env_config)
Note
The config should use $resource references for infrastructure:
bot:
llm:
$resource: default # Logical name
type: llm_providers # Resource type
temperature: 0.7 # Behavioral param (portable)
The environment config provides concrete bindings:
Source code in packages/bots/src/dataknobs_bots/bot/base.py
710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 | |
get_portable_config
staticmethod
¶
Extract portable configuration for storage.
Returns configuration with $resource references intact and environment variables unresolved. This is the config that should be stored in registries or databases for cross-environment portability.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
EnvironmentAwareConfig | dict[str, Any]
|
EnvironmentAwareConfig instance or portable dict |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Portable configuration dictionary |
Example
from dataknobs_config import EnvironmentAwareConfig
# From EnvironmentAwareConfig
env_config = EnvironmentAwareConfig.load_app("my-bot", ...)
portable = DynaBot.get_portable_config(env_config)
# Store portable config in registry
await registry.store(bot_id, portable)
# Dict passes through unchanged
portable = DynaBot.get_portable_config({"bot": {...}})
Source code in packages/bots/src/dataknobs_bots/bot/base.py
chat
async
¶
chat(
message: str,
context: BotContext,
temperature: float | None = None,
max_tokens: int | None = None,
rag_query: str | None = None,
llm_config_overrides: dict[str, Any] | None = None,
plugin_data: dict[str, Any] | None = None,
**kwargs: Any,
) -> str
Process a chat message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
User message to process |
required |
context
|
BotContext
|
Bot execution context |
required |
temperature
|
float | None
|
Optional temperature override |
None
|
max_tokens
|
int | None
|
Optional max tokens override |
None
|
rag_query
|
str | None
|
Optional explicit query for knowledge base retrieval. If provided, this is used instead of the message for RAG. Useful when the message contains literal text to analyze (e.g., "Analyze this prompt: [prompt text]") but you want to search for analysis techniques instead. |
None
|
llm_config_overrides
|
dict[str, Any] | None
|
Optional dict to override LLM config fields for this request only. Supported fields: model, temperature, max_tokens, top_p, stop_sequences, seed, options. |
None
|
plugin_data
|
dict[str, Any] | None
|
Optional dict to seed |
None
|
**kwargs
|
Any
|
Additional arguments |
{}
|
Returns:
| Type | Description |
|---|---|
str
|
Bot response as string |
Example
context = BotContext(
conversation_id="conv-123",
client_id="client-456",
user_id="user-789"
)
response = await bot.chat("Hello!", context)
# With explicit RAG query
response = await bot.chat(
"Analyze this: Write a poem about cats",
context,
rag_query="prompt analysis techniques evaluation"
)
# With LLM config overrides (switch model per-request)
response = await bot.chat(
"Explain quantum computing",
context,
llm_config_overrides={"model": "gpt-4-turbo", "temperature": 0.9}
)
Source code in packages/bots/src/dataknobs_bots/bot/base.py
1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 | |
greet
async
¶
greet(
context: BotContext,
*,
initial_context: dict[str, Any] | None = None,
plugin_data: dict[str, Any] | None = None,
) -> str | None
Generate a bot-initiated greeting before the user speaks.
Delegates to the reasoning strategy's greet() method. Returns
None if the bot has no reasoning strategy or the strategy does
not support greetings (e.g. non-wizard strategies).
No user message is added to conversation history — the greeting is a bot-initiated assistant message only.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
BotContext
|
Bot execution context |
required |
initial_context
|
dict[str, Any] | None
|
Optional dict of initial data to seed into
the reasoning strategy's state before generating the
greeting. For wizard strategies, these values are merged
into |
None
|
plugin_data
|
dict[str, Any] | None
|
Optional dict to seed When |
None
|
Returns:
| Type | Description |
|---|---|
str | None
|
Greeting string, or None if the bot does not support greetings |
Note
Middleware lifecycle for greet: on_turn_start(turn) and
before_message("") are called before greeting generation;
after_turn(turn) and after_message(...) are called on
success (only when a response is generated);
finally_turn(turn) fires on success, error, and when
the strategy returns None (no greeting).
If an error occurs, on_error hooks receive
message="" since there is no user message. If a
middleware hook itself fails, on_hook_error is called on
all middleware.
Example
Source code in packages/bots/src/dataknobs_bots/bot/base.py
1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 | |
stream_chat
async
¶
stream_chat(
message: str,
context: BotContext,
temperature: float | None = None,
max_tokens: int | None = None,
rag_query: str | None = None,
llm_config_overrides: dict[str, Any] | None = None,
plugin_data: dict[str, Any] | None = None,
**kwargs: Any,
) -> AsyncGenerator[LLMStreamResponse, None]
Stream chat response token by token.
Similar to chat() but yields LLMStreamResponse objects as they are
generated, providing both the text delta and rich metadata (usage,
finish_reason, is_final) for each chunk.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
User message to process |
required |
context
|
BotContext
|
Bot execution context |
required |
temperature
|
float | None
|
Optional temperature override |
None
|
max_tokens
|
int | None
|
Optional max tokens override |
None
|
rag_query
|
str | None
|
Optional explicit query for knowledge base retrieval. If provided, this is used instead of the message for RAG. |
None
|
llm_config_overrides
|
dict[str, Any] | None
|
Optional dict to override LLM config fields for this request only. Supported fields: model, temperature, max_tokens, top_p, stop_sequences, seed, options. |
None
|
plugin_data
|
dict[str, Any] | None
|
Optional dict to seed |
None
|
**kwargs
|
Any
|
Additional arguments passed to LLM |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncGenerator[LLMStreamResponse, None]
|
LLMStreamResponse objects with |
AsyncGenerator[LLMStreamResponse, None]
|
|
Example
context = BotContext(
conversation_id="conv-123",
client_id="client-456",
user_id="user-789"
)
# Stream and display in real-time
async for chunk in bot.stream_chat("Explain quantum computing", context):
print(chunk.delta, end="", flush=True)
print() # Newline after streaming
# Accumulate response
full_response = ""
async for chunk in bot.stream_chat("Hello!", context):
full_response += chunk.delta
# With LLM config overrides
async for chunk in bot.stream_chat(
"Explain quantum computing",
context,
llm_config_overrides={"model": "gpt-4-turbo"}
):
print(chunk.delta, end="", flush=True)
Note
Conversation history is automatically updated after streaming completes. When a reasoning_strategy is configured, the strategy produces the complete response and it is emitted as a single stream chunk.
Cleanup guarantee: finally_turn middleware fires via a
finally block inside the async generator. In Python, async
generator finally blocks execute only when the generator is
fully consumed, explicitly closed (await gen.aclose()), or
garbage collected. Callers that break out of the stream early
should use contextlib.aclosing to guarantee prompt cleanup::
from contextlib import aclosing
async with aclosing(bot.stream_chat("msg", ctx)) as stream:
async for chunk in stream:
if done:
break # aclose() fires finally_turn
Source code in packages/bots/src/dataknobs_bots/bot/base.py
1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 | |
get_conversation
async
¶
Retrieve conversation history.
This method fetches the complete conversation state including all messages, metadata, and the message tree structure. Useful for displaying conversation history, debugging, analytics, or exporting conversations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conversation_id
|
str
|
Unique identifier of the conversation to retrieve |
required |
Returns:
| Type | Description |
|---|---|
Any
|
ConversationState object containing the full conversation history, |
Any
|
or None if the conversation does not exist |
Example
See Also
- clear_conversation(): Clear/delete a conversation
- chat(): Add messages to a conversation
Source code in packages/bots/src/dataknobs_bots/bot/base.py
clear_conversation
async
¶
Clear a conversation's history.
This method removes the conversation from both persistent storage and the internal cache. The next chat() call with this conversation_id will start a fresh conversation. Useful for:
- Implementing "start over" functionality
- Privacy/data deletion requirements
- Testing and cleanup
- Resetting conversation context
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conversation_id
|
str
|
Unique identifier of the conversation to clear |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the conversation was deleted, False if it didn't exist |
Example
Note
This operation is permanent and cannot be undone. The conversation cannot be recovered after deletion.
See Also
- get_conversation(): Retrieve conversation before clearing
- chat(): Will create new conversation after clearing
Source code in packages/bots/src/dataknobs_bots/bot/base.py
get_wizard_state
async
¶
Get current wizard state for a conversation.
This method provides public access to wizard state without requiring access to private conversation managers. It checks the in-memory manager first (most current) and falls back to persisted storage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conversation_id
|
str
|
Conversation identifier |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
Wizard state dict with canonical structure, or None if no wizard |
dict[str, Any] | None
|
active or conversation not found. |
The returned dict follows the canonical schema
{ "current_stage": str, "stage_index": int, "total_stages": int, "progress": float, "completed": bool, "data": dict, "can_skip": bool, "can_go_back": bool, "suggestions": list[str], "history": list[str], }
Example
Source code in packages/bots/src/dataknobs_bots/bot/base.py
close
async
¶
Close the bot and clean up resources.
This method closes the LLM provider, conversation storage backend, reasoning strategy, and releases associated resources like HTTP connections and database connections. Should be called when the bot is no longer needed, especially in testing or when creating temporary bot instances.
Example
Note
After calling close(), the bot should not be used for further operations. Create a new bot instance if needed.
Source code in packages/bots/src/dataknobs_bots/bot/base.py
__aenter__
async
¶
Async context manager entry.
Returns:
| Type | Description |
|---|---|
Self
|
Self for use in async with statement |
__aexit__
async
¶
__aexit__(
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None
Async context manager exit - ensures cleanup.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exc_type
|
type[BaseException] | None
|
Exception type if an exception occurred |
required |
exc_val
|
BaseException | None
|
Exception value if an exception occurred |
required |
exc_tb
|
TracebackType | None
|
Exception traceback if an exception occurred |
required |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
get_conversation_manager ¶
Get a cached conversation manager by conversation ID.
Returns None if no manager exists for the given ID (i.e. no
turn has been processed for that conversation yet). Use this for
cross-layer integration testing (e.g. injecting LLM-layer
ConversationMiddleware into a manager after construction).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
conversation_id
|
str
|
Conversation identifier |
required |
Returns:
| Type | Description |
|---|---|
ConversationManager | None
|
Cached ConversationManager, or None |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
undo_last_turn
async
¶
Undo the last conversational turn (user message + bot response).
Navigates the conversation tree back to the node_id recorded before the last turn started. The next chat() call will create a new branch from that point. The original branch is preserved in the tree.
Also rolls back: - Memory layer (pop N messages based on node depth difference) - Wizard FSM state (restored from per-node metadata) - Memory banks (reverted via backend-managed checkpointing)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
BotContext
|
Bot execution context (identifies the conversation). |
required |
Returns:
| Type | Description |
|---|---|
UndoResult
|
UndoResult with details about what was undone. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If there's nothing to undo (at start of conversation). |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583 2584 2585 2586 2587 2588 2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600 2601 2602 2603 2604 2605 2606 2607 2608 2609 2610 2611 2612 2613 2614 2615 2616 2617 2618 2619 2620 2621 2622 2623 2624 2625 2626 2627 2628 2629 2630 2631 2632 2633 2634 2635 2636 2637 2638 2639 2640 2641 2642 | |
rewind_to_turn
async
¶
Rewind conversation to after the given turn number.
Turn 0 is the first user-bot exchange. Rewinding to turn -1 means back to the start (before any user messages).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
BotContext
|
Bot execution context. |
required |
turn
|
int
|
Turn number to rewind to (-1 for conversation start). |
required |
Returns:
| Type | Description |
|---|---|
UndoResult
|
UndoResult with details about what was undone. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If turn number is invalid. |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
UndoResult
dataclass
¶
UndoResult(
undone_user_message: str,
undone_bot_response: str,
remaining_turns: int,
branching: bool,
)
Result of an undo operation.
ConfigDraftManager ¶
ConfigDraftManager(
output_dir: Path,
draft_prefix: str = "_draft-",
max_age_hours: float = 24.0,
metadata_key: str = "_draft",
)
File-based draft manager for interactive config creation.
Manages the lifecycle of configuration drafts: creation, incremental updates, finalization, and cleanup of stale drafts.
Draft files are named {prefix}{draft_id}.yaml and stored in
the output directory. When a config_name is provided, a named alias
file {config_name}.yaml is also maintained.
Initialize the draft manager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
output_dir
|
Path
|
Directory for draft and config files. |
required |
draft_prefix
|
str
|
Prefix for draft file names. |
'_draft-'
|
max_age_hours
|
float
|
Default maximum age for stale draft cleanup. |
24.0
|
metadata_key
|
str
|
Key used to store draft metadata in config files. |
'_draft'
|
Methods:
| Name | Description |
|---|---|
create_draft |
Create a new draft from a config dict. |
update_draft |
Update an existing draft. |
get_draft |
Retrieve a draft and its metadata. |
finalize |
Finalize a draft into a completed configuration. |
discard |
Discard a draft by removing its file. |
list_drafts |
List all current drafts. |
cleanup_stale |
Remove drafts older than the specified age. |
Attributes:
| Name | Type | Description |
|---|---|---|
output_dir |
Path
|
The output directory for drafts. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
Attributes¶
Functions¶
create_draft ¶
Create a new draft from a config dict.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary to save as draft. |
required |
stage
|
str | None
|
Current wizard stage. |
None
|
Returns:
| Type | Description |
|---|---|
str
|
The generated draft ID. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
update_draft ¶
update_draft(
draft_id: str,
config: dict[str, Any],
stage: str | None = None,
config_name: str | None = None,
) -> None
Update an existing draft.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
draft_id
|
str
|
The draft ID to update. |
required |
config
|
dict[str, Any]
|
Updated configuration dictionary. |
required |
stage
|
str | None
|
Current wizard stage. |
None
|
config_name
|
str | None
|
Optional name for the config file alias. |
None
|
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
If the draft file does not exist. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
get_draft ¶
Retrieve a draft and its metadata.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
draft_id
|
str
|
The draft ID to retrieve. |
required |
Returns:
| Type | Description |
|---|---|
tuple[dict[str, Any], DraftMetadata] | None
|
Tuple of (config_dict, metadata), or None if not found. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
finalize ¶
Finalize a draft into a completed configuration.
Strips draft metadata, writes the final config file, and removes the draft file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
draft_id
|
str
|
The draft ID to finalize. |
required |
final_name
|
str | None
|
Name for the final config file. If not provided, uses the config_name from draft metadata. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The finalized configuration dict (without draft metadata). |
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
If the draft does not exist. |
ValueError
|
If no final name can be determined. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
discard ¶
Discard a draft by removing its file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
draft_id
|
str
|
The draft ID to discard. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the draft was found and removed, False otherwise. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
list_drafts ¶
List all current drafts.
Returns:
| Type | Description |
|---|---|
list[DraftMetadata]
|
List of DraftMetadata for all draft files. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
cleanup_stale ¶
Remove drafts older than the specified age.
Also strips stale draft metadata blocks from named config files.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_age_hours
|
float | None
|
Maximum age in hours. Defaults to the manager's configured max_age_hours. |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of stale drafts removed. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
ConfigTemplate
dataclass
¶
ConfigTemplate(
name: str,
description: str = "",
version: str = "1.0.0",
tags: list[str] = list(),
variables: list[TemplateVariable] = list(),
structure: dict[str, Any] = dict(),
)
A reusable DynaBot configuration template.
Templates define a configuration structure with variable placeholders
({{var}}) that are substituted when the template is applied.
Attributes:
| Name | Type | Description |
|---|---|---|
name |
str
|
Template identifier (underscores internally). |
description |
str
|
Human-readable description. |
version |
str
|
Semantic version string. |
tags |
list[str]
|
Tags for filtering and categorization. |
variables |
list[TemplateVariable]
|
List of template variables. |
structure |
dict[str, Any]
|
The config structure with {{var}} placeholders. |
Methods:
| Name | Description |
|---|---|
get_required_variables |
Get variables that must be provided. |
get_optional_variables |
Get variables that have defaults or are not required. |
to_dict |
Convert to dictionary representation. |
from_dict |
Create a ConfigTemplate from a dictionary. |
from_yaml_file |
Load a ConfigTemplate from a YAML file. |
Functions¶
get_required_variables ¶
get_optional_variables ¶
Get variables that have defaults or are not required.
to_dict ¶
Convert to dictionary representation.
Source code in packages/bots/src/dataknobs_bots/config/templates.py
from_dict
classmethod
¶
Create a ConfigTemplate from a dictionary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary with template fields. |
required |
Returns:
| Type | Description |
|---|---|
ConfigTemplate
|
A new ConfigTemplate instance. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
from_yaml_file
classmethod
¶
Load a ConfigTemplate from a YAML file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path
|
Path to the YAML file. |
required |
Returns:
| Type | Description |
|---|---|
ConfigTemplate
|
A new ConfigTemplate instance. |
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
If the file does not exist. |
YAMLError
|
If the file is not valid YAML. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
ConfigTemplateRegistry ¶
Registry for managing and applying configuration templates.
Supports registration, tag-based filtering, variable validation, and template application with variable substitution.
Methods:
| Name | Description |
|---|---|
register |
Register a template. |
get |
Get a template by name. |
list_templates |
List templates, optionally filtered by tags. |
load_from_file |
Load and register a template from a YAML file. |
load_from_directory |
Load and register all templates from a directory. |
apply_template |
Apply a template with variable substitution. |
validate_variables |
Validate variables against a template's requirements. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
Functions¶
register ¶
Register a template.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
template
|
ConfigTemplate
|
The template to register. |
required |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
get ¶
Get a template by name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Template name. |
required |
Returns:
| Type | Description |
|---|---|
ConfigTemplate | None
|
The template, or None if not found. |
list_templates ¶
List templates, optionally filtered by tags.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tags
|
list[str] | None
|
If provided, only return templates that have all specified tags. |
None
|
Returns:
| Type | Description |
|---|---|
list[ConfigTemplate]
|
List of matching templates. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
load_from_file ¶
Load and register a template from a YAML file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path
|
Path to the YAML file. |
required |
Returns:
| Type | Description |
|---|---|
ConfigTemplate
|
The loaded template. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
load_from_directory ¶
Load and register all templates from a directory.
Scans for *.yaml and *.yml files, skipping files named
README or base.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
directory
|
Path
|
Directory to scan. |
required |
Returns:
| Type | Description |
|---|---|
int
|
Number of templates loaded. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
apply_template ¶
Apply a template with variable substitution.
Deep-copies the template structure and substitutes all {{var}}
placeholders with values from the variables dict.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Template name. |
required |
variables
|
dict[str, Any]
|
Variable values to substitute. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The resolved configuration dict. |
Raises:
| Type | Description |
|---|---|
KeyError
|
If the template is not found. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
validate_variables ¶
Validate variables against a template's requirements.
Checks that required variables are present and that values match any defined choices constraints.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Template name. |
required |
variables
|
dict[str, Any]
|
Variable values to validate. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
ValidationResult with any issues found. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
ConfigValidator ¶
Pluggable validation engine for DynaBot configurations.
Runs a pipeline of validators against a config dict and collects all errors and warnings into a single ValidationResult.
Example
Initialize the validator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
schema
|
DynaBotConfigSchema | None
|
Optional config schema for schema-based validation. |
None
|
Methods:
| Name | Description |
|---|---|
register_validator |
Register a named validation function. |
validate |
Run all validators against a configuration. |
validate_completeness |
Check that a config has the minimum required fields. |
validate_portability |
Check that a config is portable across environments. |
validate_component |
Validate a specific component section of the config. |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
Functions¶
register_validator ¶
Register a named validation function.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Unique name for this validator. |
required |
validator
|
ValidatorFn
|
Function that takes a config dict and returns ValidationResult. |
required |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
validate ¶
Run all validators against a configuration.
Runs completeness check, schema validation (if schema provided), and all registered custom validators.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary to validate. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
Merged ValidationResult from all validators. |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
validate_completeness ¶
Check that a config has the minimum required fields.
A valid DynaBot config must have at minimum an LLM configuration and conversation storage configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary to check. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
ValidationResult with errors for missing required fields. |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
validate_portability ¶
Check that a config is portable across environments.
Wraps the portability checker from registry.portability to return a ValidationResult instead of raising exceptions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary to check. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
ValidationResult with portability issues as warnings. |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
validate_component ¶
Validate a specific component section of the config.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
component
|
str
|
Component name (e.g., 'llm', 'memory'). |
required |
config
|
dict[str, Any]
|
The component's configuration dictionary. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
ValidationResult for that component. |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
DraftMetadata
dataclass
¶
DraftMetadata(
draft_id: str,
created_at: str,
last_updated: str,
stage: str | None = None,
complete: bool = False,
config_name: str | None = None,
)
Metadata for a configuration draft.
Attributes:
| Name | Type | Description |
|---|---|---|
draft_id |
str
|
Unique identifier for the draft. |
created_at |
str
|
ISO 8601 creation timestamp. |
last_updated |
str
|
ISO 8601 last update timestamp. |
stage |
str | None
|
Current wizard stage when draft was saved. |
complete |
bool
|
Whether the draft represents a complete config. |
config_name |
str | None
|
Optional name for the final config file. |
Methods:
| Name | Description |
|---|---|
to_dict |
Convert to dictionary representation. |
from_dict |
Create DraftMetadata from a dictionary. |
Functions¶
to_dict ¶
Convert to dictionary representation.
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
from_dict
classmethod
¶
Create DraftMetadata from a dictionary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary with metadata fields. |
required |
Returns:
| Type | Description |
|---|---|
DraftMetadata
|
A new DraftMetadata instance. |
Source code in packages/bots/src/dataknobs_bots/config/drafts.py
DynaBotConfigBuilder ¶
Fluent builder for DynaBot configurations.
Provides setter methods for each DynaBot component that return self
for method chaining. Consumer-specific sections are added via
set_custom_section().
Two output formats:
- build() returns flat format compatible with DynaBot.from_config()
- build_portable() returns environment-aware format with $resource
references and a bot wrapper key
Initialize the builder.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
schema
|
DynaBotConfigSchema | None
|
Optional schema for validation. If not provided, a default schema is created. |
None
|
Methods:
| Name | Description |
|---|---|
set_llm |
Set the LLM provider configuration (flat/direct format). |
set_llm_resource |
Set the LLM configuration using a $resource reference. |
set_conversation_storage |
Set the conversation storage backend (flat/direct format). |
set_conversation_storage_resource |
Set conversation storage using a $resource reference. |
set_conversation_storage_class |
Set conversation storage using a custom ConversationStorage class. |
set_memory |
Set the memory configuration. |
set_config_base_path |
Set base path for resolving relative config file paths. |
set_reasoning |
Set the reasoning strategy. |
set_reasoning_wizard |
Set wizard reasoning with a config path, inline dict, or WizardConfig. |
set_system_prompt |
Set the system prompt configuration. |
set_knowledge_base |
Set the knowledge base configuration. |
add_tool |
Add a tool to the bot configuration. |
add_tool_by_name |
Add a tool to the config by looking up its catalog entry. |
add_tools_by_name |
Add multiple tools by name from the catalog. |
add_middleware |
Add middleware to the bot configuration. |
set_custom_section |
Set a custom (domain-specific) config section. |
from_template |
Initialize the builder from a template. |
merge_overrides |
Merge override values into the current configuration. |
validate |
Validate the current configuration. |
build |
Build the flat configuration dict. |
build_portable |
Build the portable configuration with $resource references. |
to_yaml |
Serialize the portable configuration as YAML. |
reset |
Reset the builder to an empty state. |
from_config |
Create a builder pre-populated from an existing config. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
Functions¶
set_llm ¶
Set the LLM provider configuration (flat/direct format).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
provider
|
str
|
LLM provider name (e.g., 'ollama', 'openai'). |
required |
model
|
str | None
|
Model name or identifier. |
None
|
**kwargs
|
Any
|
Additional provider-specific settings (temperature, max_tokens, etc.). |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_llm_resource ¶
set_llm_resource(
resource_name: str = "default",
resource_type: str = "llm_providers",
**overrides: Any,
) -> Self
Set the LLM configuration using a $resource reference.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resource_name
|
str
|
Resource name to resolve at runtime. |
'default'
|
resource_type
|
str
|
Resource type category. |
'llm_providers'
|
**overrides
|
Any
|
Override values applied after resolution. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_conversation_storage ¶
Set the conversation storage backend (flat/direct format).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
backend
|
str
|
Storage backend name (e.g., 'memory', 'sqlite'). |
required |
**kwargs
|
Any
|
Additional backend-specific settings. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_conversation_storage_resource ¶
set_conversation_storage_resource(
resource_name: str = "conversations",
resource_type: str = "databases",
**overrides: Any,
) -> Self
Set conversation storage using a $resource reference.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resource_name
|
str
|
Resource name to resolve at runtime. |
'conversations'
|
resource_type
|
str
|
Resource type category. |
'databases'
|
**overrides
|
Any
|
Override values applied after resolution. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_conversation_storage_class ¶
Set conversation storage using a custom ConversationStorage class.
The class must implement ConversationStorage and provide an async
create(config: dict) -> ConversationStorage classmethod.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
storage_class
|
str
|
Dotted import path to the storage class
(e.g., |
required |
**kwargs
|
Any
|
Additional config passed to |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_memory ¶
Set the memory configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
memory_type
|
str
|
Memory type (e.g., 'buffer', 'vector'). |
required |
**kwargs
|
Any
|
Additional memory settings (max_messages, etc.). |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_config_base_path ¶
Set base path for resolving relative config file paths.
When set, relative paths in nested configs (e.g. wizard_config)
are resolved against this directory instead of the current working
directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Base directory path (string or Path object). |
required |
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_reasoning ¶
Set the reasoning strategy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
strategy
|
str
|
Reasoning strategy (e.g., 'simple', 'react', 'wizard'). |
required |
**kwargs
|
Any
|
Additional strategy settings. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_reasoning_wizard ¶
Set wizard reasoning with a config path, inline dict, or WizardConfig.
When wizard_config is a WizardConfig object, the caller
is responsible for writing it to disk via wizard_config.to_file()
before the bot loads.
When wizard_config is a dict, it is stored inline in the
reasoning config and loaded via
WizardConfigLoader.load_from_dict() at bot startup.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
wizard_config
|
str | dict[str, Any] | WizardConfig
|
Path to a wizard YAML file, an inline dict
(compatible with |
required |
**kwargs
|
Any
|
Additional reasoning settings (extraction_config, etc.). |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_system_prompt ¶
set_system_prompt(
content: str | None = None,
name: str | None = None,
rag_configs: list[dict[str, Any]] | None = None,
) -> Self
Set the system prompt configuration.
Provide either content (inline prompt) or name (template
reference). Optionally add RAG configurations for prompt enhancement.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content
|
str | None
|
Inline prompt content. |
None
|
name
|
str | None
|
Prompt template name. |
None
|
rag_configs
|
list[dict[str, Any]] | None
|
RAG configurations for prompt enhancement. |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_knowledge_base ¶
Set the knowledge base configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Any
|
Knowledge base settings (enabled, type, vector_store, embedding, etc.). |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
add_tool ¶
Add a tool to the bot configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_class
|
str
|
Fully qualified tool class name. |
required |
**params
|
Any
|
Tool constructor parameters. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
add_tool_by_name ¶
Add a tool to the config by looking up its catalog entry.
Resolves the tool name to a class path via the catalog and adds it with default params (overridable).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
catalog
|
ToolCatalog
|
Tool catalog for name resolution. |
required |
name
|
str
|
Tool name to look up. |
required |
**param_overrides
|
Any
|
Override default params. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Raises:
| Type | Description |
|---|---|
NotFoundError
|
If tool name is not in the catalog. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
add_tools_by_name ¶
add_tools_by_name(
catalog: ToolCatalog,
names: Sequence[str],
overrides: dict[str, dict[str, Any]] | None = None,
) -> Self
Add multiple tools by name from the catalog.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
catalog
|
ToolCatalog
|
Tool catalog for name resolution. |
required |
names
|
Sequence[str]
|
Tool names to add. |
required |
overrides
|
dict[str, dict[str, Any]] | None
|
Per-tool param overrides keyed by tool name. |
None
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
add_middleware ¶
Add middleware to the bot configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
middleware_class
|
str
|
Fully qualified middleware class name. |
required |
**params
|
Any
|
Middleware constructor parameters. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
set_custom_section ¶
Set a custom (domain-specific) config section.
This is the extension point for consumers to add sections like
educational, customer_service, domain, etc.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Section key name. |
required |
value
|
Any
|
Section value (dict, list, or scalar). |
required |
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
from_template ¶
Initialize the builder from a template.
Deep-copies the template structure, substitutes variables, and uses the result as the builder's base configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
template
|
ConfigTemplate
|
The template to apply. |
required |
variables
|
dict[str, Any]
|
Variable values for substitution. |
required |
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
merge_overrides ¶
Merge override values into the current configuration.
Performs recursive dict merge for nested dictionaries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
overrides
|
dict[str, Any]
|
Override values to merge. |
required |
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
validate ¶
Validate the current configuration.
Returns:
| Type | Description |
|---|---|
ValidationResult
|
ValidationResult with any errors and warnings. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
build ¶
Build the flat configuration dict.
The returned dict is compatible with DynaBot.from_config().
Validates before returning and raises ValueError if there are errors.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Flat configuration dictionary. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the configuration has validation errors. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
build_portable ¶
Build the portable configuration with $resource references.
Wraps the config under a bot key and includes any custom
sections as top-level siblings.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Portable configuration dict with |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the configuration has validation errors. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
to_yaml ¶
Serialize the portable configuration as YAML.
Returns:
| Type | Description |
|---|---|
str
|
YAML string representation. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
reset ¶
Reset the builder to an empty state.
Returns:
| Type | Description |
|---|---|
Self
|
self for method chaining. |
from_config
classmethod
¶
Create a builder pre-populated from an existing config.
Supports both flat format and portable format (with bot wrapper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Existing configuration dictionary. |
required |
Returns:
| Type | Description |
|---|---|
DynaBotConfigBuilder
|
A new builder instance with the config loaded. |
Source code in packages/bots/src/dataknobs_bots/config/builder.py
DynaBotConfigSchema ¶
Queryable registry of valid DynaBot configuration options.
Auto-registers the 8 default DynaBot components on initialization. Consumers can register additional extensions for domain-specific sections.
Methods:
| Name | Description |
|---|---|
register_component |
Register a core DynaBot component schema. |
register_extension |
Register a consumer-specific config extension. |
get_component_schema |
Get the JSON Schema for a component. |
get_extension_schema |
Get the JSON Schema for an extension. |
get_valid_options |
Get valid options for a field within a component or extension. |
validate |
Validate a config against all registered schemas. |
get_full_schema |
Get the combined schema for all components and extensions. |
to_description |
Generate a human-readable description for LLM system prompts. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
Functions¶
register_component ¶
register_component(
name: str,
schema: dict[str, Any],
description: str = "",
required: bool = False,
) -> None
Register a core DynaBot component schema.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Component name. |
required |
schema
|
dict[str, Any]
|
JSON Schema-like definition. |
required |
description
|
str
|
Human-readable description. |
''
|
required
|
bool
|
Whether this component is required. |
False
|
Source code in packages/bots/src/dataknobs_bots/config/schema.py
register_extension ¶
Register a consumer-specific config extension.
Extensions are domain-specific sections (e.g., 'educational', 'customer_service') that aren't part of the core DynaBot schema.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Extension name. |
required |
schema
|
dict[str, Any]
|
JSON Schema-like definition. |
required |
description
|
str
|
Human-readable description. |
''
|
Source code in packages/bots/src/dataknobs_bots/config/schema.py
get_component_schema ¶
Get the JSON Schema for a component.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Component name. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
JSON Schema dict, or None if not registered. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
get_extension_schema ¶
Get the JSON Schema for an extension.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Extension name. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
JSON Schema dict, or None if not registered. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
get_valid_options ¶
Get valid options for a field within a component or extension.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
component
|
str
|
Component or extension name. |
required |
field_name
|
str
|
Field name to query. |
required |
Returns:
| Type | Description |
|---|---|
list[str]
|
List of valid option strings. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
validate ¶
Validate a config against all registered schemas.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Full DynaBot configuration dict. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
ValidationResult with all schema violations. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
get_full_schema ¶
Get the combined schema for all components and extensions.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict mapping component/extension names to their schemas. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
to_description ¶
Generate a human-readable description for LLM system prompts.
Returns:
| Type | Description |
|---|---|
str
|
Structured text describing all available configuration options. |
Source code in packages/bots/src/dataknobs_bots/config/schema.py
TemplateVariable
dataclass
¶
TemplateVariable(
name: str,
description: str = "",
type: str = "string",
required: bool = False,
default: Any = None,
choices: list[Any] | None = None,
validation: dict[str, Any] | None = None,
)
Definition of a template variable.
Attributes:
| Name | Type | Description |
|---|---|---|
name |
str
|
Variable name used in {{name}} placeholders. |
description |
str
|
Human-readable description. |
type |
str
|
Variable type (string, integer, boolean, enum, array). |
required |
bool
|
Whether the variable must be provided. |
default |
Any
|
Default value if not provided. |
choices |
list[Any] | None
|
Valid values for enum-type variables. |
validation |
dict[str, Any] | None
|
JSON Schema constraints for the value. |
Methods:
| Name | Description |
|---|---|
to_dict |
Convert to dictionary representation. |
from_dict |
Create a TemplateVariable from a dictionary. |
Functions¶
to_dict ¶
Convert to dictionary representation.
Source code in packages/bots/src/dataknobs_bots/config/templates.py
from_dict
classmethod
¶
Create a TemplateVariable from a dictionary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary with variable fields. |
required |
Returns:
| Type | Description |
|---|---|
TemplateVariable
|
A new TemplateVariable instance. |
Source code in packages/bots/src/dataknobs_bots/config/templates.py
ToolCatalog ¶
Registry mapping tool names to class paths and default configuration.
Provides a single source of truth for tool metadata, enabling config builders to reference tools by name and produce correct bot/wizard configs.
Built on Registry[ToolEntry] for thread safety, metrics, and
consistent error handling.
Example
Initialize the catalog.
Methods:
| Name | Description |
|---|---|
register_tool |
Register a tool in the catalog. |
register_entry |
Register a pre-built ToolEntry. |
register_from_dict |
Register a tool from a dict (e.g., loaded from YAML). |
register_many_from_dicts |
Register multiple tools from dicts. |
register_from_class |
Register a tool class that provides |
list_tools |
List all registered tools, optionally filtered by tags. |
get_names |
Get all registered tool names. |
to_bot_config |
Generate a bot config tool entry for the named tool. |
to_bot_configs |
Generate bot config entries for multiple tools. |
get_requirements |
Get the union of all requirements for the given tool names. |
check_requirements |
Check that tool requirements are satisfied by a config dict. |
instantiate_tool |
Import and instantiate a tool from its catalog entry. |
create_tool_registry |
Create a ToolRegistry populated from catalog entries. |
to_dict |
Serialize entire catalog to a dict (for YAML output). |
from_dict |
Create a catalog from a dict (e.g., loaded from YAML). |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
Functions¶
register_tool ¶
register_tool(
name: str,
class_path: str,
description: str = "",
default_params: dict[str, Any] | None = None,
tags: Sequence[str] = (),
requires: Sequence[str] = (),
) -> None
Register a tool in the catalog.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Tool's runtime name. |
required |
class_path
|
str
|
Fully-qualified class path. |
required |
description
|
str
|
Human-readable description. |
''
|
default_params
|
dict[str, Any] | None
|
Default constructor params. |
None
|
tags
|
Sequence[str]
|
Categorization tags. |
()
|
requires
|
Sequence[str]
|
Dependency identifiers. |
()
|
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
register_entry ¶
Register a pre-built ToolEntry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
entry
|
ToolEntry
|
ToolEntry to register. |
required |
register_from_dict ¶
Register a tool from a dict (e.g., loaded from YAML).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dict with |
required |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
register_many_from_dicts ¶
Register multiple tools from dicts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
entries
|
list[dict[str, Any]]
|
List of tool definition dicts. |
required |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
register_from_class ¶
Register a tool class that provides catalog_metadata().
Computes class_path automatically from the class's module path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_class
|
type
|
A tool class with a |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If tool_class does not implement |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
list_tools ¶
List all registered tools, optionally filtered by tags.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tags
|
Sequence[str] | None
|
If provided, return only tools that have ANY of the specified tags (union semantics). |
None
|
Returns:
| Type | Description |
|---|---|
list[ToolEntry]
|
List of matching ToolEntry instances. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
get_names ¶
Get all registered tool names.
Returns:
| Type | Description |
|---|---|
list[str]
|
List of tool names. |
to_bot_config ¶
Generate a bot config tool entry for the named tool.
Returns a dict suitable for DynaBot._resolve_tool():
{"class": "full.class.path", "params": {...}}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Tool name to look up. |
required |
**param_overrides
|
Any
|
Override default params. |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Bot config dict for the tool. |
Raises:
| Type | Description |
|---|---|
NotFoundError
|
If tool name is not registered. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
to_bot_configs ¶
to_bot_configs(
names: Sequence[str], overrides: dict[str, dict[str, Any]] | None = None
) -> list[dict[str, Any]]
Generate bot config entries for multiple tools.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
names
|
Sequence[str]
|
Tool names to include. |
required |
overrides
|
dict[str, dict[str, Any]] | None
|
Per-tool param overrides keyed by tool name. |
None
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of bot config dicts. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
get_requirements ¶
Get the union of all requirements for the given tool names.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
names
|
Sequence[str]
|
Tool names to check. |
required |
Returns:
| Type | Description |
|---|---|
frozenset[str]
|
Set of all requirement identifiers across the named tools. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
check_requirements ¶
Check that tool requirements are satisfied by a config dict.
Returns a list of warning messages for any unmet requirements. Tools with no requirements are always satisfied.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_names
|
Sequence[str]
|
Names of tools to check. |
required |
config
|
dict[str, Any]
|
Bot config dict to check against (top-level keys). |
required |
Returns:
| Type | Description |
|---|---|
list[str]
|
List of warning strings (empty if all requirements met). |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
instantiate_tool ¶
Import and instantiate a tool from its catalog entry.
Uses resolve_callable() to import the class, then instantiates
it with default_params merged with overrides. Prefers
from_config() if the class defines it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Tool name to instantiate. |
required |
**param_overrides
|
Any
|
Override default params. |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
Instantiated tool. |
Raises:
| Type | Description |
|---|---|
NotFoundError
|
If name not in catalog. |
ImportError
|
If class cannot be imported. |
ValueError
|
If resolved class is not callable. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
create_tool_registry ¶
create_tool_registry(
names: Sequence[str] | None = None,
overrides: dict[str, dict[str, Any]] | None = None,
strict: bool = False,
) -> Any
Create a ToolRegistry populated from catalog entries.
Imports and instantiates each named tool, registering them in a
new ToolRegistry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
names
|
Sequence[str] | None
|
Tool names to include (default: all registered). |
None
|
overrides
|
dict[str, dict[str, Any]] | None
|
Per-tool param overrides keyed by tool name. |
None
|
strict
|
bool
|
If True, raise on instantiation failure. If False (default), skip failed tools and log warnings. |
False
|
Returns:
| Type | Description |
|---|---|
Any
|
ToolRegistry with instantiated tools. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
to_dict ¶
Serialize entire catalog to a dict (for YAML output).
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
from_dict
classmethod
¶
Create a catalog from a dict (e.g., loaded from YAML).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dict with |
required |
Returns:
| Type | Description |
|---|---|
ToolCatalog
|
New ToolCatalog populated from the data. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
ToolEntry
dataclass
¶
ToolEntry(
name: str,
class_path: str,
description: str = "",
default_params: dict[str, Any] = None,
tags: frozenset[str] = frozenset(),
requires: frozenset[str] = frozenset(),
)
Metadata for a tool in the catalog.
Captures the information needed to: - Generate bot config entries (class path + params) - Reference tools in wizard stage configs (name) - Discover tools by capability (tags) - Validate tool dependencies (requires)
Methods:
| Name | Description |
|---|---|
__post_init__ |
Set default_params to empty dict if None. |
to_dict |
Serialize to dict (suitable for YAML output). |
from_dict |
Deserialize from dict (e.g., loaded from YAML). |
to_bot_config |
Generate a bot config tool entry. |
Functions¶
__post_init__ ¶
to_dict ¶
Serialize to dict (suitable for YAML output).
Omits empty/default fields for clean output.
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
from_dict
classmethod
¶
Deserialize from dict (e.g., loaded from YAML).
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
to_bot_config ¶
Generate a bot config tool entry.
Returns a dict suitable for DynaBot._resolve_tool():
{"class": "full.class.path", "params": {...}}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**param_overrides
|
Any
|
Override default params. |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Bot config dict for this tool. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
ValidationResult
dataclass
¶
Result of validating a configuration.
Attributes:
| Name | Type | Description |
|---|---|---|
valid |
bool
|
Whether the configuration passed validation. |
errors |
list[str]
|
List of error messages (validation failures). |
warnings |
list[str]
|
List of warning messages (non-blocking issues). |
Methods:
| Name | Description |
|---|---|
merge |
Merge another validation result into this one. |
ok |
Create a successful validation result. |
error |
Create a failed validation result with a single error. |
warning |
Create a successful validation result with a warning. |
to_dict |
Convert to dictionary representation. |
Functions¶
merge ¶
Merge another validation result into this one.
The merged result is valid only if both results are valid.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
ValidationResult
|
Another validation result to merge. |
required |
Returns:
| Type | Description |
|---|---|
ValidationResult
|
A new ValidationResult with combined errors and warnings. |
Source code in packages/bots/src/dataknobs_bots/config/validation.py
ok
classmethod
¶
error
classmethod
¶
Create a failed validation result with a single error.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
The error message. |
required |
warning
classmethod
¶
Create a successful validation result with a warning.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
The warning message. |
required |
to_dict ¶
RAGKnowledgeBase ¶
RAGKnowledgeBase(
vector_store: Any,
embedding_provider: Any,
chunking_config: dict[str, Any] | None = None,
merger_config: MergerConfig | None = None,
formatter_config: FormatterConfig | None = None,
)
Bases: KnowledgeBase
RAG knowledge base using dataknobs-xization for chunking and vector search.
This implementation: - Parses markdown documents using dataknobs-xization - Chunks documents intelligently based on structure - Stores chunks with embeddings in vector store - Provides semantic search for relevant context
Attributes:
| Name | Type | Description |
|---|---|---|
vector_store |
Vector store backend from dataknobs_data |
|
embedding_provider |
LLM provider for generating embeddings |
|
chunking_config |
Configuration for document chunking |
Initialize RAG knowledge base.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
vector_store
|
Any
|
Vector store backend instance |
required |
embedding_provider
|
Any
|
LLM provider with embed() method |
required |
chunking_config
|
dict[str, Any] | None
|
Configuration for chunking: - max_chunk_size: Maximum chunk size in characters - combine_under_heading: Combine text under same heading - quality_filter: ChunkQualityConfig for filtering - generate_embeddings: Whether to generate enriched embedding text |
None
|
merger_config
|
MergerConfig | None
|
Configuration for chunk merging (optional) |
None
|
formatter_config
|
FormatterConfig | None
|
Configuration for context formatting (optional) |
None
|
Methods:
| Name | Description |
|---|---|
from_config |
Create RAG knowledge base from configuration. |
load_markdown_document |
Load and chunk a markdown document from a file. |
load_documents_from_directory |
Load all markdown documents from a directory. |
load_json_document |
Load and chunk a JSON document by converting it to markdown. |
load_yaml_document |
Load and chunk a YAML document by converting it to markdown. |
load_csv_document |
Load and chunk a CSV document by converting it to markdown. |
load_from_directory |
Load documents from a directory using KnowledgeBaseConfig. |
load_markdown_text |
Load markdown content from a string. |
query |
Query knowledge base for relevant chunks. |
hybrid_query |
Query knowledge base using hybrid search (text + vector). |
format_context |
Format search results for LLM context. |
count |
Get the number of chunks in the knowledge base. |
clear |
Clear all documents from the knowledge base. |
save |
Save the knowledge base to persistent storage. |
providers |
Return the embedding provider, keyed by role. |
set_provider |
Replace the embedding provider if the role matches. |
close |
Close the knowledge base and release resources. |
__aenter__ |
Async context manager entry. |
__aexit__ |
Async context manager exit - ensures cleanup. |
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
Functions¶
from_config
async
classmethod
¶
Create RAG knowledge base from configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary with:
- vector_store: Vector store configuration
- embedding: Nested embedding config dict (preferred), e.g.
|
required |
Returns:
| Type | Description |
|---|---|
RAGKnowledgeBase
|
Configured RAGKnowledgeBase instance |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
load_markdown_document
async
¶
Load and chunk a markdown document from a file.
Reads the file and delegates to :meth:load_markdown_text for
parsing, chunking, embedding, and storage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath
|
str | Path
|
Path to markdown file |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata to attach to all chunks |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of chunks created |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
load_documents_from_directory
async
¶
Load all markdown documents from a directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
directory
|
str | Path
|
Directory path containing documents |
required |
pattern
|
str
|
Glob pattern for files to load (default: **/*.md) |
'**/*.md'
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with loading statistics: - total_files: Number of files processed - total_chunks: Total chunks created - errors: List of errors encountered |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
load_json_document
async
¶
load_json_document(
filepath: str | Path,
metadata: dict[str, Any] | None = None,
schema: str | None = None,
transformer: ContentTransformer | None = None,
title: str | None = None,
) -> int
Load and chunk a JSON document by converting it to markdown.
This method converts JSON data to markdown format using ContentTransformer, then processes it like any other markdown document.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath
|
str | Path
|
Path to JSON file |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata to attach to all chunks |
None
|
schema
|
str | None
|
Optional schema name (requires transformer with registered schema) |
None
|
transformer
|
ContentTransformer | None
|
Optional ContentTransformer instance with custom configuration |
None
|
title
|
str | None
|
Optional document title for the markdown |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of chunks created |
Example
# Generic conversion
num_chunks = await kb.load_json_document(
"data/patterns.json",
metadata={"content_type": "patterns"}
)
# With custom schema
transformer = ContentTransformer()
transformer.register_schema("pattern", {
"title_field": "name",
"sections": [
{"field": "description", "heading": "Description"},
{"field": "example", "heading": "Example", "format": "code"}
]
})
num_chunks = await kb.load_json_document(
"data/patterns.json",
transformer=transformer,
schema="pattern"
)
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
load_yaml_document
async
¶
load_yaml_document(
filepath: str | Path,
metadata: dict[str, Any] | None = None,
schema: str | None = None,
transformer: ContentTransformer | None = None,
title: str | None = None,
) -> int
Load and chunk a YAML document by converting it to markdown.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath
|
str | Path
|
Path to YAML file |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata to attach to all chunks |
None
|
schema
|
str | None
|
Optional schema name (requires transformer with registered schema) |
None
|
transformer
|
ContentTransformer | None
|
Optional ContentTransformer instance with custom configuration |
None
|
title
|
str | None
|
Optional document title for the markdown |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of chunks created |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
load_csv_document
async
¶
load_csv_document(
filepath: str | Path,
metadata: dict[str, Any] | None = None,
title: str | None = None,
title_field: str | None = None,
transformer: ContentTransformer | None = None,
) -> int
Load and chunk a CSV document by converting it to markdown.
Each row becomes a section with the first column (or title_field) as heading.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filepath
|
str | Path
|
Path to CSV file |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata to attach to all chunks |
None
|
title
|
str | None
|
Optional document title for the markdown |
None
|
title_field
|
str | None
|
Column to use as section title (default: first column) |
None
|
transformer
|
ContentTransformer | None
|
Optional ContentTransformer instance with custom configuration |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of chunks created |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
load_from_directory
async
¶
load_from_directory(
directory: str | Path,
config: KnowledgeBaseConfig | None = None,
progress_callback: Any | None = None,
) -> dict[str, Any]
Load documents from a directory using KnowledgeBaseConfig.
This method uses the xization DirectoryProcessor to process documents with configurable patterns, chunking, and metadata. It supports markdown, JSON, and JSONL files with streaming for large files.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
directory
|
str | Path
|
Directory path containing documents |
required |
config
|
KnowledgeBaseConfig | None
|
Optional KnowledgeBaseConfig. If not provided, attempts to load from knowledge_base.json/yaml in the directory, or uses defaults. |
None
|
progress_callback
|
Any | None
|
Optional callback function(file_path, num_chunks) for progress |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with loading statistics: - total_files: Number of files processed - total_chunks: Total chunks created - files_by_type: Count of files by type (markdown, json, jsonl) - errors: List of errors encountered - documents: List of processed document info |
Example
# With auto-loaded config from directory
results = await kb.load_from_directory("./docs")
# With explicit config
config = KnowledgeBaseConfig(
name="product-docs",
default_chunking={"max_chunk_size": 800},
patterns=[
FilePatternConfig(pattern="api/**/*.json", text_fields=["title", "description"]),
FilePatternConfig(pattern="**/*.md"),
],
exclude_patterns=["**/drafts/**"],
)
results = await kb.load_from_directory("./docs", config=config)
print(f"Loaded {results['total_chunks']} chunks from {results['total_files']} files")
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 | |
load_markdown_text
async
¶
load_markdown_text(
markdown_text: str, source: str, metadata: dict[str, Any] | None = None
) -> int
Load markdown content from a string.
Parses, chunks, embeds, and stores the markdown text. This is
the shared implementation used by :meth:load_markdown_document,
:meth:load_json_document, :meth:load_yaml_document, and
:meth:load_csv_document.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
markdown_text
|
str
|
Markdown content to load |
required |
source
|
str
|
Source identifier for metadata |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata to attach to all chunks |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of chunks created |
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 | |
query
async
¶
query(
query: str,
k: int = 5,
filter_metadata: dict[str, Any] | None = None,
min_similarity: float = 0.0,
merge_adjacent: bool = False,
max_chunk_size: int | None = None,
) -> list[dict[str, Any]]
Query knowledge base for relevant chunks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Query text to search for |
required |
k
|
int
|
Number of results to return |
5
|
filter_metadata
|
dict[str, Any] | None
|
Optional metadata filters |
None
|
min_similarity
|
float
|
Minimum similarity score (0-1) |
0.0
|
merge_adjacent
|
bool
|
Whether to merge adjacent chunks with same heading |
False
|
max_chunk_size
|
int | None
|
Maximum size for merged chunks (uses merger config default if not specified) |
None
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of result dictionaries with: - text: Chunk text - source: Source file - heading_path: Heading hierarchy - similarity: Similarity score - metadata: Full chunk metadata |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 | |
hybrid_query
async
¶
hybrid_query(
query: str,
k: int = 5,
text_weight: float = 0.5,
vector_weight: float = 0.5,
fusion_strategy: str = "rrf",
text_fields: list[str] | None = None,
filter_metadata: dict[str, Any] | None = None,
min_similarity: float = 0.0,
merge_adjacent: bool = False,
max_chunk_size: int | None = None,
) -> list[dict[str, Any]]
Query knowledge base using hybrid search (text + vector).
Combines keyword matching with semantic vector search for improved retrieval quality. Uses Reciprocal Rank Fusion (RRF) or weighted score fusion to combine results.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Query text to search for |
required |
k
|
int
|
Number of results to return |
5
|
text_weight
|
float
|
Weight for text search (0.0 to 1.0) |
0.5
|
vector_weight
|
float
|
Weight for vector search (0.0 to 1.0) |
0.5
|
fusion_strategy
|
str
|
Fusion method - "rrf" (default), "weighted_sum", or "native" |
'rrf'
|
text_fields
|
list[str] | None
|
Fields to search for text matching (default: ["text"]) |
None
|
filter_metadata
|
dict[str, Any] | None
|
Optional metadata filters |
None
|
min_similarity
|
float
|
Minimum combined score (0-1) |
0.0
|
merge_adjacent
|
bool
|
Whether to merge adjacent chunks with same heading |
False
|
max_chunk_size
|
int | None
|
Maximum size for merged chunks |
None
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of result dictionaries with: - text: Chunk text - source: Source file - heading_path: Heading hierarchy - similarity: Combined similarity score - text_score: Score from text search (if available) - vector_score: Score from vector search (if available) - metadata: Full chunk metadata |
Example
# Default RRF fusion
results = await kb.hybrid_query(
"How do I configure the database?",
k=5,
)
# Weighted toward vector search
results = await kb.hybrid_query(
"database configuration",
k=5,
text_weight=0.3,
vector_weight=0.7,
)
# Weighted sum fusion
results = await kb.hybrid_query(
"configure database",
k=5,
fusion_strategy="weighted_sum",
)
for result in results:
print(f"[{result['similarity']:.2f}] {result['heading_path']}")
print(f" text_score={result.get('text_score', 'N/A')}")
print(f" vector_score={result.get('vector_score', 'N/A')}")
print(result['text'])
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 | |
format_context ¶
Format search results for LLM context.
Convenience method to format results using the configured formatter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
results
|
list[dict[str, Any]]
|
Search results from query() |
required |
wrap_in_tags
|
bool
|
Whether to wrap in |
True
|
Returns:
| Type | Description |
|---|---|
str
|
Formatted context string |
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
count
async
¶
Get the number of chunks in the knowledge base.
Delegates to the underlying vector store's count method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filter
|
dict[str, Any] | None
|
Optional metadata filter to count only matching chunks |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of chunks stored (optionally filtered) |
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
clear
async
¶
Clear all documents from the knowledge base.
Warning: This removes all stored chunks and embeddings.
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
save
async
¶
Save the knowledge base to persistent storage.
This persists the vector store index and metadata to disk. Only applicable for vector stores that support persistence (e.g., FAISS).
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
providers ¶
Return the embedding provider, keyed by role.
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
set_provider ¶
Replace the embedding provider if the role matches.
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
close
async
¶
Close the knowledge base and release resources.
This method: - Saves the vector store to disk (if persistence is configured) - Closes the vector store connection - Closes the embedding provider (releases HTTP sessions)
Should be called when done using the knowledge base to prevent resource leaks (e.g., unclosed aiohttp sessions).
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
__aenter__
async
¶
Async context manager entry.
Returns:
| Type | Description |
|---|---|
Self
|
Self for use in async with statement |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
__aexit__
async
¶
__aexit__(
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None
Async context manager exit - ensures cleanup.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exc_type
|
type[BaseException] | None
|
Exception type if an exception occurred |
required |
exc_val
|
BaseException | None
|
Exception value if an exception occurred |
required |
exc_tb
|
TracebackType | None
|
Exception traceback if an exception occurred |
required |
Source code in packages/bots/src/dataknobs_bots/knowledge/rag.py
BufferMemory ¶
Bases: Memory
Simple buffer memory keeping last N messages.
This implementation uses a fixed-size buffer that keeps the most recent messages in memory. When the buffer is full, the oldest messages are automatically removed.
Attributes:
| Name | Type | Description |
|---|---|---|
max_messages |
Maximum number of messages to keep in buffer |
|
messages |
deque[dict[str, Any]]
|
Deque containing the messages |
Initialize buffer memory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_messages
|
int
|
Maximum number of messages to keep |
10
|
Methods:
| Name | Description |
|---|---|
add_message |
Add message to buffer. |
get_context |
Get all messages in buffer. |
clear |
Clear all messages from buffer. |
pop_messages |
Remove and return the last N messages from the buffer. |
Source code in packages/bots/src/dataknobs_bots/memory/buffer.py
Functions¶
add_message
async
¶
Add message to buffer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content
|
str
|
Message content |
required |
role
|
str
|
Message role |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata |
None
|
Source code in packages/bots/src/dataknobs_bots/memory/buffer.py
get_context
async
¶
Get all messages in buffer.
The current_message parameter is not used in buffer memory since we simply return all buffered messages in order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_message
|
str
|
Not used in buffer memory |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of all buffered messages |
Source code in packages/bots/src/dataknobs_bots/memory/buffer.py
clear
async
¶
pop_messages
async
¶
Remove and return the last N messages from the buffer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
count
|
int
|
Number of messages to remove from the end. |
2
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
The removed messages in the order they were stored. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If count exceeds available messages or is < 1. |
Source code in packages/bots/src/dataknobs_bots/memory/buffer.py
CompositeMemory ¶
Bases: Memory
Combines multiple memory strategies into one.
All sub-strategies receive every add_message() call independently.
On get_context(), the primary strategy's results appear first,
followed by deduplicated results from secondary strategies.
Graceful degradation: if any strategy fails on a read or write, the composite logs a warning and continues with the remaining strategies.
Attributes:
| Name | Type | Description |
|---|---|---|
primary |
Memory
|
The primary memory strategy (results appear first). |
strategies |
list[Memory]
|
All sub-strategies in order. |
Initialize composite memory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
strategies
|
list[Memory]
|
List of memory strategy instances. |
required |
primary_index
|
int
|
Index of the primary strategy in the list. |
0
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If strategies is empty or primary_index is out of range. |
Methods:
| Name | Description |
|---|---|
add_message |
Forward message to all strategies. |
get_context |
Collect context from all strategies, primary first. |
clear |
Clear all strategies. Log and continue on individual failures. |
pop_messages |
Delegate to primary strategy only. |
close |
Close all strategies. Log and continue on individual failures. |
providers |
Aggregate providers from all sub-strategies. |
set_provider |
Forward to all sub-strategies; return True if any accepted. |
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
Attributes¶
Functions¶
add_message
async
¶
Forward message to all strategies.
If a strategy raises, the error is logged and remaining strategies still receive the message.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
get_context
async
¶
Collect context from all strategies, primary first.
Results from the primary strategy appear first. All results are
deduplicated by (role, content) — if a message with the same
role and content already appeared, it is not repeated.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
clear
async
¶
Clear all strategies. Log and continue on individual failures.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
pop_messages
async
¶
Delegate to primary strategy only.
Secondary strategies (especially vector) may not support undo. If the primary doesn't support it, NotImplementedError propagates.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
close
async
¶
Close all strategies. Log and continue on individual failures.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
providers ¶
Aggregate providers from all sub-strategies.
If multiple strategies expose the same role, the last one wins and a warning is logged.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
set_provider ¶
Forward to all sub-strategies; return True if any accepted.
Source code in packages/bots/src/dataknobs_bots/memory/composite.py
Memory ¶
Bases: ABC
Abstract base class for memory implementations.
Methods:
| Name | Description |
|---|---|
add_message |
Add message to memory. |
get_context |
Get relevant context for current message. |
clear |
Clear all memory. |
providers |
Return LLM providers managed by this memory, keyed by role. |
set_provider |
Replace a provider managed by this memory. |
close |
Release resources held by this memory implementation. |
pop_messages |
Remove and return the last N messages from memory. |
Functions¶
add_message
abstractmethod
async
¶
Add message to memory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content
|
str
|
Message content |
required |
role
|
str
|
Message role (user, assistant, system, etc.) |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata for the message |
None
|
Source code in packages/bots/src/dataknobs_bots/memory/base.py
get_context
abstractmethod
async
¶
Get relevant context for current message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_message
|
str
|
The current message to get context for |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of relevant message dictionaries |
Source code in packages/bots/src/dataknobs_bots/memory/base.py
clear
abstractmethod
async
¶
providers ¶
Return LLM providers managed by this memory, keyed by role.
Subsystems declare the providers they own so that the bot can register them in the provider catalog without reaching into private attributes. The default returns an empty dict (no providers).
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict mapping provider role names to provider instances. |
Source code in packages/bots/src/dataknobs_bots/memory/base.py
set_provider ¶
Replace a provider managed by this memory.
Called by inject_providers to wire a test provider into the
actual subsystem, not just the registry catalog. The default
returns False (role not recognized). Concrete subclasses
override to accept their known roles.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
role
|
str
|
Provider role name (e.g. |
required |
provider
|
Any
|
Replacement provider instance. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
bool
|
|
Source code in packages/bots/src/dataknobs_bots/memory/base.py
close
async
¶
Release resources held by this memory implementation.
The default is a no-op. Subclasses that create providers or open
connections (e.g. VectorMemory, SummaryMemory) should
override to clean up.
Source code in packages/bots/src/dataknobs_bots/memory/base.py
pop_messages
async
¶
Remove and return the last N messages from memory.
Used for conversation undo. The count is determined by the caller based on node depth difference (not a fixed 2).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
count
|
int
|
Number of messages to remove from the end. |
2
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
The removed messages in the order they were stored. |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the implementation does not support undo. |
Source code in packages/bots/src/dataknobs_bots/memory/base.py
SummaryMemory ¶
SummaryMemory(
llm_provider: AsyncLLMProvider,
recent_window: int = 10,
summary_prompt: str | None = None,
*,
owns_llm_provider: bool = False,
)
Bases: Memory
Memory that summarizes older messages to maintain long context windows.
Maintains a rolling buffer of recent messages. When the buffer exceeds a configurable threshold, the oldest messages are compressed into a running summary using the LLM provider. This trades exact message recall for a much longer effective context window.
get_context() returns the summary (if any) as a system message,
followed by the recent verbatim messages.
Attributes:
| Name | Type | Description |
|---|---|---|
llm_provider |
LLM provider used for generating summaries |
|
recent_window |
Number of recent messages to keep verbatim |
|
summary_prompt |
Template for the summarization prompt |
Initialize summary memory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
llm_provider
|
AsyncLLMProvider
|
Async LLM provider for generating summaries |
required |
recent_window
|
int
|
Number of recent messages to keep verbatim.
When the buffer has more than |
10
|
summary_prompt
|
str | None
|
Custom summarization prompt template. Must
contain |
None
|
owns_llm_provider
|
bool
|
Whether this instance owns the provider's lifecycle. True when a dedicated provider was created for this memory; False when reusing the bot's main LLM. |
False
|
Methods:
| Name | Description |
|---|---|
add_message |
Add a message and trigger summarization if the buffer is full. |
get_context |
Return the running summary followed by recent messages. |
providers |
Return the summary LLM provider for catalog registration. |
set_provider |
Replace the summary LLM provider if the role matches. |
close |
Close the LLM provider if this instance owns it. |
clear |
Clear both the running summary and the message buffer. |
pop_messages |
Remove and return the last N messages from the recent window. |
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
Functions¶
add_message
async
¶
Add a message and trigger summarization if the buffer is full.
When the number of buffered messages exceeds recent_window,
the oldest messages are summarized into the running summary using
the LLM provider. On LLM failure, older messages are dropped to
keep the buffer within bounds (graceful degradation).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content
|
str
|
Message content |
required |
role
|
str
|
Message role (user, assistant, system) |
required |
metadata
|
dict[str, Any] | None
|
Optional metadata for the message |
None
|
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
get_context
async
¶
Return the running summary followed by recent messages.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_message
|
str
|
The current message (not used by summary memory, kept for interface compatibility) |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of message dicts. If a summary exists it is the first |
list[dict[str, Any]]
|
element with |
list[dict[str, Any]]
|
the recent verbatim messages. |
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
providers ¶
Return the summary LLM provider for catalog registration.
Always reports the provider for discovery and observability.
The _owns_llm_provider flag controls lifecycle (close()),
not visibility — consistent with VectorMemory, RAGKnowledgeBase,
and WizardReasoning.
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
set_provider ¶
Replace the summary LLM provider if the role matches.
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
close
async
¶
Close the LLM provider if this instance owns it.
When a dedicated provider was created for this memory (via the
llm config key), this instance owns its lifecycle. When the
bot's main LLM was passed in as a fallback, the bot owns it.
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
clear
async
¶
pop_messages
async
¶
Remove and return the last N messages from the recent window.
Only messages still in the recent buffer can be popped. Messages that have already been summarized are irreversibly compressed and cannot be individually removed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
count
|
int
|
Number of messages to remove from the end. |
2
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
The removed messages in the order they were stored. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If count exceeds available (unsummarized) messages or is < 1. |
Source code in packages/bots/src/dataknobs_bots/memory/summary.py
VectorMemory ¶
VectorMemory(
vector_store: Any,
embedding_provider: Any,
max_results: int = 5,
similarity_threshold: float = 0.7,
default_metadata: dict[str, Any] | None = None,
default_filter: dict[str, Any] | None = None,
owns_embedding_provider: bool = False,
owns_vector_store: bool = False,
)
Bases: Memory
Vector-based semantic memory using dataknobs-data vector stores.
This implementation stores messages with vector embeddings and retrieves relevant messages based on semantic similarity.
Attributes:
| Name | Type | Description |
|---|---|---|
vector_store |
Vector store backend from dataknobs_data.vector.stores |
|
embedding_provider |
LLM provider for generating embeddings |
|
max_results |
Maximum number of results to return |
|
similarity_threshold |
Minimum similarity score for results |
Initialize vector memory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
vector_store
|
Any
|
Vector store backend instance |
required |
embedding_provider
|
Any
|
LLM provider with embed() method |
required |
max_results
|
int
|
Maximum number of similar messages to return |
5
|
similarity_threshold
|
float
|
Minimum similarity score (0-1) |
0.7
|
default_metadata
|
dict[str, Any] | None
|
Metadata merged into every |
None
|
default_filter
|
dict[str, Any] | None
|
Filter merged into every |
None
|
owns_embedding_provider
|
bool
|
If True, |
False
|
owns_vector_store
|
bool
|
If True, |
False
|
Methods:
| Name | Description |
|---|---|
from_config |
Create VectorMemory from configuration. |
add_message |
Add message with vector embedding. |
get_context |
Get semantically relevant messages. |
providers |
Return the embedding provider, keyed by role. |
set_provider |
Replace the embedding provider if the role matches. |
close |
Close owned resources. |
clear |
Clear all vectors from memory. |
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
Functions¶
from_config
async
classmethod
¶
Create VectorMemory from configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dictionary with:
- backend: Vector store backend type
- dimension: Vector store dimension (singular; default 1536)
- collection: Collection/index name (optional)
- embedding: Nested embedding config dict (preferred), e.g.
|
required |
Returns:
| Type | Description |
|---|---|
VectorMemory
|
Configured VectorMemory instance |
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
add_message
async
¶
Add message with vector embedding.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content
|
str
|
Message content |
required |
role
|
str
|
Message role |
required |
metadata
|
dict[str, Any] | None
|
Optional caller-supplied metadata. Merged after
|
None
|
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
get_context
async
¶
Get semantically relevant messages.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_message
|
str
|
Current message to find context for |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of relevant message dictionaries sorted by similarity |
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
providers ¶
Return the embedding provider, keyed by role.
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
set_provider ¶
Replace the embedding provider if the role matches.
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
close
async
¶
Close owned resources.
Only closes resources that this instance owns (created in
from_config). Externally-injected resources are left open
for the caller to manage.
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
clear
async
¶
Clear all vectors from memory.
Delegates to the vector store's clear() method. Use with caution
if the store is shared across multiple memory instances.
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the backing vector store does not
support |
Source code in packages/bots/src/dataknobs_bots/memory/vector.py
CostTrackingMiddleware ¶
Bases: Middleware
Middleware for tracking LLM API costs and usage.
Monitors token usage across different providers (Ollama, OpenAI, Anthropic, etc.) to help optimize costs and track budgets.
Attributes:
| Name | Type | Description |
|---|---|---|
track_tokens |
Whether to track token usage |
|
cost_rates |
Token cost rates per provider/model |
|
usage_stats |
Accumulated usage statistics by client_id |
Example
# Create middleware with default rates
middleware = CostTrackingMiddleware()
# Or with custom rates
middleware = CostTrackingMiddleware(
cost_rates={
"openai": {
"gpt-4o": {"input": 0.0025, "output": 0.01},
},
}
)
# Get stats
stats = middleware.get_client_stats("my-client")
total = middleware.get_total_cost()
# Export to JSON
json_data = middleware.export_stats_json()
Initialize cost tracking middleware.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
track_tokens
|
bool
|
Enable token tracking |
True
|
cost_rates
|
dict[str, Any] | None
|
Optional custom cost rates (merged with defaults) |
None
|
Methods:
| Name | Description |
|---|---|
on_turn_start |
Log estimated input tokens at the start of a turn. |
after_turn |
Track costs after turn completion using TurnState data. |
on_error |
Log errors but don't track costs for failed requests. |
on_hook_error |
Track middleware hook failures. |
get_client_stats |
Get usage statistics for a client. |
get_all_stats |
Get all usage statistics. |
get_total_cost |
Get total cost across all clients. |
get_total_tokens |
Get total tokens across all clients. |
clear_stats |
Clear usage statistics. |
export_stats_json |
Export all statistics as JSON. |
export_stats_csv |
Export statistics as CSV (one row per client). |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
Functions¶
on_turn_start
async
¶
Log estimated input tokens at the start of a turn.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Turn state at the start of the pipeline. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
None (no message transform). |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
after_turn
async
¶
Track costs after turn completion using TurnState data.
Uses real token usage when the provider reports it, otherwise estimates from message/response text length (~4 chars per token).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Completed turn state with usage and response data. |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
on_error
async
¶
Log errors but don't track costs for failed requests.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
error
|
Exception
|
The exception that occurred |
required |
message
|
str
|
User message that caused the error |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
on_hook_error
async
¶
Track middleware hook failures.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hook_name
|
str
|
Name of the hook that failed |
required |
error
|
Exception
|
The exception raised by the middleware hook |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
get_client_stats ¶
Get usage statistics for a client.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client_id
|
str
|
Client identifier |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | None
|
Usage statistics or None if not found |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
get_all_stats ¶
Get all usage statistics.
Returns:
| Type | Description |
|---|---|
dict[str, dict[str, Any]]
|
Dictionary mapping client_id to statistics |
get_total_cost ¶
Get total cost across all clients.
Returns:
| Type | Description |
|---|---|
float
|
Total cost in USD |
get_total_tokens ¶
Get total tokens across all clients.
Returns:
| Type | Description |
|---|---|
dict[str, int]
|
Dictionary with 'input', 'output', and 'total' token counts |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
clear_stats ¶
Clear usage statistics.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client_id
|
str | None
|
If provided, clear only this client. Otherwise clear all. |
None
|
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
export_stats_json ¶
Export all statistics as JSON.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
indent
|
int
|
JSON indentation level |
2
|
Returns:
| Type | Description |
|---|---|
str
|
JSON string of all statistics |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
export_stats_csv ¶
Export statistics as CSV (one row per client).
Returns:
| Type | Description |
|---|---|
str
|
CSV string with headers |
Source code in packages/bots/src/dataknobs_bots/middleware/cost.py
LoggingMiddleware ¶
LoggingMiddleware(
log_level: str = "INFO",
include_metadata: bool = True,
json_format: bool = False,
)
Bases: Middleware
Middleware for tracking conversation interactions.
Logs all user messages and bot responses with context for monitoring, debugging, and analytics.
Uses the unified TurnState hooks:
on_turn_start— logs incoming user messageafter_turn— logs turn completion with response, usage, tools
Attributes:
| Name | Type | Description |
|---|---|---|
log_level |
Logging level to use (default: INFO) |
|
include_metadata |
Whether to include full context metadata |
|
json_format |
Whether to output logs in JSON format |
Example
Initialize logging middleware.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
log_level
|
str
|
Logging level (DEBUG, INFO, WARNING, ERROR) |
'INFO'
|
include_metadata
|
bool
|
Whether to log full context metadata |
True
|
json_format
|
bool
|
Whether to output in JSON format |
False
|
Methods:
| Name | Description |
|---|---|
on_turn_start |
Log incoming user message at the start of a turn. |
after_turn |
Log turn completion with unified data for all turn types. |
on_error |
Called when an error occurs during message processing. |
on_hook_error |
Called when a middleware hook itself raises. |
Source code in packages/bots/src/dataknobs_bots/middleware/logging.py
Functions¶
on_turn_start
async
¶
Log incoming user message at the start of a turn.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Turn state at the start of the pipeline. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
None (no message transform). |
Source code in packages/bots/src/dataknobs_bots/middleware/logging.py
after_turn
async
¶
Log turn completion with unified data for all turn types.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Completed turn state. |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/logging.py
on_error
async
¶
Called when an error occurs during message processing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
error
|
Exception
|
The exception that occurred |
required |
message
|
str
|
User message that caused the error |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/logging.py
on_hook_error
async
¶
Called when a middleware hook itself raises.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hook_name
|
str
|
Name of the hook that failed |
required |
error
|
Exception
|
The exception raised by the middleware hook |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/logging.py
Middleware ¶
Base class for bot middleware.
Middleware provides hooks into the bot request/response lifecycle. All hooks are concrete no-ops — subclasses override only the hooks they need.
Preferred hooks (receive full TurnState):
on_turn_start(turn)— before processing; can writeplugin_dataand optionally transform the message.after_turn(turn)— after any turn completes (chat, stream, greet); unified successor toafter_messageandpost_stream.finally_turn(turn)— fires after every turn on both success and error paths. Use for resource cleanup. Forstream_chat, requires full consumption oraclosing().on_tool_executed(execution, context)— after each tool call.
Legacy hooks (kept for backward compatibility):
before_message(message, context)— useon_turn_startinstead.after_message(response, context, **kwargs)— useafter_turninstead.post_stream(message, response, context)— useafter_turninstead.
Error hooks (no TurnState equivalent — still primary):
on_error(error, message, context)— request failed.on_hook_error(hook_name, error, context)— a hook failed.
Error semantics
on_error fires when the bot request fails — the caller does NOT
receive a response. on_hook_error fires when a middleware's own
hook raises after the request already succeeded — the caller DID
receive a response, but a middleware could not complete its
post-processing.
Example
class MyMiddleware(Middleware):
async def on_turn_start(self, turn):
turn.plugin_data["started"] = True
return None # or return transformed message
async def after_turn(self, turn):
log.info("Turn %s done", turn.mode.value)
async def on_error(self, error, message, context):
log.error("Request failed: %s", error)
Methods:
| Name | Description |
|---|---|
before_message |
Called before processing user message. |
after_message |
Called after generating bot response (non-streaming). |
post_stream |
Called after streaming response completes. |
on_error |
Called when a request-level error occurs during message processing. |
on_hook_error |
Called when a middleware hook itself raises an exception. |
on_turn_start |
Called at the start of every turn, before message processing. |
after_turn |
Called after any turn completes (chat, stream, or greet). |
finally_turn |
Called after every turn, on both success and error paths. |
on_tool_executed |
Called after each tool execution within a turn. |
Functions¶
before_message
async
¶
Called before processing user message.
.. deprecated::
Use on_turn_start instead, which provides the full
TurnState including plugin_data for cross-middleware
communication and supports message transforms.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
User's input message |
required |
context
|
BotContext
|
Bot context with conversation and user info |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
after_message
async
¶
Called after generating bot response (non-streaming).
.. deprecated::
Use after_turn instead, which fires for all turn types
(chat, stream, greet) and provides the full TurnState
with usage data, tool executions, and plugin data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
str
|
Bot's generated response |
required |
context
|
BotContext
|
Bot context |
required |
**kwargs
|
Any
|
Additional data (e.g., tokens_used, response_time_ms, provider, model) |
{}
|
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
post_stream
async
¶
Called after streaming response completes.
.. deprecated::
Use after_turn instead, which fires for all turn types
and provides real token usage data from the provider.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Original user message that triggered the stream |
required |
response
|
str
|
Complete accumulated response from streaming |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
on_error
async
¶
Called when a request-level error occurs during message processing.
This hook fires when the bot request fails (preparation, generation,
or memory/middleware post-processing in before_message). The
caller does NOT receive a response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
error
|
Exception
|
The exception that occurred |
required |
message
|
str
|
User message that caused the error |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
on_hook_error
async
¶
Called when a middleware hook itself raises an exception.
This fires when a post-generation middleware hook raises:
after_turn, finally_turn, on_tool_executed,
after_message, post_stream, or on_error. On the
success path, the response was already delivered; on the error
path (e.g. finally_turn after a failed turn), it was not.
In either case, the middleware could not complete its own
post-processing (e.g., a logging sink was unreachable, a
metrics backend timed out).
Note: on_turn_start exceptions are NOT routed here —
they are re-raised to abort the request (matching
before_message semantics), so on_error fires instead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hook_name
|
str
|
Name of the hook that failed (e.g.
|
required |
error
|
Exception
|
The exception raised by the middleware hook |
required |
context
|
BotContext
|
Bot context |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
on_turn_start
async
¶
Called at the start of every turn, before message processing.
Receives the full TurnState including plugin_data for
cross-middleware communication. Middleware can:
- Write to
turn.plugin_datato share data with downstream pipeline participants (LLM middleware, tools,after_turn). - Return a transformed message string to replace
turn.messagebefore it reaches the LLM (e.g., PII stripping, attack sanitization). Transforms chain: each middleware receives the message as modified by the previous one. - Return
Noneto leave the message unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Turn state at the start of the pipeline. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
Transformed message string, or |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
after_turn
async
¶
Called after any turn completes (chat, stream, or greet).
Provides the full TurnState with usage data, tool executions,
and response content regardless of how the turn was initiated.
This is the unified successor to after_message and
post_stream — implement this for uniform post-turn handling.
The legacy hooks (after_message / post_stream) continue
to fire as well, so existing middleware is unaffected.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Complete turn state with all pipeline data. |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
finally_turn
async
¶
Called after every turn, on both success and error paths.
Use this for resource cleanup (closing DB sessions, releasing
locks, flushing buffers) that must happen regardless of outcome.
after_turn is conditional — it does not fire on error paths
or when greet() returns None. Do not assume
after_turn has already run when writing finally_turn
logic.
For chat() and greet(), this hook fires reliably via a
finally block. For stream_chat() (an async generator),
the finally block fires only when the generator is fully
consumed, explicitly closed (aclose()), or garbage
collected. Callers that break out of the stream early should
use contextlib.aclosing to guarantee prompt cleanup.
plugin_data populated by on_turn_start (or seeded from
the call site via the plugin_data parameter on chat() /
stream_chat() / greet()) is available here.
This is an observational hook — failures are logged and reported
via on_hook_error but do not prevent other middleware from
running.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
turn
|
TurnState
|
Turn state at the end of the pipeline. On error paths
or no-strategy |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
on_tool_executed
async
¶
Called after each tool execution within a turn.
Fired once per tool invocation, before after_turn. All
on_tool_executed calls happen post-turn during
_finalize_turn(), not in real-time as tools execute — this
hook is for auditing and logging, not for aborting or
rate-limiting mid-turn.
Ordering note: DynaBot-level tool executions appear first, followed by strategy-level executions (e.g. ReAct). In practice only one source produces executions per turn.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution
|
ToolExecution
|
Record of the tool execution (name, params, result, error, duration). |
required |
context
|
BotContext
|
Bot context for the current turn. |
required |
Source code in packages/bots/src/dataknobs_bots/middleware/base.py
ReActReasoning ¶
ReActReasoning(
max_iterations: int = 5,
verbose: bool = False,
store_trace: bool = False,
artifact_registry: Any | None = None,
review_executor: Any | None = None,
context_builder: Any | None = None,
extra_context: dict[str, Any] | None = None,
prompt_refresher: Callable[[], str] | None = None,
greeting_template: str | None = None,
)
Bases: ReasoningStrategy
ReAct (Reasoning + Acting) strategy.
This strategy implements the ReAct pattern where the LLM: 1. Reasons about what to do (Thought) 2. Takes an action (using tools if needed) 3. Observes the result 4. Repeats until task is complete
This is useful for: - Multi-step problem solving - Tasks requiring tool use - Complex reasoning chains
Attributes:
| Name | Type | Description |
|---|---|---|
max_iterations |
Maximum number of reasoning loops |
|
verbose |
Whether to enable debug-level logging |
|
store_trace |
Whether to store reasoning trace in conversation metadata |
Example
Initialize ReAct reasoning strategy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_iterations
|
int
|
Maximum reasoning/action iterations |
5
|
verbose
|
bool
|
Enable debug-level logging for reasoning steps |
False
|
store_trace
|
bool
|
Store reasoning trace in conversation metadata |
False
|
artifact_registry
|
Any | None
|
Optional ArtifactRegistry for artifact management |
None
|
review_executor
|
Any | None
|
Optional ReviewExecutor for running reviews |
None
|
context_builder
|
Any | None
|
Optional ContextBuilder for building conversation context |
None
|
extra_context
|
dict[str, Any] | None
|
Optional extra key-value pairs to merge into the ToolExecutionContext for every tool call (e.g. banks, custom state) |
None
|
prompt_refresher
|
Callable[[], str] | None
|
Optional callback that returns a fresh system
prompt string. Called after tool execution in each iteration
to update |
None
|
greeting_template
|
str | None
|
Optional Jinja2 template for bot-initiated greetings (inherited from ReasoningStrategy). |
None
|
Methods:
| Name | Description |
|---|---|
from_config |
Create ReActReasoning from a configuration dict. |
generate |
Generate response using ReAct loop. |
Source code in packages/bots/src/dataknobs_bots/reasoning/react.py
Attributes¶
Functions¶
from_config
classmethod
¶
Create ReActReasoning from a configuration dict.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Configuration dict with optional keys: max_iterations, verbose, store_trace, greeting_template. |
required |
**_kwargs
|
Any
|
Ignored (no KB or provider injection needed). |
{}
|
Returns:
| Type | Description |
|---|---|
ReActReasoning
|
Configured ReActReasoning instance. |
Source code in packages/bots/src/dataknobs_bots/reasoning/react.py
generate
async
¶
Generate response using ReAct loop.
The ReAct loop: 1. Generate response (may include tool calls) 2. If tool calls present, execute them 3. Add observations to conversation 4. Repeat until no more tool calls or max iterations
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
manager
|
Any
|
ConversationManager instance |
required |
llm
|
Any
|
LLM provider instance |
required |
tools
|
list[Any] | None
|
Optional list of available tools |
None
|
**kwargs
|
Any
|
Generation parameters |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
Final LLM response |
Source code in packages/bots/src/dataknobs_bots/reasoning/react.py
131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 | |
ReasoningStrategy ¶
Bases: ABC
Abstract base class for reasoning strategies.
Reasoning strategies control how the bot processes information and generates responses. Different strategies can implement different levels of reasoning complexity.
All strategies support an optional greeting_template — a Jinja2
template string rendered with initial_context variables when
greet() is called. Strategies that need richer greeting behavior
(e.g. WizardReasoning with FSM-driven stage responses) override
greet() entirely.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
greeting_template
|
str | None
|
Optional Jinja2 template for bot-initiated
greetings. Variables from |
None
|
Examples:
- Simple: Direct LLM call
- Chain-of-Thought: Break down reasoning into steps
- ReAct: Reason and act in a loop with tools
Methods:
| Name | Description |
|---|---|
capabilities |
Declare what this strategy manages autonomously. |
from_config |
Create a strategy instance from a configuration dict. |
get_source_configs |
Extract source configuration dicts from a strategy config. |
get_and_clear_tool_executions |
Return tool executions recorded during the last generate() call. |
greet |
Generate an initial bot greeting before the user speaks. |
add_source |
Add a retrieval source to this strategy. |
providers |
Return LLM providers managed by this strategy, keyed by role. |
set_provider |
Replace a provider managed by this strategy. |
close |
Release resources held by this strategy. |
generate |
Generate response using this reasoning strategy. |
stream_generate |
Stream response using this reasoning strategy. |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
Functions¶
capabilities
classmethod
¶
Declare what this strategy manages autonomously.
The default returns no capabilities. Concrete strategies override to declare their actual capabilities.
Returns:
| Type | Description |
|---|---|
StrategyCapabilities
|
Frozen dataclass describing strategy capabilities. |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
from_config
classmethod
¶
Create a strategy instance from a configuration dict.
The base implementation extracts greeting_template and
passes it to the constructor. Concrete strategies with richer
configuration override this classmethod.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Strategy configuration dict. |
required |
**_kwargs
|
Any
|
Additional context (e.g. |
{}
|
Returns:
| Type | Description |
|---|---|
Self
|
Configured strategy instance. |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
get_source_configs
classmethod
¶
Extract source configuration dicts from a strategy config.
DynaBot calls this after creating the strategy to discover
which sources to construct and wire in via :meth:add_source.
The default looks for a top-level "sources" key, which is
the convention used by GroundedReasoning.
Strategies with non-standard config layouts (e.g.
HybridReasoning, where sources are nested under
"grounded") override this to return the correct list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
The full strategy configuration dict. |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
List of source configuration dicts (may be empty). |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
get_and_clear_tool_executions ¶
Return tool executions recorded during the last generate() call.
Strategies that execute tools (e.g. ReAct) append
ToolExecution records to self._tool_executions during
their generation loop. DynaBot calls this after
generate() returns to collect the records and fire
on_tool_executed middleware hooks.
Returns:
| Type | Description |
|---|---|
list[ToolExecution]
|
List of tool execution records (cleared after retrieval). |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
greet
async
¶
greet(
manager: ReasoningManagerProtocol,
llm: Any,
*,
initial_context: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any | None
Generate an initial bot greeting before the user speaks.
The default implementation renders greeting_template (if set)
with initial_context variables using Jinja2 and returns the
result as an LLMResponse. Returns None when no template
is configured.
WizardReasoning fully overrides this with FSM-driven greeting
generation from the wizard's start stage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
manager
|
ReasoningManagerProtocol
|
ConversationManager or compatible manager instance |
required |
llm
|
Any
|
LLM provider instance |
required |
initial_context
|
dict[str, Any] | None
|
Optional dict of data available as Jinja2
template variables (e.g. |
None
|
**kwargs
|
Any
|
Additional generation parameters |
{}
|
Returns:
| Type | Description |
|---|---|
Any | None
|
LLMResponse if a greeting was generated, or None |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
add_source ¶
Add a retrieval source to this strategy.
Strategies that declare manages_sources=True in their
:meth:capabilities MUST override this method. DynaBot
calls it during config-driven source construction.
The default raises NotImplementedError so that a 3rd-party
strategy that forgets to implement it fails loudly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
source
|
Any
|
A |
required |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If not overridden by a subclass. |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
providers ¶
Return LLM providers managed by this strategy, keyed by role.
Subsystems declare the providers they own so that the bot can register them in the provider catalog without reaching into private attributes. The default returns an empty dict (no providers).
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict mapping provider role names to provider instances. |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
set_provider ¶
Replace a provider managed by this strategy.
Called by inject_providers to wire a test provider into the
actual subsystem, not just the registry catalog. The default
returns False (role not recognized). Concrete subclasses
override to accept their known roles.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
role
|
str
|
Provider role name (e.g. |
required |
provider
|
Any
|
Replacement provider instance. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
bool
|
|
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
close
async
¶
Release resources held by this strategy.
Default no-op. Subclasses that hold resources (LLM providers,
database connections, asyncio tasks) should override to release
them. Called by DynaBot.close().
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
generate
abstractmethod
async
¶
generate(
manager: ReasoningManagerProtocol,
llm: Any,
tools: list[Any] | None = None,
**kwargs: Any,
) -> Any
Generate response using this reasoning strategy.
Pass tools through to manager.complete(tools=tools) so
the LLM can see available tools. If the returned response
contains tool_calls, DynaBot will execute them
automatically in a post-strategy loop — strategies do not need
to handle tool execution unless they want full control (like
ReActReasoning).
Strategies that execute tools internally should record them via
self._tool_executions.append(ToolExecution(...)) and
consume tool_calls before returning, so the DynaBot loop
is a no-op.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
manager
|
ReasoningManagerProtocol
|
ConversationManager or compatible manager instance |
required |
llm
|
Any
|
LLM provider instance |
required |
tools
|
list[Any] | None
|
Optional list of available tools. Forward to
|
None
|
**kwargs
|
Any
|
Additional generation parameters (temperature, max_tokens, etc.) |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
LLM response object |
Example
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
stream_generate
async
¶
stream_generate(
manager: ReasoningManagerProtocol,
llm: Any,
tools: list[Any] | None = None,
**kwargs: Any,
) -> AsyncIterator[Any]
Stream response using this reasoning strategy.
The default implementation wraps generate() and yields the
complete response as a single item. Subclasses that support true
token-level streaming (e.g. SimpleReasoning) should override
this to yield incremental chunks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
manager
|
ReasoningManagerProtocol
|
ConversationManager or compatible manager instance |
required |
llm
|
Any
|
LLM provider instance |
required |
tools
|
list[Any] | None
|
Optional list of available tools |
None
|
**kwargs
|
Any
|
Additional generation parameters |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[Any]
|
LLM response or stream chunk objects |
Source code in packages/bots/src/dataknobs_bots/reasoning/base.py
SimpleReasoning ¶
Bases: ReasoningStrategy
Simple reasoning strategy that makes direct LLM calls.
This is the most straightforward strategy - it simply passes the conversation to the LLM and returns the response without any additional reasoning steps.
Use this when: - You want direct, fast responses - The task doesn't require complex reasoning - You're using a powerful model that doesn't need guidance
Example
Methods:
| Name | Description |
|---|---|
generate |
Generate response with a simple LLM call. |
stream_generate |
Stream response with true token-level streaming. |
Source code in packages/bots/src/dataknobs_bots/reasoning/simple.py
Functions¶
generate
async
¶
Generate response with a simple LLM call.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
manager
|
Any
|
ConversationManager instance |
required |
llm
|
Any
|
LLM provider instance (not used directly) |
required |
tools
|
list[Any] | None
|
Optional list of tools |
None
|
**kwargs
|
Any
|
Generation parameters |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
LLM response |
Source code in packages/bots/src/dataknobs_bots/reasoning/simple.py
stream_generate
async
¶
stream_generate(
manager: Any, llm: Any, tools: list[Any] | None = None, **kwargs: Any
) -> AsyncIterator[Any]
Stream response with true token-level streaming.
Delegates to manager.stream_complete() which yields
LLMStreamResponse chunks as they arrive from the provider.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
manager
|
Any
|
ConversationManager instance |
required |
llm
|
Any
|
LLM provider instance (not used directly) |
required |
tools
|
list[Any] | None
|
Optional list of tools |
None
|
**kwargs
|
Any
|
Generation parameters |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[Any]
|
LLM stream response chunks |
Source code in packages/bots/src/dataknobs_bots/reasoning/simple.py
StrategyCapabilities
dataclass
¶
Declares what a reasoning strategy manages autonomously.
Used by DynaBot and other consumers to decide which orchestration
steps to perform (e.g. source construction, auto-context) without
hard-coding strategy names.
All fields default to False; concrete strategies override only
the capabilities they possess. New fields can be added with
default=False without breaking existing strategies.
Attributes:
| Name | Type | Description |
|---|---|---|
manages_sources |
bool
|
Strategy manages its own retrieval sources
(grounded/hybrid). When |
StrategyRegistry ¶
Registry mapping strategy names to their factories.
Unlike PluginRegistry (which caches singleton instances and has
a (key, config) factory signature), StrategyRegistry creates
a fresh instance per call — strategies are per-bot, not singletons.
When PluginRegistry gains a create() method for fresh-instance
factory invocation (consumer-gaps plan Item 65), this class should
be migrated to use it as its backing store.
Methods:
| Name | Description |
|---|---|
register |
Register a strategy factory under the given name. |
create |
Create a strategy instance from a config dict. |
get_factory |
Return the factory for a strategy name, or |
is_registered |
Check whether a strategy name is registered. |
list_keys |
Return sorted list of registered strategy names. |
Source code in packages/bots/src/dataknobs_bots/reasoning/registry.py
Functions¶
register ¶
Register a strategy factory under the given name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Strategy name (used in config |
required |
factory
|
StrategyFactory
|
A |
required |
override
|
bool
|
If |
False
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in packages/bots/src/dataknobs_bots/reasoning/registry.py
create ¶
Create a strategy instance from a config dict.
Extracts config["strategy"] (default "simple"), looks up
the factory, and calls it. For ReasoningStrategy subclasses
the factory is cls.from_config(config, **kwargs). For plain
callables the factory is called as factory(config, **kwargs).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Strategy configuration dict (must contain
|
required |
**kwargs
|
Any
|
Forwarded to the factory (e.g. |
{}
|
Returns:
| Type | Description |
|---|---|
ReasoningStrategy
|
Configured strategy instance. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the strategy name is not registered. |
Source code in packages/bots/src/dataknobs_bots/reasoning/registry.py
get_factory ¶
Return the factory for a strategy name, or None.
is_registered ¶
list_keys ¶
BotTestHarness ¶
BotTestHarness(
bot: Any,
provider: EchoProvider,
extractor: ConfigurableExtractor | None,
context: Any,
)
High-level test helper for ALL DynaBot behavioral tests.
Wraps the full setup ceremony (bot creation, provider injection, tool
registration, middleware wiring, context management) into one object.
Use create() to build, chat()/greet() to run turns.
For non-wizard tests, use bot_config= with any DynaBot config:
Example
async with await BotTestHarness.create(
bot_config={
"llm": {"provider": "echo", "model": "test"},
"conversation_storage": {"backend": "memory"},
"reasoning": {"strategy": "simple"},
},
main_responses=[
tool_call_response("my_tool", {"q": "test"}),
text_response("Here are the results"),
],
tools=[my_tool],
middleware=[my_middleware],
) as harness:
result = await harness.chat("search")
assert result.response == "Here are the results"
# Streaming: harness.bot.stream_chat("msg", harness.context)
For wizard tests, use wizard_config= with WizardConfigBuilder:
Example
async with await BotTestHarness.create(
wizard_config=config,
main_responses=["Got it!", "All set!"],
extraction_results=[
[{"name": "Alice"}],
[{"domain_id": "chess"}, {"name": "Alice", "domain_id": "chess"}],
],
) as harness:
result = await harness.chat("My name is Alice")
assert harness.wizard_data["name"] == "Alice"
assert harness.wizard_stage == "gather"
Methods:
| Name | Description |
|---|---|
create |
Create a harness with a fully wired DynaBot. |
chat |
Run a chat turn and capture wizard state. |
greet |
Run a greet turn and capture wizard state. |
close |
Close the bot and release resources. |
Attributes:
| Name | Type | Description |
|---|---|---|
wizard_stage |
str | None
|
Current wizard stage from the last turn. |
wizard_data |
dict[str, Any]
|
Wizard state data dict from the last turn. |
wizard_state |
dict[str, Any] | None
|
Full wizard state from the last turn. |
last_response |
str
|
Response text from the last turn. |
turn_count |
int
|
Number of turns executed. |
bot |
Any
|
The underlying DynaBot instance. |
context |
Any
|
The BotContext used for this harness's turns. |
provider |
EchoProvider
|
The main EchoProvider (for call history assertions). |
extractor |
ConfigurableExtractor | None
|
The ConfigurableExtractor (for call verification). |
Source code in packages/bots/src/dataknobs_bots/testing.py
Attributes¶
extractor
property
¶
The ConfigurableExtractor (for call verification).
Functions¶
create
async
classmethod
¶
create(
*,
wizard_config: dict[str, Any] | None = None,
bot_config: dict[str, Any] | None = None,
main_responses: list[Any] | None = None,
extraction_results: list[list[dict[str, Any]]] | None = None,
system_prompt: str = "You are a helpful assistant.",
conversation_id: str = "test-conv",
client_id: str = "test",
extraction_scope: str = "current_message",
tools: list[Any] | None = None,
middleware: list[Any] | None = None,
strict_tools: bool = True,
strict: bool = False,
) -> BotTestHarness
Create a harness with a fully wired DynaBot.
Provide either wizard_config (auto-wires bot config) or
bot_config (full control).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
wizard_config
|
dict[str, Any] | None
|
Wizard config dict (e.g. from
|
None
|
bot_config
|
dict[str, Any] | None
|
Complete bot config dict for |
None
|
main_responses
|
list[Any] | None
|
Responses to queue on the main EchoProvider.
Accepts strings or |
None
|
extraction_results
|
list[list[dict[str, Any]]] | None
|
Per-turn extraction results. Each inner list
contains dicts for one turn's extraction calls. Flattened
into a |
None
|
system_prompt
|
str
|
System prompt text. |
'You are a helpful assistant.'
|
conversation_id
|
str
|
Conversation ID for the test context. |
'test-conv'
|
client_id
|
str
|
Client ID for the test context. |
'test'
|
extraction_scope
|
str
|
Default extraction scope for the wizard.
Only applies when |
'current_message'
|
tools
|
list[Any] | None
|
Optional list of |
None
|
middleware
|
list[Any] | None
|
Optional list of |
None
|
strict_tools
|
bool
|
If True (default), the EchoProvider raises ValueError when a scripted response contains tool_calls but no tools were provided to complete(). Set to False for tests that intentionally exercise unexpected tool_calls with no registered tools. |
True
|
strict
|
bool
|
If True, the EchoProvider raises
|
False
|
Returns:
| Type | Description |
|---|---|
BotTestHarness
|
Configured |
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither |
Source code in packages/bots/src/dataknobs_bots/testing.py
658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 | |
chat
async
¶
Run a chat turn and capture wizard state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
User message. |
required |
**kwargs
|
Any
|
Additional kwargs passed to |
{}
|
Returns:
| Type | Description |
|---|---|
TurnResult
|
|
Source code in packages/bots/src/dataknobs_bots/testing.py
greet
async
¶
Run a greet turn and capture wizard state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Any
|
Additional kwargs passed to |
{}
|
Returns:
| Type | Description |
|---|---|
TurnResult
|
|
Source code in packages/bots/src/dataknobs_bots/testing.py
CaptureReplay ¶
Loads a capture JSON file and creates pre-loaded EchoProviders.
Capture files contain serialized LLM request/response pairs from real provider runs, organized by turn. CaptureReplay deserializes these and creates EchoProviders queued with the correct responses, enabling deterministic replay of captured conversations.
Attributes:
| Name | Type | Description |
|---|---|---|
metadata |
dict[str, Any]
|
Capture session metadata (description, model info, timestamps) |
turns |
list[dict[str, Any]]
|
List of turn dicts with wizard state, user messages, bot responses |
format_version |
str
|
Capture file format version |
Example
Methods:
| Name | Description |
|---|---|
from_file |
Load a capture replay from a JSON file. |
from_dict |
Create a CaptureReplay from a dict (e.g., already-parsed JSON). |
main_provider |
Create an EchoProvider queued with main-role responses. |
extraction_provider |
Create an EchoProvider queued with extraction-role responses. |
inject_into_bot |
Replace providers on a DynaBot with capture-replay EchoProviders. |
Source code in packages/bots/src/dataknobs_bots/testing.py
Functions¶
from_file
classmethod
¶
Load a capture replay from a JSON file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to the capture JSON file |
required |
Returns:
| Type | Description |
|---|---|
CaptureReplay
|
CaptureReplay instance |
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
If the file does not exist |
JSONDecodeError
|
If the file is not valid JSON |
Source code in packages/bots/src/dataknobs_bots/testing.py
from_dict
classmethod
¶
Create a CaptureReplay from a dict (e.g., already-parsed JSON).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Capture data dict |
required |
Returns:
| Type | Description |
|---|---|
CaptureReplay
|
CaptureReplay instance |
Source code in packages/bots/src/dataknobs_bots/testing.py
main_provider ¶
Create an EchoProvider queued with main-role responses.
Returns:
| Type | Description |
|---|---|
EchoProvider
|
EchoProvider with responses in capture order |
Source code in packages/bots/src/dataknobs_bots/testing.py
extraction_provider ¶
Create an EchoProvider queued with extraction-role responses.
Returns:
| Type | Description |
|---|---|
EchoProvider
|
EchoProvider with responses in capture order |
Source code in packages/bots/src/dataknobs_bots/testing.py
inject_into_bot ¶
Replace providers on a DynaBot with capture-replay EchoProviders.
Creates main and extraction EchoProviders from the captured data
and injects them into the bot using inject_providers.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot
|
Any
|
A DynaBot instance |
required |
Source code in packages/bots/src/dataknobs_bots/testing.py
TurnResult
dataclass
¶
TurnResult(
response: str,
wizard_stage: str | None = None,
wizard_data: dict[str, Any] = dict(),
wizard_state: dict[str, Any] | None = None,
turn_index: int = 0,
)
Result of a single bot.chat() or bot.greet() turn.
Captures the bot response along with a snapshot of wizard state at the end of the turn.
Attributes:
| Name | Type | Description |
|---|---|---|
response |
str
|
Bot response text. |
wizard_stage |
str | None
|
Current wizard stage after this turn, or None if no wizard. |
wizard_data |
dict[str, Any]
|
Wizard state data dict after this turn. |
wizard_state |
dict[str, Any] | None
|
Full normalized wizard state after this turn, or None. |
turn_index |
int
|
One-based turn index (1 = first turn). |
Attributes¶
wizard_stage
class-attribute
instance-attribute
¶
Current wizard stage after this turn, or None if no wizard.
wizard_data
class-attribute
instance-attribute
¶
Wizard state data dict after this turn.
wizard_state
class-attribute
instance-attribute
¶
Full normalized wizard state after this turn, or None.
turn_index
class-attribute
instance-attribute
¶
One-based turn index (1 = first turn).
WizardConfigBuilder ¶
Fluent builder for wizard configuration dicts.
Replaces verbose inline dict construction (40+ lines) with a readable chained API. Performs build-time validation to catch common mistakes.
Example
config = (WizardConfigBuilder("my-wizard")
.stage("gather", is_start=True, prompt="Tell me your name.")
.field("name", field_type="string", required=True)
.field("domain", field_type="string")
.transition("done", "data.get('name') and data.get('domain')")
.stage("done", is_end=True, prompt="All done!")
.settings(extraction_scope="current_message")
.build())
Methods:
| Name | Description |
|---|---|
stage |
Add a stage to the wizard config. |
field |
Add a field to the current stage's schema. |
transition |
Add a transition from the current stage. |
settings |
Set wizard-level settings. |
build |
Build and validate the wizard config dict. |
Source code in packages/bots/src/dataknobs_bots/testing.py
Functions¶
stage ¶
stage(
name: str,
*,
is_start: bool = False,
is_end: bool = False,
prompt: str = "",
response_template: str | None = None,
mode: str | None = None,
extraction_scope: str | None = None,
auto_advance: bool | None = None,
skip_extraction: bool | None = None,
derivation_enabled: bool | None = None,
recovery_enabled: bool | None = None,
confirm_first_render: bool | None = None,
confirm_on_new_data: bool | None = None,
can_skip: bool | None = None,
skip_default: bool | None = None,
can_go_back: bool | None = None,
reasoning: str | None = None,
max_iterations: int | None = None,
capture_mode: str | None = None,
routing_transforms: list[str] | None = None,
**extra_fields: Any,
) -> WizardConfigBuilder
Add a stage to the wizard config.
After calling stage(), subsequent field() and
transition() calls apply to this stage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Stage name (unique identifier). |
required |
is_start
|
bool
|
Whether this is the start stage. |
False
|
is_end
|
bool
|
Whether this is an end stage. |
False
|
prompt
|
str
|
Stage prompt text. |
''
|
response_template
|
str | None
|
Jinja2 template rendered after extraction to confirm captured data. |
None
|
mode
|
str | None
|
Stage mode (e.g. |
None
|
extraction_scope
|
str | None
|
Per-stage extraction scope override. |
None
|
auto_advance
|
bool | None
|
Per-stage auto-advance override. |
None
|
skip_extraction
|
bool | None
|
Whether to skip extraction on this stage. |
None
|
derivation_enabled
|
bool | None
|
Per-stage field derivation override.
Set to |
None
|
recovery_enabled
|
bool | None
|
Per-stage recovery pipeline override.
Set to |
None
|
confirm_first_render
|
bool | None
|
Whether to pause for confirmation on
first render when new data is extracted. Default |
None
|
confirm_on_new_data
|
bool | None
|
Whether to re-confirm when schema property values change on subsequent renders. |
None
|
can_skip
|
bool | None
|
Whether the user can skip this stage. |
None
|
skip_default
|
bool | None
|
Default value to use when the stage is skipped. |
None
|
can_go_back
|
bool | None
|
Whether the user can navigate back from this stage. |
None
|
reasoning
|
str | None
|
Reasoning strategy for this stage
(e.g. |
None
|
max_iterations
|
int | None
|
Maximum ReAct iterations for this stage. |
None
|
capture_mode
|
str | None
|
Extraction capture mode — |
None
|
routing_transforms
|
list[str] | None
|
List of transform function names to execute before transition condition evaluation. |
None
|
**extra_fields
|
Any
|
Additional stage config fields passed through
to the stage dict verbatim. Use for less common fields
(e.g. |
{}
|
Returns:
| Type | Description |
|---|---|
WizardConfigBuilder
|
Self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/testing.py
107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 | |
field ¶
field(
name: str,
*,
field_type: str = "string",
required: bool = False,
description: str | None = None,
enum: list[str] | None = None,
default: Any = None,
x_extraction: dict[str, Any] | None = None,
) -> WizardConfigBuilder
Add a field to the current stage's schema.
Must be called after stage().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Field name. |
required |
field_type
|
str
|
JSON Schema type ( |
'string'
|
required
|
bool
|
Whether this field is required. |
False
|
description
|
str | None
|
Field description. |
None
|
enum
|
list[str] | None
|
Allowed values. |
None
|
default
|
Any
|
Default value. |
None
|
x_extraction
|
dict[str, Any] | None
|
Extraction hints ( |
None
|
Returns:
| Type | Description |
|---|---|
WizardConfigBuilder
|
Self for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no current stage is set. |
Source code in packages/bots/src/dataknobs_bots/testing.py
transition ¶
transition(
target: str, condition: str | None = None, priority: int | None = None
) -> WizardConfigBuilder
Add a transition from the current stage.
Must be called after stage().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target
|
str
|
Target stage name. |
required |
condition
|
str | None
|
Python expression evaluated against wizard state. |
None
|
priority
|
int | None
|
Transition evaluation priority (lower = first). |
None
|
Returns:
| Type | Description |
|---|---|
WizardConfigBuilder
|
Self for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no current stage is set. |
Source code in packages/bots/src/dataknobs_bots/testing.py
settings ¶
Set wizard-level settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Any
|
Settings key-value pairs (e.g.
|
{}
|
Returns:
| Type | Description |
|---|---|
WizardConfigBuilder
|
Self for method chaining. |
Source code in packages/bots/src/dataknobs_bots/testing.py
build ¶
Build and validate the wizard config dict.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Complete wizard configuration dict compatible with |
dict[str, Any]
|
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If validation fails (no start stage, no end stage, transition to nonexistent stage). |
Source code in packages/bots/src/dataknobs_bots/testing.py
AddKBResourceTool ¶
Bases: ContextAwareTool
Tool for adding a resource to the knowledge base resource list.
Supports adding file references (from the source directory) or inline content that gets written to the knowledge directory.
Wizard data read/written:
- _kb_resources: list[dict] — resource list (append)
- domain_id: str — used for knowledge directory organization
Attributes:
| Name | Type | Description |
|---|---|---|
_knowledge_dir |
Optional base directory for writing inline content. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
knowledge_dir
|
Path | None
|
Base directory for knowledge files. Used when
writing inline content to disk. Resolved from wizard data
|
None
|
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
Add a KB resource. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext,
path: str,
title: str = "",
resource_type: str = "file",
content: str | None = None,
description: str | None = None,
**kwargs: Any,
) -> dict[str, Any]
Add a KB resource.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
path
|
str
|
Resource path or filename. |
required |
title
|
str
|
Optional display title. |
''
|
resource_type
|
str
|
Type of resource ('file' or 'inline'). |
'file'
|
content
|
str | None
|
Inline content (required if resource_type='inline'). |
None
|
description
|
str | None
|
Optional resource description. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with add result. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 | |
CheckKnowledgeSourceTool ¶
Bases: ContextAwareTool
Tool for verifying a knowledge source directory exists and has content.
Checks the specified path for files matching common document patterns and records the results in wizard data for subsequent tools.
Wizard data written:
- source_verified: bool — whether the source was found
- files_found: list[str] — matching file names
- _source_path_resolved: str — the resolved absolute path
- _kb_resources: list[dict] — initialized if not present
Initialize the tool.
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
Check the knowledge source directory. |
Attributes:
| Name | Type | Description |
|---|---|---|
schema |
dict[str, Any]
|
Return JSON Schema for tool parameters. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext,
source_path: str,
file_patterns: list[str] | None = None,
**kwargs: Any,
) -> dict[str, Any]
Check the knowledge source directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
source_path
|
str
|
Path to the knowledge source directory. |
required |
file_patterns
|
list[str] | None
|
Optional glob patterns to match files. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with verification results. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
GetTemplateDetailsTool ¶
Bases: ContextAwareTool
Tool for getting detailed information about a template.
Returns the full template definition including all variables, their types, defaults, and constraints.
Attributes:
| Name | Type | Description |
|---|---|---|
_registry |
Template registry to query. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
template_registry
|
ConfigTemplateRegistry
|
Registry containing available templates. |
required |
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
from_config |
Create from YAML-compatible configuration. |
execute_with_context |
Get template details. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
from_config
classmethod
¶
Create from YAML-compatible configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Dict with |
required |
Returns:
| Type | Description |
|---|---|
GetTemplateDetailsTool
|
Configured GetTemplateDetailsTool instance. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext, template_name: str, **kwargs: Any
) -> dict[str, Any]
Get template details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context. |
required |
template_name
|
str
|
Name of the template. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with template details, or error if not found. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
IngestKnowledgeBaseTool ¶
Bases: ContextAwareTool
Tool for writing the KB ingestion manifest and finalizing KB config.
Writes a manifest.json file listing resources and chunking
parameters, and updates wizard data with the final KB configuration
for inclusion in the bot config.
Wizard data read:
- _kb_resources: list[dict] — resources to include
- domain_id: str — domain identifier
- files_found: list[str] — auto-discovered files (fallback)
- _source_path_resolved: str — resolved source path
Wizard data written:
- kb_config: dict — final KB configuration for the bot config
- kb_resources: list[dict] — finalized resource list (public key)
- ingestion_complete: bool — whether ingestion manifest was written
Attributes:
| Name | Type | Description |
|---|---|---|
_knowledge_dir |
Optional base directory for knowledge files. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
knowledge_dir
|
Path | None
|
Base directory for knowledge files. Resolved
from wizard data |
None
|
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
Write ingestion manifest and finalize KB config. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext, chunk_size: int = 512, **kwargs: Any
) -> dict[str, Any]
Write ingestion manifest and finalize KB config.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
chunk_size
|
int
|
Size of text chunks. |
512
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with ingestion result. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 | |
KnowledgeSearchTool ¶
Bases: ContextAwareTool
Tool for searching the knowledge base.
This tool allows LLMs to search the bot's knowledge base for relevant information during conversations.
Demonstrates the umbrella pattern for tools: - Static dependency: knowledge_base (via constructor injection) - Dynamic context: conversation_id, user_id (via ToolExecutionContext)
Example
# Create tool with knowledge base (static dependency)
tool = KnowledgeSearchTool(knowledge_base=kb)
# Register with bot
bot.tool_registry.register_tool(tool)
# LLM can now call the tool
# Context is automatically injected by reasoning strategy
results = await tool.execute(
query="How do I configure the database?",
max_results=3
)
Initialize knowledge search tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
knowledge_base
|
Any
|
RAGKnowledgeBase instance to search |
required |
name
|
str
|
Tool name (default: knowledge_search) |
'knowledge_search'
|
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
Execute knowledge base search with context. |
Attributes:
| Name | Type | Description |
|---|---|---|
schema |
dict[str, Any]
|
Get JSON schema for tool parameters. |
Source code in packages/bots/src/dataknobs_bots/tools/knowledge_search.py
Attributes¶
schema
property
¶
Get JSON schema for tool parameters.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
JSON Schema for the tool parameters |
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/knowledge_search.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext,
query: str,
max_results: int = 3,
**kwargs: Any,
) -> dict[str, Any]
Execute knowledge base search with context.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with conversation/user info |
required |
query
|
str
|
Search query text |
required |
max_results
|
int
|
Maximum number of results (default: 3) |
3
|
**kwargs
|
Any
|
Additional arguments (ignored) |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with search results: - query: Original query - results: List of relevant chunks - num_results: Number of results found - conversation_id: ID of conversation (if available) |
Example
Source code in packages/bots/src/dataknobs_bots/tools/knowledge_search.py
ListAvailableToolsTool ¶
Bases: ContextAwareTool
Tool for listing tools available to configure for a bot.
Takes a constructor-injected catalog of available tools and lets the LLM browse them, optionally filtering by category. The catalog data is consumer-specific — each DynaBot consumer provides its own list.
Attributes:
| Name | Type | Description |
|---|---|---|
_tools |
The available tool catalog. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
available_tools
|
list[dict[str, Any]]
|
List of tool descriptors. Each dict should
have at minimum |
required |
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
List available tools. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext, category: str | None = None, **kwargs: Any
) -> dict[str, Any]
List available tools.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context. |
required |
category
|
str | None
|
Optional category to filter by. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with matching tools, count, and available categories. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
ListKBResourcesTool ¶
Bases: ContextAwareTool
Tool for listing currently tracked knowledge base resources.
Reads _kb_resources and _source_path_resolved from wizard data
to show what resources have been added so far.
Initialize the tool.
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
List KB resources. |
Attributes:
| Name | Type | Description |
|---|---|---|
schema |
dict[str, Any]
|
Return JSON Schema for tool parameters. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
execute_with_context
async
¶
List KB resources.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with resource list and source path. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
ListTemplatesTool ¶
Bases: ContextAwareTool
Tool for listing available configuration templates.
Allows the LLM to discover what templates are available, optionally filtered by tags.
Attributes:
| Name | Type | Description |
|---|---|---|
_registry |
Template registry to query. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
template_registry
|
ConfigTemplateRegistry
|
Registry containing available templates. |
required |
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
from_config |
Create from YAML-compatible configuration. |
execute_with_context |
List available templates. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
from_config
classmethod
¶
Create from YAML-compatible configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Dict with |
required |
Returns:
| Type | Description |
|---|---|
ListTemplatesTool
|
Configured ListTemplatesTool instance. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext, tags: list[str] | None = None, **kwargs: Any
) -> dict[str, Any]
List available templates.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context. |
required |
tags
|
list[str] | None
|
Optional tags to filter by. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with list of template summaries. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
PreviewConfigTool ¶
Bases: ContextAwareTool
Tool for previewing the configuration being built.
Uses a consumer-provided builder_factory to construct the
configuration from wizard data. This is the key extension point:
the factory encapsulates domain-specific logic.
Attributes:
| Name | Type | Description |
|---|---|---|
_builder_factory |
Callable that creates a configured builder from wizard data. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
builder_factory
|
Callable[[dict[str, Any]], DynaBotConfigBuilder]
|
Function that takes wizard collected data and returns a configured DynaBotConfigBuilder. This is where consumers inject domain-specific config logic. |
required |
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
from_config |
Create from YAML-compatible configuration. |
execute_with_context |
Preview the current configuration. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
from_config
classmethod
¶
Create from YAML-compatible configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Dict with |
required |
Returns:
| Type | Description |
|---|---|
PreviewConfigTool
|
Configured PreviewConfigTool instance. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext, format: str = "summary", **kwargs: Any
) -> dict[str, Any]
Preview the current configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
format
|
str
|
Output format ('summary', 'full', or 'yaml'). |
'summary'
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with the configuration preview. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
RemoveKBResourceTool ¶
Bases: ContextAwareTool
Tool for removing a resource from the knowledge base resource list.
Wizard data read/written:
- _kb_resources: list[dict] — resource list (remove by name)
Initialize the tool.
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
execute_with_context |
Remove a KB resource. |
Attributes:
| Name | Type | Description |
|---|---|---|
schema |
dict[str, Any]
|
Return JSON Schema for tool parameters. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
execute_with_context
async
¶
Remove a KB resource.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
path
|
str
|
Path of the resource to remove. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with removal result. |
Source code in packages/bots/src/dataknobs_bots/tools/kb_tools.py
SaveConfigTool ¶
SaveConfigTool(
draft_manager: ConfigDraftManager,
on_save: Callable[[str, dict[str, Any]], Any] | None = None,
builder_factory: Callable[[dict[str, Any]], DynaBotConfigBuilder]
| None = None,
portable: bool = False,
)
Bases: ContextAwareTool
Tool for saving/finalizing the configuration.
Finalizes the draft and writes the final config file. Optionally calls a consumer-provided callback for post-save actions (e.g., registering the bot with a manager).
When portable=True, the builder's build_portable() method is
used instead of _build_internal(), producing a config with a
bot wrapper key suitable for environment-aware deployment.
Attributes:
| Name | Type | Description |
|---|---|---|
_draft_manager |
Draft manager for file operations. |
|
_on_save |
Optional callback invoked after successful save. |
|
_builder_factory |
Optional factory for building config from wizard data. |
|
_portable |
Whether to use portable (bot-wrapped) output format. |
Initialize the tool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
draft_manager
|
ConfigDraftManager
|
Manager for draft file operations. |
required |
on_save
|
Callable[[str, dict[str, Any]], Any] | None
|
Optional callback called with (config_name, config) after successful save. Can be used for post-save actions like bot registration. |
None
|
builder_factory
|
Callable[[dict[str, Any]], DynaBotConfigBuilder] | None
|
Optional factory to build final config from wizard data before saving. |
None
|
portable
|
bool
|
When True, use |
False
|
Methods:
| Name | Description |
|---|---|
catalog_metadata |
Return catalog metadata for this tool class. |
from_config |
Create from YAML-compatible configuration. |
execute_with_context |
Save the configuration. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
Attributes¶
Functions¶
catalog_metadata
classmethod
¶
Return catalog metadata for this tool class.
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
from_config
classmethod
¶
Create from YAML-compatible configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Dict with keys:
- |
required |
Returns:
| Type | Description |
|---|---|
SaveConfigTool
|
Configured SaveConfigTool instance. |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
execute_with_context
async
¶
execute_with_context(
context: ToolExecutionContext,
config_name: str | None = None,
activate: bool = False,
**kwargs: Any,
) -> dict[str, Any]
Save the configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
ToolExecutionContext
|
Execution context with wizard state. |
required |
config_name
|
str | None
|
Name for the config file. |
None
|
activate
|
bool
|
Whether to activate the bot. |
False
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dict with save result (success, file path, etc.). |
Source code in packages/bots/src/dataknobs_bots/tools/config_tools.py
672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 | |
Functions¶
normalize_wizard_state ¶
Normalize wizard metadata to canonical structure.
Handles both old nested format (fsm_state.current_stage) and new flat format (current_stage directly).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
wizard_meta
|
dict[str, Any]
|
Raw wizard metadata from manager or storage |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Normalized wizard state dict with canonical fields: |
dict[str, Any]
|
current_stage, stage_index, total_stages, progress, completed, |
dict[str, Any]
|
data, can_skip, can_go_back, suggestions, history, stages, |
dict[str, Any]
|
subflow_depth, and (when in a subflow) subflow_stage. |
Source code in packages/bots/src/dataknobs_bots/bot/base.py
create_default_catalog ¶
Create a new ToolCatalog pre-populated with built-in tools.
Returns a fresh catalog (not the module-level singleton) so consumers
can extend it without affecting other users of default_catalog.
Returns:
| Type | Description |
|---|---|
ToolCatalog
|
New ToolCatalog with all built-in tools registered. |
Source code in packages/bots/src/dataknobs_bots/config/tool_catalog.py
create_knowledge_base_from_config
async
¶
Create knowledge base from configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Knowledge base configuration with: - type: Type of knowledge base (currently only 'rag' supported) - vector_store: Vector store configuration - embedding_provider: LLM provider for embeddings - embedding_model: Model to use for embeddings - chunking: Optional chunking configuration - documents_path: Optional path to load documents - document_pattern: Optional file pattern |
required |
Returns:
| Type | Description |
|---|---|
KnowledgeBase
|
Configured knowledge base instance |
Raises:
| Type | Description |
|---|---|
ValueError
|
If knowledge base type is not supported |
Example
Source code in packages/bots/src/dataknobs_bots/knowledge/__init__.py
create_memory_from_config
async
¶
Create memory instance from configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Memory configuration with 'type' field and type-specific params |
required |
llm_provider
|
Any | None
|
Optional LLM provider instance, required for summary memory |
None
|
Returns:
| Type | Description |
|---|---|
Memory
|
Configured Memory instance |
Raises:
| Type | Description |
|---|---|
ValueError
|
If memory type is not recognized or required params missing |
Example
# Buffer memory
config = {
"type": "buffer",
"max_messages": 10
}
memory = await create_memory_from_config(config)
# Vector memory
config = {
"type": "vector",
"backend": "faiss",
"dimension": 768,
"embedding_provider": "ollama",
"embedding_model": "nomic-embed-text"
}
memory = await create_memory_from_config(config)
# Summary memory (uses bot's LLM as fallback)
config = {
"type": "summary",
"recent_window": 10,
}
memory = await create_memory_from_config(config, llm_provider=llm)
# Summary memory with its own dedicated LLM
config = {
"type": "summary",
"recent_window": 10,
"llm": {
"provider": "ollama",
"model": "gemma3:1b",
},
}
memory = await create_memory_from_config(config)
# Composite memory (multiple strategies)
config = {
"type": "composite",
"strategies": [
{"type": "buffer", "max_messages": 50},
{
"type": "vector",
"backend": "memory",
"dimension": 768,
"embedding_provider": "ollama",
"embedding_model": "nomic-embed-text",
},
],
"primary": 0,
}
memory = await create_memory_from_config(config)
Source code in packages/bots/src/dataknobs_bots/memory/__init__.py
56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | |
create_reasoning_from_config ¶
create_reasoning_from_config(
config: dict[str, Any], *, knowledge_base: Any | None = None
) -> ReasoningStrategy
Create reasoning strategy from configuration.
Delegates to the :class:StrategyRegistry singleton. Built-in
strategies (simple, react, wizard, grounded, hybrid) are registered
automatically; 3rd-party strategies can be added via
:func:register_strategy.
See each strategy class's from_config() for available config
keys (e.g. ReActReasoning.from_config,
WizardReasoning.from_config).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Reasoning configuration dict. The |
required |
knowledge_base
|
Any | None
|
Optional knowledge base instance forwarded as a kwarg to the strategy factory. |
None
|
Returns:
| Type | Description |
|---|---|
ReasoningStrategy
|
Configured reasoning strategy instance. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If strategy type is not registered. |
Example
# Simple reasoning
config = {"strategy": "simple"}
strategy = create_reasoning_from_config(config)
# Grounded reasoning (deterministic KB retrieval)
config = {
"strategy": "grounded",
"intent": {"mode": "extract", "num_queries": 3},
"retrieval": {"top_k": 5},
}
strategy = create_reasoning_from_config(config, knowledge_base=kb)
Source code in packages/bots/src/dataknobs_bots/reasoning/__init__.py
register_strategy ¶
Register a custom reasoning strategy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Strategy name (used in |
required |
factory
|
StrategyFactory
|
|
required |
override
|
bool
|
Replace existing registration if |
False
|
Example::
from dataknobs_bots.reasoning.registry import register_strategy
class MyStrategy(ReasoningStrategy):
...
register_strategy("my_strategy", MyStrategy)
Source code in packages/bots/src/dataknobs_bots/reasoning/registry.py
inject_providers ¶
inject_providers(
bot: Any,
main_provider: AsyncLLMProvider | None = None,
extraction_provider: AsyncLLMProvider | None = None,
*,
extractor: Any | None = None,
**role_providers: AsyncLLMProvider,
) -> None
Inject LLM providers into a DynaBot instance for testing.
For main_provider, directly replaces bot.llm (the "main"
role is always served from this attribute, not the registry catalog).
For extraction_provider and **role_providers, updates both the
registry catalog and the actual subsystem wiring via set_provider().
For extractor, calls strategy.set_extractor() to replace
the reasoning strategy's extractor entirely. Use this to inject a
ConfigurableExtractor (which is not an AsyncLLMProvider and
cannot be wired through set_provider()).
Lifecycle note: bot.close() will close self.llm (the main
provider) unconditionally — the caller should be aware that an
injected main_provider will be closed when the bot is closed.
For subsystem providers (memory embedding, extraction), ownership
flags control whether close() acts on them.
If bot does not implement register_provider, catalog
registration is skipped; only subsystem wiring via set_provider()
is performed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bot
|
Any
|
A DynaBot instance (or any object with |
required |
main_provider
|
AsyncLLMProvider | None
|
Provider to use for main LLM calls. If None, the existing provider is kept. |
None
|
extraction_provider
|
AsyncLLMProvider | None
|
Provider to use for schema extraction. If None, the existing provider is kept. |
None
|
extractor
|
Any | None
|
A |
None
|
**role_providers
|
AsyncLLMProvider
|
Additional providers keyed by role name
(e.g. |
{}
|
Example
Source code in packages/bots/src/dataknobs_bots/testing.py
939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 | |