API Reference

This document provides reference documentation for the Cognitive Engine's public APIs.

Table of Contents

• Core Engine API
• Cognitive Layers API
• Model Classes API
• Memory System API
• Agent API
• Learning System API
• Dashboard API
• LLM Client API
• Tools API

---

Core Engine API

CognitiveEngine

Module: core.engine

Class: CognitiveEngine

Main orchestration engine that coordinates all cognitive layers.

Constructor

def __init__(config: Optional[Config] = None)

Parameters:

• config (Optional[Config]): Configuration object. If None, uses default config.

Example:

from core.engine import CognitiveEngine

engine = CognitiveEngine()

process()

def process(self, input: str) -> ProcessResult

Process a user input through the cognitive engine.

Parameters:

• input (str): User query or problem statement

Returns:

• ProcessResult: Object containing final output and metadata

Example:

result = engine.process("What is the meaning of life?")
print(result.final_output)
print(result.iteration_count)
print(result.reasoning_trace)

initialize_layers()

def initialize_layers(self) -> None

Initialize all cognitive layers (Interpreter, Generator, Deliberator, Committer, Meta).

Example:

engine.initialize_layers()

get_thought_graph()

def get_thought_graph(self) -> ThoughtGraph

Get the current thought graph containing all generated thoughts.

Returns:

• ThoughtGraph: Graph of interconnected thoughts

Example:

graph = engine.get_thought_graph()
for thought_id, thought in graph.thoughts.items():
    print(f"{thought.premise}: {thought.confidence}")

get_memory()

def get_memory(self) -> ThreeLayerMemory

Get the three-layer memory system.

Returns:

• ThreeLayerMemory: Memory system with episodic, pattern, and rule layers

Example:

memory = engine.get_memory()
episodic = memory.get_episodic_memory()
patterns = memory.get_pattern_memory()
rules = memory.get_rule_memory()

---

Cognitive Layers API

Interpreter

Module: layers.interpreter

Class: Interpreter

Transform raw input into structured ProblemState.

interpret()

def interpret(self, input: str) -> ProblemState

Transform raw input into structured problem state.

Parameters:

• input (str): Raw user input

Returns:

• ProblemState: Structured problem definition

Example:

from layers.interpreter import Interpreter

interpreter = Interpreter()
state = interpreter.interpret("How do I learn Python?")
print(state.goals)
print(state.constraints)
print(state.knowns)

Generator

Module: layers.generator

Class: Generator

Create multiple competing thought candidates.

generate()

def generate(self, problem_state: ProblemState) -> List[Thought]

Generate thought candidates from problem state.

Parameters:

• problem_state (ProblemState): Structured problem definition

Returns:

• List[Thought]: List of generated thought objects

Example:

from layers.generator import Generator

generator = Generator()
thoughts = generator.generate(problem_state)
for thought in thoughts:
    print(f"{thought.premise}: {thought.confidence}")

Deliberator

Module: layers.deliberator

Class: Deliberator

Evaluate, test, and evolve thoughts.

deliberate()

def deliberate(self, thoughts: List[Thought]) -> List[Thought]

Evaluate and evolve a list of thoughts.

Parameters:

• thoughts (List[Thought]): List of thoughts to deliberate

Returns:

• List[Thought]: Evolved and scored list of thoughts

Example:

from layers.deliberator import Deliberator

deliberator = Deliberator()
evolved_thoughts = deliberator.deliberate(thoughts)
for thought in evolved_thoughts:
    print(f"Score: {thought.score}, Confidence: {thought.confidence}")

Committer

Module: layers.committer

Class: Committer

Select best thought and express output.

commit()

def commit(self, thoughts: List[Thought]) -> ProcessResult

Commit to the best thought and produce output.

Parameters:

• thoughts (List[Thought]): List of thoughts to choose from

Returns:

• ProcessResult: Final result with output and metadata

Example:

from layers.committer import Committer

committer = Committer()
result = committer.commit(evolved_thoughts)
print(result.final_output)

Meta-Cognition

Module: layers.meta

Class: MetaCognition

Govern the thinking process itself.

should_continue()

def should_continue(self, iteration: int, thoughts: List[Thought]) -> bool

Determine if cognitive process should continue.

Parameters:

• iteration (int): Current iteration number
• thoughts (List[Thought]): Current thoughts

Returns:

• bool: True if should continue, False if should stop

Example:

from layers.meta import MetaCognition

meta = MetaCognition({
    "min_iterations": 3,
    "max_iterations": 50,
    "early_stop_confidence": 0.95
})

if meta.should_continue(iteration, thoughts):
    # Continue deliberation
    pass
else:
    # Stop and commit
    pass

---

Model Classes API

Thought

Module: models.thought

Class: Thought

First-class thought object with state, history, and evaluative properties.

Constructor

def __init__(
    id: str,
    premise: str,
    confidence: float = 0.5,
    parent_nodes: Optional[List[str]] = None
)

Parameters:

• id (str): Unique identifier
• premise (str): Core hypothesis
• confidence (float): Initial confidence (0-1)
• parent_nodes (Optional[List[str]]): Parent thought IDs

Example:

from models.thought import Thought

thought = Thought(
    id="thought_1",
    premise="Python is a versatile programming language",
    confidence=0.7
)

Methods

update_confidence()

def update_confidence(self, new_confidence: float) -> None

add_weakness()

def add_weakness(self, weakness: str) -> None

record_test()

def record_test(self, test_result: Dict[str, Any]) -> None

revise()

def revise(self, new_premise: str) -> Thought

ThoughtGraph

Module: models.thought

Class: ThoughtGraph

Graph of interconnected thoughts.

Methods

add_thought()

def add_thought(self, thought: Thought) -> None

get_thought()

def get_thought(self, thought_id: str) -> Optional[Thought]

get_related()

def get_related(self, thought_id: str) -> List[Thought]

find_path()

def find_path(self, from_id: str, to_id: str) -> List[Thought]

ProblemState

Module: models.state

Class: ProblemState

Structured problem definition.

Constructor

def __init__(
    goals: List[str],
    constraints: List[str] = None,
    knowns: List[str] = None,
    unknowns: List[str] = None,
    context: str = ""
)

Example:

from models.state import ProblemState

state = ProblemState(
    goals=["Learn Python programming"],
    constraints=["Limited time available"],
    knowns=["Python is popular", "Python has good documentation"],
    unknowns=["Best learning resources", "Time required"]
)

---

Memory System API

ThreeLayerMemory

Module: core.memory

Class: ThreeLayerMemory

Three-layer memory system (Episodic, Pattern, Rule).

Methods

store_episodic()

def store_episodic(self, event: Dict[str, Any]) -> None

Store a raw event in episodic memory.

Parameters:

• event (Dict): Event data to store

Example:

memory.store_episodic({
    "type": "thought_generation",
    "timestamp": datetime.now(),
    "data": {"thought_premise": "..."}
})

extract_patterns()

def extract_patterns(self) -> List[Pattern]

Extract recurring patterns from episodic memory.

Returns:

• List[Pattern]: List of extracted patterns

synthesize_rules()

def synthesize_rules(self, patterns: List[Pattern]) -> List[Rule]

Convert patterns into actionable rules.

Parameters:

• patterns (List[Pattern]): Patterns to convert

Returns:

• List[Rule]: List of synthesized rules

apply_rules()

def apply_rules(self, state: ProblemState) -> ProblemState

Apply learned rules to problem state.

Parameters:

• state (ProblemState): Problem state to modify

Returns:

• ProblemState: Modified problem state

get_episodic_memory()

def get_episodic_memory(self) -> List[Dict]

Retrieve episodic memory.

Returns:

• List[Dict]: All episodic events

get_pattern_memory()

def get_pattern_memory(self) -> List[Pattern]

Retrieve pattern memory.

Returns:

• List[Pattern]: All extracted patterns

get_rule_memory()

def get_rule_memory(self) -> List[Rule]

Retrieve rule memory.

Returns:

• List[Rule]: All learned rules

---

Agent API

Agent

Module: agent.agent

Class: Agent

Autonomous agent with goal-directed behavior.

Constructor

def __init__(self, tool_registry: Optional[ToolRegistry] = None)

run()

def run(self, goal: str) -> AgentResult

Execute agent with a goal.

Parameters:

• goal (str): Goal to achieve

Returns:

• AgentResult: Result of agent execution

Example:

from agent.agent import Agent

agent = Agent()
result = agent.run("Research the latest developments in AI")
print(result.final_state)
print(result.actions_taken)

Planner

Module: agent.planner

Class: Planner

Convert goals into actionable plans.

create_plan()

def create_plan(self, goal: str) -> Plan

Create execution plan from goal.

Parameters:

• goal (str): Goal to plan for

Returns:

• Plan: Execution plan

Executor

Module: agent.executor

Class: Executor

Execute actions using tools.

execute()

def execute(self, plan: Plan) -> ActionResult

Execute a plan.

Parameters:

• plan (Plan): Plan to execute

Returns:

• ActionResult: Result of execution

Observer

Module: agent.observer

Class: Observer

Interpret and analyze execution results.

observe()

def observe(self, result: ActionResult) -> Observation

Observe and analyze execution results.

Parameters:

• result (ActionResult): Result to observe

Returns:

• Observation: Analysis of results

---

Learning System API

PatternExtractor

Module: learning.extractor

Class: PatternExtractor

Extract patterns from memory.

extract_patterns()

def extract_patterns(self, memory: List[Dict]) -> List[Pattern]

Extract patterns from episodic memory.

Parameters:

• memory (List[Dict]): Episodic memory events

Returns:

• List[Pattern]: Extracted patterns

RuleSynthesizer

Module: learning.synthesizer

Class: RuleSynthesizer

Convert patterns into rules.

synthesize_rules()

def synthesize_rules(self, patterns: List[Pattern]) -> List[Rule]

Synthesize rules from patterns.

Parameters:

• patterns (List[Pattern]): Patterns to convert

Returns:

• List[Rule]: Synthesized rules

KnowledgeUpdater

Module: learning.updater

Class: KnowledgeUpdater

Inject learned knowledge into reasoning system.

update_rules()

def update_rules(self, rules: List[Rule]) -> None

Update rule memory with new rules.

Parameters:

• rules (List[Rule]): Rules to add

apply_rule()

def apply_rule(self, rule: Rule, state: ProblemState) -> ProblemState

Apply a rule to problem state.

Parameters:

• rule (Rule): Rule to apply
• state (ProblemState): Problem state to modify

Returns:

• ProblemState: Modified problem state

---

Dashboard API

DashboardServer

Module: dashboard.server

Class: DashboardServer

WebSocket server for real-time telemetry.

Constructor

def __init__(self, host: str = "0.0.0.0", port: int = 8000)

start()

def start(self) -> None

Start the WebSocket server.

Example:

from dashboard.server import DashboardServer

server = DashboardServer(port=8000)
server.start()

DashboardStreamer

Module: dashboard.stream

Class: DashboardStreamer

Stream cognitive events to dashboard.

stream_event()

def stream_event(self, event: CognitiveEvent) -> None

Stream a cognitive event to dashboard.

Parameters:

• event (CognitiveEvent): Event to stream

Example:

from dashboard.stream import dashboard_streamer
from dashboard.events import ThoughtEvent

event = ThoughtEvent(
    thought_id="thought_1",
    premise="Example thought",
    confidence=0.8
)
dashboard_streamer.stream_event(event)

---

LLM Client API

LLMClient

Module: llm.client

Class: LLMClient

Single entry point for LLM interactions.

Constructor

def __init__(self, provider: str = "openai")

Parameters:

• provider (str): LLM provider ("openai" or "anthropic")

generate()

def generate(self, prompt: str) -> str

Generate response from LLM.

Parameters:

• prompt (str): Prompt to send to LLM

Returns:

• `str**: LLM response

Example:

from llm.client import LLMClient

client = LLMClient(provider="openai")
response = client.generate("Explain quantum computing")
print(response)

generate_with_context()

def generate_with_context(self, prompt: str, context: Dict) -> str

Generate response with additional context.

Parameters:

• prompt (str): Prompt to send
• context (Dict): Additional context

Returns:

• str: LLM response

stream_response()

def stream_response(self, prompt: str) -> Iterator[str]

Stream response from LLM.

Parameters:

• prompt (str): Prompt to send

Returns:

• Iterator[str]: Stream of response chunks

---

Tools API

ToolRegistry

Module: tools.registry

Class: ToolRegistry

Manage and execute tools.

register_tool()

def register_tool(self, tool: Tool) -> None

Register a tool.

Parameters:

• tool (Tool): Tool to register

call_tool()

def call_tool(self, name: str, params: Dict) -> ToolResult

Call a registered tool.

Parameters:

• name (str): Tool name
• params (Dict): Tool parameters

Returns:

• ToolResult: Result of tool execution

Example:

from tools.registry import tool_registry

result = tool_registry.call_tool("web_search", {"query": "AI news"})
print(result.data)

WebSearchTool

Module: tools.web_search

Class: WebSearchTool

Web search tool for information retrieval.

search()

def search(self, query: str) -> List[SearchResult]

Execute web search.

Parameters:

• query (str): Search query

Returns:

• List[SearchResult]: Search results

Example:

from tools.web_search import WebSearchTool

tool = WebSearchTool()
results = tool.search("latest AI developments")
for result in results:
    print(f"{result.title}: {result.url}")

CodeExecutionTool

Module: tools.code_exec

Class: CodeExecutionTool

Safe code execution tool.

execute()

def execute(self, code: str) -> ExecutionResult

Execute code safely.

Parameters:

• code (str): Code to execute

Returns:

• ExecutionResult: Execution result with output

Example:

from tools.code_exec import CodeExecutionTool

tool = CodeExecutionTool()
result = tool.execute("print('Hello, World!')")
print(result.output)

---

Configuration API

Config

Module: core.config

Class: Config

Configuration management.

Constructor

def __init__(
    min_iterations: int = 3,
    max_iterations: int = 50,
    early_stop_confidence: float = 0.95,
    confidence_threshold: float = 0.7,
    enable_dashboard: bool = True,
    dashboard_port: int = 8000,
    default_llm_provider: str = "openai"
)

load_from_env()

@classmethod
def load_from_env(cls) -> Config

Load configuration from environment variables.

Returns:

• Config: Configuration object

Example:

from core.config import Config

config = Config.load_from_env()
print(config.min_iterations)
print(config.default_llm_provider)

---

Error Handling

Common Exceptions

CognitiveEngineError
Base exception for cognitive engine errors.

MemoryError
Exception raised for memory-related errors.

ToolExecutionError
Exception raised when tool execution fails.

LLMError
Exception raised for LLM-related errors.

PatternExtractionError
Exception raised for pattern extraction failures.

RuleSynthesisError
Exception raised for rule synthesis failures.

Example Error Handling

from core.engine import CognitiveEngine
from core.engine import CognitiveEngineError

try:
    engine = CognitiveEngine()
    result = engine.process("Your query here")
except CognitiveEngineError as e:
    print(f"Cognitive engine error: {e}")
except MemoryError as e:
    print(f"Memory error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

---

Type Definitions

ProcessResult

class ProcessResult:
    final_output: str
    iteration_count: int
    reasoning_trace: Optional[str]
    thought_graph: ThoughtGraph
    confidence: float
    timestamp: datetime

AgentResult

class AgentResult:
    final_state: Dict
    actions_taken: List[Action]
    observations: List[Observation]
    goal_achieved: bool
    execution_time: float

ToolResult

class ToolResult:
    success: bool
    data: Any
    error: Optional[str]
    execution_time: float

SearchResult

class SearchResult:
    title: str
    url: str
    snippet: str
    relevance: float

ExecutionResult

class ExecutionResult:
    output: str
    error: Optional[str]
    execution_time: float
    success: bool

---

Usage Examples

Basic Usage

from core.engine import CognitiveEngine

# Initialize engine
engine = CognitiveEngine()

# Process a query
result = engine.process("What is cognitive science?")

# Access results
print(result.final_output)
print(result.confidence)
print(result.iteration_count)

With Custom Configuration

from core.engine import CognitiveEngine
from core.config import Config

# Create custom config
config = Config(
    min_iterations=5,
    max_iterations=20,
    confidence_threshold=0.8
)

# Initialize with config
engine = CognitiveEngine(config)
result = engine.process("Explain quantum computing")

Accessing Thought Graph

from core.engine import CognitiveEngine

engine = CognitiveEngine()
result = engine.process("Compare Python and JavaScript")

# Access thought graph
graph = engine.get_thought_graph()
for thought_id, thought in graph.thoughts.items():
    print(f"Thought: {thought.premise}")
    print(f"Confidence: {thought.confidence}")
    print(f"Score: {thought.score}")

Using Memory

from core.engine import CognitiveEngine

engine = CognitiveEngine()

# Process multiple queries
engine.process("What is machine learning?")
engine.process("How do neural networks work?")

# Access memory
memory = engine.get_memory()
episodic = memory.get_episodic_memory()
patterns = memory.get_pattern_memory()
rules = memory.get_rule_memory()

print(f"Episodic entries: {len(episodic)}")
print(f"Patterns found: {len(patterns)}")
print(f"Rules learned: {len(rules)}")

Agent Mode

from agent.agent import Agent

agent = Agent()
result = agent.run("Research the latest developments in AI")

print(f"Goal achieved: {result.goal_achieved}")
print(f"Actions taken: {len(result.actions_taken)}")

Using Tools

from tools.registry import tool_registry

# Call web search tool
result = tool_registry.call_tool("web_search", {
    "query": "latest AI news"
})

# Call code execution tool
result = tool_registry.call_tool("code_exec", {
    "code": "print(2 + 2)"
})