INTRODUCTION TO ADVANCED REASONING IN LANGUAGE MODELS
The landscape of artificial intelligence has been transformed by large language models, but their true potential emerges when we enhance them with sophisticated reasoning techniques. This tutorial explores the construction of an advanced chatbot system that leverages Chain-of-Thought reasoning, Tree-of-Thought exploration, and Self-Reflection mechanisms to produce higher quality responses than standard prompt-response systems.
Chain-of-Thought reasoning enables the model to break down complex problems into intermediate steps, making its reasoning process transparent and more accurate. Tree-of-Thought extends this by exploring multiple reasoning paths simultaneously, evaluating different approaches before selecting the best solution. Self-Reflection adds a critical layer where the system evaluates its own outputs, identifies potential improvements, and refines responses iteratively.
Our system will support both local and remote language models, providing flexibility for different deployment scenarios. Local models offer privacy and control, running on your own hardware with support for diverse GPU architectures including NVIDIA CUDA, AMD ROCm, Intel, and Apple Metal Performance Shaders. Remote models provide access to more powerful systems without local computational requirements.
The chatbot will implement intelligent prompt enhancement, where user inputs are analyzed and transformed into more effective prompts that elicit better responses from the underlying language model. It will maintain conversation history with intelligent summarization to handle long dialogues without exceeding context limits. Session management allows users to save conversations and resume them later, creating a persistent conversational experience.
The web-based user interface will provide an intuitive, visually appealing environment for interacting with the chatbot, displaying not only the final responses but also the intermediate reasoning steps, allowing users to understand how the system arrived at its conclusions.
ARCHITECTURAL OVERVIEW OF THE SYSTEM
The system architecture consists of several interconnected components working together to create an intelligent conversational experience. At the foundation lies the Language Model Interface layer, which abstracts the differences between local and remote models, providing a unified API for the rest of the system. This abstraction allows seamless switching between different model providers without changing higher-level code.
The Reasoning Engine sits above the model interface and implements the three core reasoning strategies. The Chain-of-Thought module breaks complex queries into sequential reasoning steps. The Tree-of-Thought module generates multiple reasoning branches, evaluates them, and selects the most promising paths. The Self-Reflection module analyzes generated responses, identifies weaknesses, and triggers refinement when necessary.
The Prompt Enhancement system analyzes incoming user messages and transforms them into optimized prompts. It identifies the user's intent, extracts key information, and reformulates the query to maximize the language model's understanding and response quality. This component uses meta-prompting techniques where one language model call analyzes and improves the prompt for subsequent calls.
The Conversation Manager handles chat history, implementing intelligent summarization when conversations grow too long. It maintains context windows appropriate for each model type, compressing older messages while preserving essential information. The Session Storage component persists conversations to a database, enabling users to pause and resume discussions across different sessions.
The Web Interface provides the user-facing component, built with modern web technologies to create a responsive, real-time experience. It communicates with the backend through WebSocket connections for streaming responses and displays the reasoning process transparently to users.
SETTING UP THE DEVELOPMENT ENVIRONMENT
Before implementing the system, we need to establish a proper development environment with all necessary dependencies. The backend will use Python for its rich ecosystem of machine learning libraries and frameworks. The frontend will use HTML, CSS, and JavaScript with modern frameworks for reactive user interfaces.
For local model support, we will use the llama-cpp-python library, which provides Python bindings to the llama.cpp project. This library supports multiple GPU backends through compile-time flags, enabling acceleration on NVIDIA, AMD, Intel, and Apple hardware. For remote models, we will integrate with APIs from providers like OpenAI, Anthropic, and others through standardized interfaces.
The development environment requires Python 3.9 or higher. We will use FastAPI for the web server framework, providing automatic API documentation and excellent WebSocket support. SQLAlchemy will handle database operations for session storage. Pydantic will ensure type safety and data validation throughout the system.
For the frontend, we will use vanilla JavaScript with modern ES6+ features, avoiding heavy framework dependencies while maintaining clean, maintainable code. The interface will use CSS Grid and Flexbox for responsive layouts, with CSS custom properties for theming.
Here is the initial project structure and dependency configuration:
# requirements.txt - Python dependencies for the project
# Web framework and server
fastapi==0.104.1
uvicorn[standard]==0.24.0
python-multipart==0.0.6
# WebSocket support
websockets==12.0
# Database and ORM
sqlalchemy==2.0.23
aiosqlite==0.19.0
# Data validation and settings
pydantic==2.5.0
pydantic-settings==2.1.0
# Local LLM support with GPU acceleration
llama-cpp-python==0.2.20
# Remote LLM API clients
openai==1.3.7
anthropic==0.7.7
# Utilities
python-dotenv==1.0.0
tiktoken==0.5.2
The llama-cpp-python library requires special installation to enable GPU support. For NVIDIA CUDA support, you would install with the CMAKE_ARGS environment variable specifying CUDA backend. For AMD ROCm, you specify the ROCm backend. For Apple Metal, the Metal backend is enabled. For Intel, you would use the appropriate Intel backend flags.
The installation commands vary by platform. On a system with NVIDIA CUDA, you would set CMAKE_ARGS to include LLAMA_CUBLAS=on before installing llama-cpp-python. On AMD systems with ROCm, you would set LLAMA_HIPBLAS=on. On Apple Silicon, you would set LLAMA_METAL=on. The library automatically detects and uses available acceleration when properly compiled.
IMPLEMENTING THE LANGUAGE MODEL INTERFACE LAYER
The foundation of our system is a unified interface for interacting with different language models. This abstraction allows the reasoning engine and other components to work with any model without knowing whether it is local or remote, what API it uses, or what specific capabilities it has.
We define a base abstract class that all model implementations must inherit from. This class specifies the essential methods that any language model must provide: generating completions, streaming responses, and reporting model capabilities like context window size and token limits.
# models/base.py - Abstract base class for all language model implementations
from abc import ABC, abstractmethod
from typing import List, Dict, Any, AsyncGenerator, Optional
from dataclasses import dataclass
@dataclass
class ModelConfig:
"""Configuration for a language model instance."""
model_name: str
temperature: float = 0.7
max_tokens: int = 2048
top_p: float = 0.9
context_window: int = 4096
gpu_layers: Optional[int] = None # For local models
api_key: Optional[str] = None # For remote models
@dataclass
class Message:
"""Represents a single message in a conversation."""
role: str # 'system', 'user', or 'assistant'
content: str
def to_dict(self) -> Dict[str, str]:
"""Convert message to dictionary format."""
return {"role": self.role, "content": self.content}
class LanguageModel(ABC):
"""Abstract base class for all language model implementations."""
def __init__(self, config: ModelConfig):
"""Initialize the language model with configuration.
Args:
config: ModelConfig object containing model parameters
"""
self.config = config
self.model_name = config.model_name
@abstractmethod
async def generate(self, messages: List[Message], **kwargs) -> str:
"""Generate a completion for the given messages.
Args:
messages: List of Message objects representing the conversation
**kwargs: Additional generation parameters
Returns:
Generated text response as a string
"""
pass
@abstractmethod
async def stream_generate(
self,
messages: List[Message],
**kwargs
) -> AsyncGenerator[str, None]:
"""Stream generation tokens as they are produced.
Args:
messages: List of Message objects representing the conversation
**kwargs: Additional generation parameters
Yields:
Individual tokens or chunks of generated text
"""
pass
@abstractmethod
def count_tokens(self, text: str) -> int:
"""Count the number of tokens in the given text.
Args:
text: Text to count tokens for
Returns:
Number of tokens as an integer
"""
pass
@abstractmethod
def get_context_window(self) -> int:
"""Get the maximum context window size for this model.
Returns:
Maximum number of tokens the model can process
"""
pass
This base class establishes the contract that all model implementations must fulfill. The Message dataclass provides a clean representation of conversation turns, with a role indicating whether the message comes from the system, user, or assistant, and content containing the actual text.
The ModelConfig dataclass encapsulates all configuration parameters needed to initialize a model. It includes both common parameters like temperature and max_tokens that apply to all models, as well as optional parameters like gpu_layers for local models and api_key for remote models.
The abstract methods define the core functionality. The generate method produces a complete response synchronously, while stream_generate yields tokens as they are produced, enabling real-time display in the user interface. The count_tokens method is essential for managing context windows, and get_context_window reports the model's maximum capacity.
IMPLEMENTING LOCAL MODEL SUPPORT
Local models provide privacy, control, and independence from external services. We implement support for local models using the llama-cpp-python library, which can load and run models in GGUF format with GPU acceleration across multiple platforms.
The LocalLanguageModel class inherits from our LanguageModel base class and implements all required methods using llama-cpp-python's API. During initialization, it loads the model file and configures GPU acceleration based on the available hardware.
# models/local_model.py - Implementation for local LLM support
from typing import List, AsyncGenerator, Optional
import asyncio
from llama_cpp import Llama
from models.base import LanguageModel, ModelConfig, Message
class LocalLanguageModel(LanguageModel):
"""Implementation of LanguageModel for local models using llama.cpp."""
def __init__(self, config: ModelConfig, model_path: str):
"""Initialize a local language model.
Args:
config: ModelConfig object with model parameters
model_path: Path to the GGUF model file on disk
"""
super().__init__(config)
self.model_path = model_path
# Determine GPU layers to offload based on configuration
n_gpu_layers = config.gpu_layers if config.gpu_layers is not None else -1
# Initialize the llama.cpp model with GPU acceleration
# The n_gpu_layers parameter controls how many layers run on GPU
# Setting it to -1 offloads all layers to GPU if available
self.model = Llama(
model_path=model_path,
n_ctx=config.context_window,
n_gpu_layers=n_gpu_layers,
verbose=False
)
def _format_messages(self, messages: List[Message]) -> str:
"""Format messages into a prompt string for the model.
Args:
messages: List of Message objects
Returns:
Formatted prompt string
"""
# Different models use different chat templates
# This is a generic format that works with many models
prompt_parts = []
for message in messages:
if message.role == "system":
prompt_parts.append(f"System: {message.content}\n")
elif message.role == "user":
prompt_parts.append(f"User: {message.content}\n")
elif message.role == "assistant":
prompt_parts.append(f"Assistant: {message.content}\n")
# Add the assistant prefix to prompt for a response
prompt_parts.append("Assistant:")
return "\n".join(prompt_parts)
async def generate(self, messages: List[Message], **kwargs) -> str:
"""Generate a completion using the local model.
Args:
messages: Conversation history as Message objects
**kwargs: Additional generation parameters
Returns:
Generated response text
"""
prompt = self._format_messages(messages)
# Extract generation parameters with defaults from config
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
# Run model inference in a thread pool to avoid blocking
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stop=["User:", "\n\n"],
echo=False
)
)
# Extract the generated text from the response
generated_text = response['choices'][0]['text'].strip()
return generated_text
async def stream_generate(
self,
messages: List[Message],
**kwargs
) -> AsyncGenerator[str, None]:
"""Stream generation tokens from the local model.
Args:
messages: Conversation history
**kwargs: Generation parameters
Yields:
Individual tokens as they are generated
"""
prompt = self._format_messages(messages)
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
# Create a streaming generator from the model
stream = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stop=["User:", "\n\n"],
stream=True,
echo=False
)
# Yield tokens as they are produced
for output in stream:
token = output['choices'][0]['text']
yield token
# Small delay to prevent overwhelming the event loop
await asyncio.sleep(0.001)
def count_tokens(self, text: str) -> int:
"""Count tokens in text using the model's tokenizer.
Args:
text: Text to tokenize
Returns:
Number of tokens
"""
tokens = self.model.tokenize(text.encode('utf-8'))
return len(tokens)
def get_context_window(self) -> int:
"""Get the model's context window size.
Returns:
Maximum context length in tokens
"""
return self.config.context_window
The LocalLanguageModel implementation handles the complexities of running models locally. The initialization loads the model file and configures GPU acceleration. The n_gpu_layers parameter determines how many transformer layers are offloaded to the GPU, with negative one meaning all available layers should use GPU acceleration.
The format_messages method converts our Message objects into a text prompt that the model can process. Different models expect different chat formats, but this generic format with role prefixes works well with many models. More sophisticated implementations would detect the model type and use its specific chat template.
The generate method performs synchronous generation by running the model in a thread pool executor. This prevents blocking the async event loop while the model processes the request. The model returns a dictionary containing the generated text and metadata, from which we extract the actual response.
The stream_generate method enables token-by-token streaming, yielding each token as it is produced. This allows the user interface to display responses progressively, creating a more interactive experience. The small asyncio.sleep call prevents the generator from overwhelming the event loop with tokens.
Token counting uses the model's built-in tokenizer, ensuring accuracy for context window management. This is critical because different models use different tokenization schemes, and accurate token counts prevent context overflow errors.
IMPLEMENTING REMOTE MODEL SUPPORT
Remote models provide access to powerful language models without requiring local computational resources. We implement support for remote models through API clients, starting with OpenAI's GPT models as they are widely used and provide a well-documented API.
The RemoteLanguageModel class implements the same interface as LocalLanguageModel, but instead of running inference locally, it makes HTTP requests to remote API endpoints.
# models/remote_model.py - Implementation for remote LLM APIs
from typing import List, AsyncGenerator
import asyncio
from openai import AsyncOpenAI
from models.base import LanguageModel, ModelConfig, Message
import tiktoken
class RemoteLanguageModel(LanguageModel):
"""Implementation for remote language models via API."""
def __init__(self, config: ModelConfig):
"""Initialize a remote language model client.
Args:
config: ModelConfig with API key and model parameters
"""
super().__init__(config)
if not config.api_key:
raise ValueError("API key is required for remote models")
# Initialize the OpenAI async client
self.client = AsyncOpenAI(api_key=config.api_key)
# Load the appropriate tokenizer for token counting
try:
self.encoding = tiktoken.encoding_for_model(config.model_name)
except KeyError:
# Fallback to cl100k_base encoding for unknown models
self.encoding = tiktoken.get_encoding("cl100k_base")
async def generate(self, messages: List[Message], **kwargs) -> str:
"""Generate a completion using the remote API.
Args:
messages: Conversation history
**kwargs: Generation parameters
Returns:
Generated response text
"""
# Convert Message objects to API format
api_messages = [msg.to_dict() for msg in messages]
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
# Make API request
response = await self.client.chat.completions.create(
model=self.config.model_name,
messages=api_messages,
temperature=temperature,
max_tokens=max_tokens,
top_p=top_p
)
# Extract the assistant's message from the response
generated_text = response.choices[0].message.content
return generated_text
async def stream_generate(
self,
messages: List[Message],
**kwargs
) -> AsyncGenerator[str, None]:
"""Stream generation from the remote API.
Args:
messages: Conversation history
**kwargs: Generation parameters
Yields:
Chunks of generated text
"""
api_messages = [msg.to_dict() for msg in messages]
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
# Create streaming request
stream = await self.client.chat.completions.create(
model=self.config.model_name,
messages=api_messages,
temperature=temperature,
max_tokens=max_tokens,
top_p=top_p,
stream=True
)
# Yield chunks as they arrive
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
yield chunk.choices[0].delta.content
def count_tokens(self, text: str) -> int:
"""Count tokens using the model's tokenizer.
Args:
text: Text to count tokens for
Returns:
Number of tokens
"""
tokens = self.encoding.encode(text)
return len(tokens)
def get_context_window(self) -> int:
"""Get the context window size for this model.
Returns:
Maximum context length
"""
# Context windows for common OpenAI models
context_windows = {
"gpt-4": 8192,
"gpt-4-32k": 32768,
"gpt-4-turbo-preview": 128000,
"gpt-3.5-turbo": 4096,
"gpt-3.5-turbo-16k": 16384,
}
return context_windows.get(
self.config.model_name,
self.config.context_window
)
The RemoteLanguageModel uses the OpenAI Python client library to interact with the API. During initialization, it creates an async client with the provided API key and loads the appropriate tokenizer for accurate token counting. The tiktoken library provides tokenizers that match OpenAI's models exactly.
The generate method converts our Message objects to the dictionary format expected by the OpenAI API and makes an async request. The API returns a completion object containing the generated text, which we extract and return. Error handling in production code would catch API errors, rate limits, and network issues.
Streaming works similarly but sets the stream parameter to true and iterates over the async generator returned by the API. Each chunk contains a delta with new content, which we yield to our caller. This enables real-time display of responses as they are generated.
Token counting uses the tiktoken library, which implements the exact tokenization algorithms used by OpenAI's models. This ensures our token counts match what the API sees, preventing context window issues. The get_context_window method returns known context sizes for common models, with a fallback to the configured value.
IMPLEMENTING THE MODEL FACTORY
To simplify model instantiation and provide a clean interface for the rest of the system, we implement a factory pattern that creates the appropriate model type based on configuration.
# models/factory.py - Factory for creating language model instances
from typing import Union
from models.base import LanguageModel, ModelConfig
from models.local_model import LocalLanguageModel
from models.remote_model import RemoteLanguageModel
class ModelFactory:
"""Factory for creating language model instances."""
@staticmethod
def create_model(
model_type: str,
config: ModelConfig,
model_path: str = None
) -> LanguageModel:
"""Create a language model instance based on type.
Args:
model_type: Either 'local' or 'remote'
config: ModelConfig object with model parameters
model_path: Path to model file (required for local models)
Returns:
LanguageModel instance
Raises:
ValueError: If model_type is invalid or required params missing
"""
if model_type == "local":
if not model_path:
raise ValueError("model_path is required for local models")
return LocalLanguageModel(config, model_path)
elif model_type == "remote":
return RemoteLanguageModel(config)
else:
raise ValueError(
f"Unknown model type: {model_type}. "
"Must be 'local' or 'remote'"
)
@staticmethod
def create_from_settings(settings: dict) -> LanguageModel:
"""Create a model from a settings dictionary.
Args:
settings: Dictionary containing model configuration
Returns:
LanguageModel instance
"""
model_type = settings.get("type", "remote")
config = ModelConfig(
model_name=settings.get("model_name", "gpt-3.5-turbo"),
temperature=settings.get("temperature", 0.7),
max_tokens=settings.get("max_tokens", 2048),
top_p=settings.get("top_p", 0.9),
context_window=settings.get("context_window", 4096),
gpu_layers=settings.get("gpu_layers"),
api_key=settings.get("api_key")
)
model_path = settings.get("model_path")
return ModelFactory.create_model(model_type, config, model_path)
The ModelFactory provides two static methods for creating models. The create_model method takes explicit parameters and instantiates the appropriate model class. The create_from_settings method accepts a dictionary of settings, making it easy to load configuration from files or environment variables.
This factory pattern decouples the rest of the system from the specific model implementations. Code that needs a language model can simply call the factory without knowing whether it will get a local or remote model, as both implement the same interface.
IMPLEMENTING CHAIN-OF-THOUGHT REASONING
Chain-of-Thought reasoning enhances language model outputs by encouraging the model to show its work, breaking complex problems into intermediate steps. This technique significantly improves accuracy on reasoning tasks and makes the model's decision process transparent.
The implementation works by modifying prompts to explicitly request step-by-step reasoning. We create a ChainOfThought class that wraps a language model and automatically adds reasoning instructions to user queries.
# reasoning/chain_of_thought.py - Chain-of-Thought reasoning implementation
from typing import List, Dict, Any
from models.base import LanguageModel, Message
class ChainOfThought:
"""Implements Chain-of-Thought reasoning for language models."""
def __init__(self, model: LanguageModel):
"""Initialize Chain-of-Thought reasoner.
Args:
model: LanguageModel instance to use for reasoning
"""
self.model = model
# System prompt that encourages step-by-step reasoning
self.cot_system_prompt = """You are a helpful assistant that thinks step-by-step.
When answering questions, break down your reasoning into clear steps:
1. First, analyze what the question is asking
2. Identify the key information needed
3. Work through the problem step-by-step
4. Arrive at a conclusion based on your reasoning
Show your thought process clearly so the user can follow your logic."""
async def reason(
self,
messages: List[Message],
show_reasoning: bool = True
) -> Dict[str, Any]:
"""Apply Chain-of-Thought reasoning to generate a response.
Args:
messages: Conversation history
show_reasoning: Whether to include reasoning steps in output
Returns:
Dictionary containing reasoning steps and final answer
"""
# Create a modified message list with CoT system prompt
cot_messages = [Message(role="system", content=self.cot_system_prompt)]
# Add existing messages, filtering out any existing system messages
# to avoid conflicts
for msg in messages:
if msg.role != "system":
cot_messages.append(msg)
# Add instruction to show reasoning
if cot_messages[-1].role == "user":
original_content = cot_messages[-1].content
cot_messages[-1] = Message(
role="user",
content=f"{original_content}\n\nPlease think through this step-by-step."
)
# Generate response with reasoning
response = await self.model.generate(cot_messages)
# Parse the response to extract reasoning steps and final answer
parsed_response = self._parse_reasoning(response)
if not show_reasoning:
# Return only the final answer if reasoning display is disabled
return {
"reasoning_steps": [],
"final_answer": parsed_response["final_answer"]
}
return parsed_response
def _parse_reasoning(self, response: str) -> Dict[str, Any]:
"""Parse a response to extract reasoning steps and final answer.
Args:
response: Generated response text
Returns:
Dictionary with reasoning_steps list and final_answer string
"""
# Split response into lines for analysis
lines = response.split('\n')
reasoning_steps = []
final_answer_lines = []
in_conclusion = False
for line in lines:
line = line.strip()
if not line:
continue
# Detect conclusion markers
conclusion_markers = [
"therefore", "thus", "in conclusion", "finally",
"the answer is", "conclusion:"
]
if any(marker in line.lower() for marker in conclusion_markers):
in_conclusion = True
# Lines with step numbers or bullet points are reasoning steps
if line[0].isdigit() or line.startswith('-') or line.startswith('*'):
if not in_conclusion:
reasoning_steps.append(line)
else:
final_answer_lines.append(line)
else:
if in_conclusion:
final_answer_lines.append(line)
else:
# Regular text before conclusion is part of reasoning
if reasoning_steps:
reasoning_steps[-1] += " " + line
else:
reasoning_steps.append(line)
# If we couldn't identify a clear conclusion, treat last step as answer
if not final_answer_lines:
if reasoning_steps:
final_answer_lines = [reasoning_steps.pop()]
else:
final_answer_lines = [response]
return {
"reasoning_steps": reasoning_steps,
"final_answer": "\n".join(final_answer_lines)
}
The ChainOfThought class wraps a language model and modifies interactions to encourage step-by-step reasoning. The cot_system_prompt provides instructions that guide the model to break down its thinking process. This system message is prepended to the conversation, establishing the reasoning behavior.
The reason method takes the conversation history and creates a modified version with the Chain-of-Thought system prompt. It also appends an explicit instruction to the user's last message, asking the model to think step-by-step. This dual approach of system-level and prompt-level instructions maximizes the likelihood of getting structured reasoning.
After generating the response, the parse_reasoning method analyzes the text to separate reasoning steps from the final answer. It looks for numbered lists, bullet points, and conclusion markers to identify different parts of the response. This parsing enables the user interface to display reasoning steps separately from the final answer, making the thought process transparent.
The show_reasoning parameter allows callers to control whether reasoning steps are returned. When false, only the final answer is provided, which is useful when you want the benefits of Chain-of-Thought reasoning for accuracy but don't need to display the intermediate steps to users.
IMPLEMENTING TREE-OF-THOUGHT REASONING
Tree-of-Thought extends Chain-of-Thought by exploring multiple reasoning paths simultaneously. Instead of following a single chain of reasoning, it generates several different approaches to a problem, evaluates each path, and selects the most promising solution.
The implementation generates multiple reasoning branches, scores them based on quality criteria, and can either select the best path or combine insights from multiple paths.
# reasoning/tree_of_thought.py - Tree-of-Thought reasoning implementation
from typing import List, Dict, Any, Tuple
import asyncio
from models.base import LanguageModel, Message
class TreeOfThought:
"""Implements Tree-of-Thought reasoning with multiple exploration paths."""
def __init__(self, model: LanguageModel, num_branches: int = 3):
"""Initialize Tree-of-Thought reasoner.
Args:
model: LanguageModel instance for reasoning
num_branches: Number of reasoning paths to explore
"""
self.model = model
self.num_branches = num_branches
self.tot_system_prompt = """You are a creative problem solver who explores multiple approaches.
When given a problem, consider different angles and strategies for solving it.
Be thorough and analytical in your exploration of each approach."""
async def reason(
self,
messages: List[Message],
selection_strategy: str = "best"
) -> Dict[str, Any]:
"""Apply Tree-of-Thought reasoning to explore multiple solution paths.
Args:
messages: Conversation history
selection_strategy: How to select final answer - 'best' or 'consensus'
Returns:
Dictionary containing all branches and selected answer
"""
# Generate multiple reasoning branches in parallel
branches = await self._generate_branches(messages)
# Evaluate each branch to determine quality
evaluated_branches = await self._evaluate_branches(branches, messages)
# Select final answer based on strategy
if selection_strategy == "best":
final_answer = self._select_best_branch(evaluated_branches)
elif selection_strategy == "consensus":
final_answer = await self._build_consensus(
evaluated_branches,
messages
)
else:
raise ValueError(f"Unknown selection strategy: {selection_strategy}")
return {
"branches": evaluated_branches,
"final_answer": final_answer,
"strategy": selection_strategy
}
async def _generate_branches(
self,
messages: List[Message]
) -> List[str]:
"""Generate multiple reasoning branches in parallel.
Args:
messages: Conversation history
Returns:
List of generated reasoning paths
"""
# Create prompts for different reasoning approaches
branch_prompts = self._create_branch_prompts(messages)
# Generate all branches concurrently
tasks = []
for prompt_messages in branch_prompts:
task = self.model.generate(prompt_messages)
tasks.append(task)
branches = await asyncio.gather(*tasks)
return branches
def _create_branch_prompts(
self,
messages: List[Message]
) -> List[List[Message]]:
"""Create different prompt variations to encourage diverse reasoning.
Args:
messages: Original conversation history
Returns:
List of message lists, each representing a different branch prompt
"""
branch_prompts = []
# Get the user's question from the last message
if messages[-1].role != "user":
raise ValueError("Last message must be from user")
user_question = messages[-1].content
# Define different reasoning approaches to explore
approaches = [
"Approach this problem analytically, breaking it into logical components.",
"Think about this creatively, considering unconventional solutions.",
"Consider this from first principles, building up from fundamental concepts.",
"Examine this problem from multiple perspectives before concluding.",
"Use analogies and related examples to reason about this problem."
]
# Create a prompt for each approach, using only as many as num_branches
for i in range(min(self.num_branches, len(approaches))):
branch_messages = [
Message(role="system", content=self.tot_system_prompt)
]
# Add conversation history except the last user message
for msg in messages[:-1]:
if msg.role != "system":
branch_messages.append(msg)
# Add modified user message with approach instruction
modified_question = f"{approaches[i]}\n\n{user_question}"
branch_messages.append(
Message(role="user", content=modified_question)
)
branch_prompts.append(branch_messages)
return branch_prompts
async def _evaluate_branches(
self,
branches: List[str],
original_messages: List[Message]
) -> List[Dict[str, Any]]:
"""Evaluate the quality of each reasoning branch.
Args:
branches: List of generated reasoning paths
original_messages: Original conversation for context
Returns:
List of dictionaries containing branch content and scores
"""
evaluated = []
for i, branch in enumerate(branches):
# Create evaluation prompt
eval_messages = [
Message(
role="system",
content="You are an expert evaluator of reasoning quality."
),
Message(
role="user",
content=f"""Evaluate the following reasoning on a scale of 1-10 based on:
- Logical coherence
- Completeness of analysis
- Correctness of conclusions
- Clarity of explanation
Original question: {original_messages[-1].content}
Reasoning to evaluate:
{branch}
Provide only a numeric score from 1-10."""
)
]
# Get evaluation score
score_text = await self.model.generate(eval_messages, temperature=0.3)
# Extract numeric score from response
try:
# Look for a number in the response
import re
numbers = re.findall(r'\d+', score_text)
score = int(numbers[0]) if numbers else 5
score = max(1, min(10, score)) # Clamp to 1-10 range
except (ValueError, IndexError):
score = 5 # Default middle score if parsing fails
evaluated.append({
"branch_id": i,
"content": branch,
"score": score
})
return evaluated
def _select_best_branch(
self,
evaluated_branches: List[Dict[str, Any]]
) -> str:
"""Select the highest-scoring branch as the final answer.
Args:
evaluated_branches: Branches with scores
Returns:
Content of the best branch
"""
best_branch = max(evaluated_branches, key=lambda b: b["score"])
return best_branch["content"]
async def _build_consensus(
self,
evaluated_branches: List[Dict[str, Any]],
original_messages: List[Message]
) -> str:
"""Build a consensus answer from multiple branches.
Args:
evaluated_branches: All branches with scores
original_messages: Original conversation
Returns:
Synthesized consensus answer
"""
# Create a synthesis prompt with all branches
branches_text = "\n\n---\n\n".join([
f"Approach {b['branch_id'] + 1} (Score: {b['score']}/10):\n{b['content']}"
for b in evaluated_branches
])
synthesis_messages = [
Message(
role="system",
content="You synthesize insights from multiple reasoning paths."
),
Message(
role="user",
content=f"""Given these different reasoning approaches to the question:
"{original_messages[-1].content}"
{branches_text}
Synthesize the best insights from all approaches into a single, coherent answer.
Focus on points where multiple approaches agree, and note any important unique insights."""
)
]
consensus = await self.model.generate(synthesis_messages)
return consensus
The TreeOfThought class implements multi-path exploration by generating several different reasoning approaches in parallel. The num_branches parameter controls how many paths to explore, with three being a good default that balances thoroughness with computational cost.
The reason method orchestrates the entire Tree-of-Thought process. It generates multiple branches concurrently using asyncio.gather, evaluates each branch for quality, and then selects or synthesizes a final answer based on the chosen strategy.
The create_branch_prompts method creates different prompt variations that encourage diverse reasoning. Each branch receives the same question but with different framing instructions, such as analytical thinking, creative thinking, or first-principles reasoning. This diversity increases the likelihood of finding the best solution approach.
Branch evaluation uses the language model itself as a judge, asking it to score each reasoning path on multiple quality criteria. This meta-evaluation approach leverages the model's ability to assess logical coherence and completeness. The scores are extracted from the evaluation response using regular expressions to find numeric values.
Two selection strategies are supported. The best strategy simply chooses the highest-scoring branch as the final answer. The consensus strategy synthesizes insights from all branches, creating an answer that incorporates the strongest points from each approach. Consensus is particularly valuable when different branches highlight complementary aspects of a problem.
IMPLEMENTING SELF-REFLECTION
Self-Reflection adds a critical evaluation layer where the system examines its own outputs, identifies potential issues, and refines responses iteratively. This technique significantly improves response quality by catching errors and improving clarity.
The implementation generates an initial response, evaluates it for quality and accuracy, and then refines it based on the self-critique.
# reasoning/self_reflection.py - Self-Reflection implementation
from typing import List, Dict, Any, Optional
from models.base import LanguageModel, Message
class SelfReflection:
"""Implements self-reflection and iterative refinement."""
def __init__(self, model: LanguageModel, max_iterations: int = 2):
"""Initialize self-reflection system.
Args:
model: LanguageModel for generation and reflection
max_iterations: Maximum number of refinement iterations
"""
self.model = model
self.max_iterations = max_iterations
self.reflection_system_prompt = """You are a critical evaluator who identifies weaknesses in responses.
When reviewing an answer, look for:
- Factual errors or inconsistencies
- Unclear explanations
- Missing important information
- Logical flaws
- Ways to improve clarity and completeness"""
self.refinement_system_prompt = """You are an expert at improving responses based on feedback.
Take the original response and the critique, then create an improved version that addresses all issues raised."""
async def reflect_and_refine(
self,
messages: List[Message],
initial_response: Optional[str] = None
) -> Dict[str, Any]:
"""Generate a response with self-reflection and refinement.
Args:
messages: Conversation history
initial_response: Optional pre-generated response to refine
Returns:
Dictionary with reflection history and final refined response
"""
# Generate initial response if not provided
if initial_response is None:
current_response = await self.model.generate(messages)
else:
current_response = initial_response
reflection_history = []
# Iteratively reflect and refine
for iteration in range(self.max_iterations):
# Reflect on current response
critique = await self._generate_critique(
messages,
current_response
)
reflection_history.append({
"iteration": iteration,
"response": current_response,
"critique": critique
})
# Check if response is satisfactory
if self._is_satisfactory(critique):
break
# Refine based on critique
refined_response = await self._refine_response(
messages,
current_response,
critique
)
current_response = refined_response
# Add final response to history
reflection_history.append({
"iteration": len(reflection_history),
"response": current_response,
"critique": "Final response accepted"
})
return {
"reflection_history": reflection_history,
"final_response": current_response,
"iterations_used": len(reflection_history) - 1
}
async def _generate_critique(
self,
original_messages: List[Message],
response: str
) -> str:
"""Generate a critique of the response.
Args:
original_messages: Original conversation
response: Response to critique
Returns:
Critique text identifying issues and improvements
"""
critique_messages = [
Message(role="system", content=self.reflection_system_prompt),
Message(
role="user",
content=f"""Original question: {original_messages[-1].content}
Response to evaluate:
{response}
Please provide a critical evaluation. If the response is excellent, say "SATISFACTORY".
Otherwise, identify specific issues and suggest improvements."""
)
]
critique = await self.model.generate(
critique_messages,
temperature=0.3 # Lower temperature for more focused critique
)
return critique
def _is_satisfactory(self, critique: str) -> bool:
"""Determine if a response is satisfactory based on critique.
Args:
critique: Critique text
Returns:
True if response is satisfactory, False otherwise
"""
# Check for satisfaction indicators
satisfaction_markers = [
"satisfactory",
"excellent",
"no issues",
"well done",
"no improvements needed"
]
critique_lower = critique.lower()
return any(marker in critique_lower for marker in satisfaction_markers)
async def _refine_response(
self,
original_messages: List[Message],
current_response: str,
critique: str
) -> str:
"""Refine a response based on critique.
Args:
original_messages: Original conversation
current_response: Current response to improve
critique: Critique identifying issues
Returns:
Refined response addressing the critique
"""
refinement_messages = [
Message(role="system", content=self.refinement_system_prompt),
Message(
role="user",
content=f"""Original question: {original_messages[-1].content}
Current response:
{current_response}
Critique and suggestions for improvement:
{critique}
Please provide an improved response that addresses all the issues raised in the critique."""
)
]
refined = await self.model.generate(refinement_messages)
return refined
The SelfReflection class implements an iterative refinement process. It generates an initial response, then enters a loop where it critiques the response and refines it based on the critique. The max_iterations parameter prevents infinite loops while allowing multiple rounds of improvement.
The reflect_and_refine method orchestrates the entire process. It accepts an optional initial_response parameter, which allows integration with other reasoning techniques. If no initial response is provided, it generates one using the base model. The method maintains a reflection_history that records each iteration's response and critique, providing transparency into the refinement process.
The generate_critique method creates a meta-prompt asking the model to evaluate its own output. The critique prompt specifically asks the model to identify factual errors, unclear explanations, missing information, and logical flaws. Using a lower temperature for critique generation produces more focused, analytical evaluations.
The is_satisfactory method checks whether the critique indicates the response is acceptable. It looks for keywords like satisfactory or excellent that signal no further refinement is needed. This prevents unnecessary iterations when the response is already high quality.
When refinement is needed, the refine_response method creates a prompt that includes the original question, current response, and critique. It asks the model to generate an improved version that addresses all identified issues. This targeted refinement is more effective than simply regenerating from scratch.
IMPLEMENTING PROMPT ENHANCEMENT
The Prompt Enhancement system analyzes user inputs and transforms them into more effective prompts that elicit better responses from language models. This meta-prompting technique significantly improves output quality by clarifying intent, adding context, and structuring queries optimally.
# reasoning/prompt_enhancer.py - Prompt enhancement implementation
from typing import List, Dict, Any
from models.base import LanguageModel, Message
class PromptEnhancer:
"""Enhances user prompts for better language model responses."""
def __init__(self, model: LanguageModel):
"""Initialize prompt enhancer.
Args:
model: LanguageModel for analyzing and enhancing prompts
"""
self.model = model
self.enhancement_system_prompt = """You are an expert at crafting effective prompts for language models.
When given a user's question or request, analyze it and create an enhanced version that:
1. Clarifies any ambiguity in the original question
2. Adds relevant context that helps answer the question
3. Structures the request for optimal model understanding
4. Specifies the desired format or detail level if unclear
5. Breaks complex requests into clear sub-questions if needed
Provide the enhanced prompt(s) in a clear, structured format."""
async def enhance_prompt(
self,
user_message: str,
conversation_context: List[Message] = None
) -> Dict[str, Any]:
"""Enhance a user prompt for better model performance.
Args:
user_message: Original user message to enhance
conversation_context: Optional conversation history for context
Returns:
Dictionary with original prompt, enhanced prompt(s), and analysis
"""
# Analyze the user's intent and requirements
analysis = await self._analyze_prompt(user_message, conversation_context)
# Generate enhanced prompt based on analysis
enhanced = await self._generate_enhanced_prompt(
user_message,
analysis,
conversation_context
)
return {
"original_prompt": user_message,
"analysis": analysis,
"enhanced_prompt": enhanced["primary"],
"alternative_prompts": enhanced.get("alternatives", []),
"enhancement_strategy": enhanced["strategy"]
}
async def _analyze_prompt(
self,
user_message: str,
context: List[Message] = None
) -> Dict[str, Any]:
"""Analyze a user prompt to understand intent and requirements.
Args:
user_message: User's message to analyze
context: Conversation history
Returns:
Dictionary with analysis results
"""
context_summary = ""
if context:
# Summarize relevant context
context_messages = [msg.content for msg in context[-3:]]
context_summary = f"\n\nRecent conversation context:\n" + "\n".join(context_messages)
analysis_messages = [
Message(
role="system",
content="You analyze prompts to identify intent, ambiguities, and requirements."
),
Message(
role="user",
content=f"""Analyze this user message:{context_summary}
User message: {user_message}
Identify:
1. Primary intent (what the user wants to achieve)
2. Any ambiguities or unclear aspects
3. Missing context that would help answer better
4. Complexity level (simple, moderate, complex)
5. Optimal response format (explanation, step-by-step, list, etc.)
Provide your analysis in a structured format."""
)
]
analysis_text = await self.model.generate(
analysis_messages,
temperature=0.3
)
# Parse the analysis into structured data
return self._parse_analysis(analysis_text)
def _parse_analysis(self, analysis_text: str) -> Dict[str, Any]:
"""Parse analysis text into structured data.
Args:
analysis_text: Raw analysis from model
Returns:
Structured analysis dictionary
"""
# Simple parsing based on common patterns
analysis = {
"intent": "",
"ambiguities": [],
"missing_context": [],
"complexity": "moderate",
"response_format": "explanation"
}
lines = analysis_text.split('\n')
current_section = None
for line in lines:
line = line.strip()
if not line:
continue
lower_line = line.lower()
if "intent" in lower_line or "purpose" in lower_line:
current_section = "intent"
elif "ambiguit" in lower_line or "unclear" in lower_line:
current_section = "ambiguities"
elif "missing" in lower_line or "context" in lower_line:
current_section = "missing_context"
elif "complexity" in lower_line:
current_section = "complexity"
elif "format" in lower_line:
current_section = "response_format"
elif current_section:
# Add content to current section
if current_section == "intent":
analysis["intent"] += " " + line
elif current_section in ["ambiguities", "missing_context"]:
if line.startswith('-') or line.startswith('*') or line[0].isdigit():
analysis[current_section].append(line.lstrip('-*0123456789. '))
elif current_section == "complexity":
if "simple" in lower_line:
analysis["complexity"] = "simple"
elif "complex" in lower_line:
analysis["complexity"] = "complex"
elif current_section == "response_format":
analysis["response_format"] = line.lstrip('-*0123456789. ')
return analysis
async def _generate_enhanced_prompt(
self,
original_prompt: str,
analysis: Dict[str, Any],
context: List[Message] = None
) -> Dict[str, Any]:
"""Generate enhanced version of the prompt.
Args:
original_prompt: Original user message
analysis: Analysis results
context: Conversation history
Returns:
Dictionary with enhanced prompt and alternatives
"""
enhancement_messages = [
Message(role="system", content=self.enhancement_system_prompt),
Message(
role="user",
content=f"""Original prompt: {original_prompt}
Analysis:
- Intent: {analysis['intent']}
- Ambiguities: {', '.join(analysis['ambiguities']) if analysis['ambiguities'] else 'None'}
- Missing context: {', '.join(analysis['missing_context']) if analysis['missing_context'] else 'None'}
- Complexity: {analysis['complexity']}
- Desired format: {analysis['response_format']}
Create an enhanced version of this prompt that addresses the ambiguities, adds necessary context,
and structures the request optimally. If the request is complex, consider breaking it into sub-questions."""
)
]
enhanced_text = await self.model.generate(enhancement_messages)
# Determine enhancement strategy based on complexity
if analysis['complexity'] == 'complex':
strategy = "decomposition"
# Try to extract multiple sub-prompts
alternatives = self._extract_sub_prompts(enhanced_text)
else:
strategy = "clarification"
alternatives = []
return {
"primary": enhanced_text,
"alternatives": alternatives,
"strategy": strategy
}
def _extract_sub_prompts(self, enhanced_text: str) -> List[str]:
"""Extract sub-prompts from enhanced text for complex queries.
Args:
enhanced_text: Enhanced prompt text
Returns:
List of sub-prompts
"""
sub_prompts = []
lines = enhanced_text.split('\n')
for line in lines:
line = line.strip()
# Look for numbered items or questions
if (line and (line[0].isdigit() or line.startswith('-') or
line.startswith('*')) and '?' in line):
# Clean up the line
cleaned = line.lstrip('-*0123456789. ')
if cleaned:
sub_prompts.append(cleaned)
return sub_prompts
The PromptEnhancer class implements a two-stage enhancement process. First, it analyzes the user's prompt to understand intent, identify ambiguities, and determine optimal response format. Second, it generates an enhanced version that addresses these factors.
The enhance_prompt method is the main entry point. It accepts the user message and optional conversation context, performs analysis and enhancement, and returns a comprehensive result including the original prompt, analysis, enhanced prompt, and any alternative formulations.
The analyze_prompt method creates a meta-prompt asking the model to examine the user's message and identify key characteristics. It looks for the primary intent, any unclear aspects, missing context, complexity level, and desired response format. This analysis guides the enhancement process.
The parse_analysis method converts the model's textual analysis into structured data. It looks for section headers and extracts information into appropriate fields. This structured representation makes it easier for subsequent steps to use the analysis results.
The generate_enhanced_prompt method creates the improved version based on the analysis. For simple queries, it focuses on clarification and adding context. For complex queries, it may decompose the request into multiple sub-questions that can be addressed sequentially.
The extract_sub_prompts method identifies individual questions within the enhanced text, enabling the system to break complex requests into manageable pieces. This decomposition is particularly valuable for multi-part questions that would be difficult to answer comprehensively in a single response.
IMPLEMENTING CONVERSATION MANAGEMENT
The Conversation Manager handles chat history, implementing intelligent summarization to prevent context overflow while preserving essential information. It maintains the conversation state and ensures the language model always has appropriate context.
# conversation/manager.py - Conversation management with summarization
from typing import List, Dict, Any, Optional
from models.base import LanguageModel, Message
import json
class ConversationManager:
"""Manages conversation history with intelligent summarization."""
def __init__(
self,
model: LanguageModel,
max_context_tokens: Optional[int] = None,
summary_trigger_ratio: float = 0.75
):
"""Initialize conversation manager.
Args:
model: LanguageModel for summarization
max_context_tokens: Maximum tokens to keep in context
summary_trigger_ratio: Ratio of context usage that triggers summarization
"""
self.model = model
self.max_context_tokens = max_context_tokens or model.get_context_window()
self.summary_trigger_ratio = summary_trigger_ratio
# Reserve tokens for response generation
self.reserved_tokens = 1024
self.effective_max_tokens = self.max_context_tokens - self.reserved_tokens
self.summarization_prompt = """Summarize the following conversation, preserving:
1. Key facts and information discussed
2. Important decisions or conclusions reached
3. The overall context and topic
4. Any unresolved questions or ongoing discussions
Be concise but comprehensive. The summary will be used to maintain context in an ongoing conversation."""
def get_messages_for_context(
self,
messages: List[Message],
system_message: Optional[Message] = None
) -> List[Message]:
"""Get messages that fit within context window, summarizing if needed.
Args:
messages: Full conversation history
system_message: Optional system message to include
Returns:
List of messages that fit in context window
"""
# Calculate current token usage
total_tokens = 0
if system_message:
total_tokens += self.model.count_tokens(system_message.content)
for msg in messages:
total_tokens += self.model.count_tokens(msg.content)
# Check if summarization is needed
if total_tokens <= self.effective_max_tokens * self.summary_trigger_ratio:
# No summarization needed
result = []
if system_message:
result.append(system_message)
result.extend(messages)
return result
# Summarization needed
return self._summarize_and_compress(messages, system_message)
def _summarize_and_compress(
self,
messages: List[Message],
system_message: Optional[Message]
) -> List[Message]:
"""Summarize old messages and keep recent ones.
Args:
messages: Full message history
system_message: System message to preserve
Returns:
Compressed message list with summary
"""
# Keep the most recent messages without summarization
# Typically keep last 4-6 exchanges (8-12 messages)
recent_message_count = 10
if len(messages) <= recent_message_count:
# Not enough messages to benefit from summarization
result = []
if system_message:
result.append(system_message)
result.extend(messages)
return result
# Split into old (to summarize) and recent (to keep)
old_messages = messages[:-recent_message_count]
recent_messages = messages[-recent_message_count:]
# Generate summary of old messages
summary = self._generate_summary(old_messages)
# Construct compressed message list
result = []
if system_message:
result.append(system_message)
# Add summary as a system message
result.append(Message(
role="system",
content=f"Previous conversation summary:\n{summary}"
))
# Add recent messages
result.extend(recent_messages)
return result
def _generate_summary(self, messages: List[Message]) -> str:
"""Generate a summary of conversation messages.
Args:
messages: Messages to summarize
Returns:
Summary text
"""
# Format messages for summarization
conversation_text = self._format_messages_for_summary(messages)
# Create summarization request
summary_messages = [
Message(role="system", content=self.summarization_prompt),
Message(role="user", content=conversation_text)
]
# Generate summary synchronously (we're already in an async context)
import asyncio
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
summary = loop.run_until_complete(
self.model.generate(summary_messages, temperature=0.3, max_tokens=500)
)
return summary
def _format_messages_for_summary(self, messages: List[Message]) -> str:
"""Format messages into text for summarization.
Args:
messages: Messages to format
Returns:
Formatted conversation text
"""
formatted_parts = []
for msg in messages:
role_label = msg.role.capitalize()
formatted_parts.append(f"{role_label}: {msg.content}")
return "\n\n".join(formatted_parts)
def estimate_tokens_needed(
self,
messages: List[Message],
system_message: Optional[Message] = None
) -> int:
"""Estimate total tokens needed for messages.
Args:
messages: Messages to estimate
system_message: Optional system message
Returns:
Estimated token count
"""
total = 0
if system_message:
total += self.model.count_tokens(system_message.content)
for msg in messages:
total += self.model.count_tokens(msg.content)
return total
The ConversationManager class handles the complexity of maintaining conversation context within token limits. It monitors token usage and automatically triggers summarization when the conversation grows too long.
The get_messages_for_context method is the primary interface. It takes the full conversation history and returns a version that fits within the context window. If the conversation is short enough, it returns all messages unchanged. When the conversation exceeds the trigger threshold, it invokes summarization.
The summarize_and_compress method implements the core compression logic. It splits the conversation into old messages that will be summarized and recent messages that are kept verbatim. The old messages are condensed into a summary, which is inserted as a system message. This approach preserves detailed context for recent exchanges while maintaining awareness of earlier discussion.
The generate_summary method creates the actual summary using the language model. It formats the messages into readable text and asks the model to extract key information. The summary prompt specifically requests preservation of facts, decisions, context, and unresolved questions, ensuring important information is not lost.
Token estimation helps the system make informed decisions about when summarization is needed. The estimate_tokens_needed method counts tokens across all messages, providing an accurate picture of context usage.
IMPLEMENTING SESSION STORAGE
Session Storage enables users to save conversations and resume them later. It persists conversation history, metadata, and settings to a database, creating a continuous experience across multiple interactions.
# storage/session_store.py - Session storage implementation
from typing import List, Dict, Any, Optional
from datetime import datetime
import json
import uuid
from sqlalchemy import create_engine, Column, String, DateTime, Text, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from models.base import Message
Base = declarative_base()
class SessionModel(Base):
"""Database model for conversation sessions."""
__tablename__ = 'sessions'
id = Column(String, primary_key=True)
user_id = Column(String, nullable=False, index=True)
title = Column(String, nullable=False)
created_at = Column(DateTime, nullable=False)
updated_at = Column(DateTime, nullable=False)
messages = Column(Text, nullable=False) # JSON-encoded messages
metadata = Column(Text) # JSON-encoded metadata
message_count = Column(Integer, default=0)
class SessionStore:
"""Manages persistent storage of conversation sessions."""
def __init__(self, database_url: str = "sqlite:///sessions.db"):
"""Initialize session store with database connection.
Args:
database_url: SQLAlchemy database URL
"""
self.engine = create_engine(database_url)
Base.metadata.create_all(self.engine)
self.SessionLocal = sessionmaker(bind=self.engine)
def create_session(
self,
user_id: str,
title: Optional[str] = None,
initial_messages: Optional[List[Message]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> str:
"""Create a new conversation session.
Args:
user_id: Identifier for the user
title: Optional session title
initial_messages: Optional starting messages
metadata: Optional session metadata
Returns:
Session ID
"""
session_id = str(uuid.uuid4())
now = datetime.utcnow()
messages = initial_messages or []
messages_json = json.dumps([self._message_to_dict(m) for m in messages])
metadata_json = json.dumps(metadata or {})
session = SessionModel(
id=session_id,
user_id=user_id,
title=title or f"Conversation {now.strftime('%Y-%m-%d %H:%M')}",
created_at=now,
updated_at=now,
messages=messages_json,
metadata=metadata_json,
message_count=len(messages)
)
db = self.SessionLocal()
try:
db.add(session)
db.commit()
finally:
db.close()
return session_id
def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
"""Retrieve a session by ID.
Args:
session_id: Session identifier
Returns:
Session data dictionary or None if not found
"""
db = self.SessionLocal()
try:
session = db.query(SessionModel).filter(
SessionModel.id == session_id
).first()
if not session:
return None
return self._session_to_dict(session)
finally:
db.close()
def update_session(
self,
session_id: str,
messages: Optional[List[Message]] = None,
title: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None
) -> bool:
"""Update an existing session.
Args:
session_id: Session identifier
messages: Updated message list
title: Updated title
metadata: Updated metadata
Returns:
True if update successful, False if session not found
"""
db = self.SessionLocal()
try:
session = db.query(SessionModel).filter(
SessionModel.id == session_id
).first()
if not session:
return False
if messages is not None:
session.messages = json.dumps(
[self._message_to_dict(m) for m in messages]
)
session.message_count = len(messages)
if title is not None:
session.title = title
if metadata is not None:
session.metadata = json.dumps(metadata)
session.updated_at = datetime.utcnow()
db.commit()
return True
finally:
db.close()
def list_sessions(
self,
user_id: str,
limit: int = 50,
offset: int = 0
) -> List[Dict[str, Any]]:
"""List sessions for a user.
Args:
user_id: User identifier
limit: Maximum number of sessions to return
offset: Number of sessions to skip
Returns:
List of session summaries
"""
db = self.SessionLocal()
try:
sessions = db.query(SessionModel).filter(
SessionModel.user_id == user_id
).order_by(
SessionModel.updated_at.desc()
).limit(limit).offset(offset).all()
return [self._session_to_summary(s) for s in sessions]
finally:
db.close()
def delete_session(self, session_id: str) -> bool:
"""Delete a session.
Args:
session_id: Session identifier
Returns:
True if deleted, False if not found
"""
db = self.SessionLocal()
try:
session = db.query(SessionModel).filter(
SessionModel.id == session_id
).first()
if not session:
return False
db.delete(session)
db.commit()
return True
finally:
db.close()
def _message_to_dict(self, message: Message) -> Dict[str, str]:
"""Convert Message object to dictionary.
Args:
message: Message object
Returns:
Dictionary representation
"""
return {
"role": message.role,
"content": message.content
}
def _dict_to_message(self, data: Dict[str, str]) -> Message:
"""Convert dictionary to Message object.
Args:
data: Dictionary with message data
Returns:
Message object
"""
return Message(role=data["role"], content=data["content"])
def _session_to_dict(self, session: SessionModel) -> Dict[str, Any]:
"""Convert database session to dictionary.
Args:
session: SessionModel instance
Returns:
Complete session data
"""
messages_data = json.loads(session.messages)
messages = [self._dict_to_message(m) for m in messages_data]
metadata = json.loads(session.metadata) if session.metadata else {}
return {
"id": session.id,
"user_id": session.user_id,
"title": session.title,
"created_at": session.created_at.isoformat(),
"updated_at": session.updated_at.isoformat(),
"messages": messages,
"metadata": metadata,
"message_count": session.message_count
}
def _session_to_summary(self, session: SessionModel) -> Dict[str, Any]:
"""Convert session to summary (without full messages).
Args:
session: SessionModel instance
Returns:
Session summary
"""
return {
"id": session.id,
"title": session.title,
"created_at": session.created_at.isoformat(),
"updated_at": session.updated_at.isoformat(),
"message_count": session.message_count
}
The SessionStore class provides a complete persistence layer for conversations. It uses SQLAlchemy for database abstraction, supporting SQLite for development and PostgreSQL or MySQL for production deployments.
The SessionModel defines the database schema. Each session has a unique identifier, user association, title, timestamps, and the conversation messages stored as JSON. The metadata field allows storing additional session-specific information like model settings or reasoning preferences.
The create_session method initializes a new conversation session. It generates a unique identifier, sets timestamps, and serializes the initial messages to JSON. The title defaults to a timestamp-based name but can be customized by the user.
The get_session method retrieves a complete session including all messages. It deserializes the JSON-stored messages back into Message objects, making them ready for use by the conversation manager and reasoning components.
The update_session method allows modifying an existing session. It can update messages, title, or metadata while automatically updating the timestamp. This method is called after each user interaction to persist the growing conversation.
The list_sessions method provides a paginated list of sessions for a user, ordered by most recent activity. It returns summaries without full message content to minimize data transfer, which is important when users have many saved conversations.
Session deletion is handled by delete_session, which removes a session from the database. This gives users control over their stored conversations and helps manage storage usage.
IMPLEMENTING THE REASONING ORCHESTRATOR
The Reasoning Orchestrator coordinates the different reasoning techniques, deciding when to apply Chain-of-Thought, Tree-of-Thought, or Self-Reflection based on the query characteristics and user preferences.
# reasoning/orchestrator.py - Coordinates reasoning strategies
from typing import List, Dict, Any, Optional
from models.base import LanguageModel, Message
from reasoning.chain_of_thought import ChainOfThought
from reasoning.tree_of_thought import TreeOfThought
from reasoning.self_reflection import SelfReflection
from reasoning.prompt_enhancer import PromptEnhancer
class ReasoningOrchestrator:
"""Orchestrates different reasoning strategies."""
def __init__(self, model: LanguageModel):
"""Initialize reasoning orchestrator.
Args:
model: LanguageModel to use for reasoning
"""
self.model = model
self.cot = ChainOfThought(model)
self.tot = TreeOfThought(model)
self.reflection = SelfReflection(model)
self.enhancer = PromptEnhancer(model)
async def process_query(
self,
messages: List[Message],
strategy: str = "auto",
enhance_prompt: bool = True,
show_reasoning: bool = True
) -> Dict[str, Any]:
"""Process a query using appropriate reasoning strategy.
Args:
messages: Conversation history
strategy: Reasoning strategy - 'auto', 'cot', 'tot', 'reflection', or 'simple'
enhance_prompt: Whether to enhance the user prompt
show_reasoning: Whether to include reasoning details in response
Returns:
Dictionary with response and reasoning information
"""
result = {
"strategy_used": strategy,
"prompt_enhanced": enhance_prompt,
"reasoning_visible": show_reasoning
}
# Enhance prompt if requested
if enhance_prompt and messages:
enhancement = await self.enhancer.enhance_prompt(
messages[-1].content,
messages[:-1] if len(messages) > 1 else None
)
# Replace last user message with enhanced version
enhanced_messages = messages[:-1] + [
Message(role="user", content=enhancement["enhanced_prompt"])
]
result["prompt_enhancement"] = enhancement
else:
enhanced_messages = messages
result["prompt_enhancement"] = None
# Determine strategy if auto
if strategy == "auto":
strategy = self._select_strategy(enhanced_messages)
result["strategy_used"] = strategy
# Apply selected reasoning strategy
if strategy == "cot":
reasoning_result = await self.cot.reason(
enhanced_messages,
show_reasoning=show_reasoning
)
result["response"] = reasoning_result["final_answer"]
result["reasoning_steps"] = reasoning_result.get("reasoning_steps", [])
elif strategy == "tot":
reasoning_result = await self.tot.reason(
enhanced_messages,
selection_strategy="best"
)
result["response"] = reasoning_result["final_answer"]
result["branches"] = reasoning_result.get("branches", []) if show_reasoning else []
elif strategy == "reflection":
reasoning_result = await self.reflection.reflect_and_refine(
enhanced_messages
)
result["response"] = reasoning_result["final_response"]
result["reflection_history"] = reasoning_result.get("reflection_history", []) if show_reasoning else []
elif strategy == "simple":
# Direct generation without special reasoning
response = await self.model.generate(enhanced_messages)
result["response"] = response
else:
raise ValueError(f"Unknown strategy: {strategy}")
return result
def _select_strategy(self, messages: List[Message]) -> str:
"""Automatically select appropriate reasoning strategy.
Args:
messages: Conversation messages
Returns:
Selected strategy name
"""
if not messages:
return "simple"
last_message = messages[-1].content.lower()
# Heuristics for strategy selection
# Tree-of-Thought for complex problems with multiple approaches
tot_indicators = [
"compare", "evaluate", "which is better", "pros and cons",
"different approaches", "alternatives", "options"
]
if any(indicator in last_message for indicator in tot_indicators):
return "tot"
# Chain-of-Thought for step-by-step reasoning
cot_indicators = [
"how", "why", "explain", "step by step", "calculate",
"solve", "prove", "derive"
]
if any(indicator in last_message for indicator in cot_indicators):
return "cot"
# Self-Reflection for tasks requiring accuracy
reflection_indicators = [
"accurate", "precise", "correct", "verify", "check",
"important", "critical"
]
if any(indicator in last_message for indicator in reflection_indicators):
return "reflection"
# Default to Chain-of-Thought for general queries
return "cot"
async def stream_response(
self,
messages: List[Message],
strategy: str = "simple",
enhance_prompt: bool = False
):
"""Stream a response (simplified, no complex reasoning).
Args:
messages: Conversation history
strategy: Must be 'simple' for streaming
enhance_prompt: Whether to enhance prompt
Yields:
Response chunks
"""
if strategy != "simple":
raise ValueError("Only 'simple' strategy supports streaming")
# Optionally enhance prompt
if enhance_prompt and messages:
enhancement = await self.enhancer.enhance_prompt(
messages[-1].content,
messages[:-1] if len(messages) > 1 else None
)
enhanced_messages = messages[:-1] + [
Message(role="user", content=enhancement["enhanced_prompt"])
]
else:
enhanced_messages = messages
# Stream response
async for chunk in self.model.stream_generate(enhanced_messages):
yield chunk
The ReasoningOrchestrator provides a unified interface for all reasoning capabilities. It initializes instances of each reasoning component and coordinates their use based on query characteristics and user preferences.
The process_query method is the main entry point. It accepts the conversation history, desired strategy, and configuration options. When prompt enhancement is enabled, it first improves the user's query before applying reasoning. The strategy parameter can be set explicitly or left as auto for automatic selection.
The automatic strategy selection uses heuristics based on the query content. Questions asking for comparisons or evaluations trigger Tree-of-Thought reasoning. Questions asking how or why something works trigger Chain-of-Thought. Questions emphasizing accuracy trigger Self-Reflection. This intelligent selection ensures appropriate reasoning is applied without requiring users to understand the different techniques.
The stream_response method provides streaming for simple queries where real-time response display is more important than advanced reasoning. Complex reasoning strategies like Tree-of-Thought cannot be streamed effectively because they require generating and evaluating multiple complete responses before selecting an answer.
The result dictionary returned by process_query includes not only the final response but also metadata about which strategy was used, whether the prompt was enhanced, and the intermediate reasoning steps if requested. This transparency helps users understand how the system arrived at its answer.
IMPLEMENTING THE WEB SERVER BACKEND
The web server backend provides HTTP and WebSocket endpoints for the frontend to interact with the chatbot system. It uses FastAPI for its excellent async support, automatic API documentation, and WebSocket capabilities.
# server/main.py - FastAPI web server implementation
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, FileResponse
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import json
import asyncio
from datetime import datetime
from models.factory import ModelFactory
from models.base import Message
from conversation.manager import ConversationManager
from reasoning.orchestrator import ReasoningOrchestrator
from storage.session_store import SessionStore
# Pydantic models for API requests/responses
class ChatRequest(BaseModel):
"""Request model for chat endpoint."""
message: str
session_id: Optional[str] = None
strategy: str = "auto"
enhance_prompt: bool = True
show_reasoning: bool = True
class ChatResponse(BaseModel):
"""Response model for chat endpoint."""
response: str
session_id: str
strategy_used: str
reasoning_steps: Optional[List[str]] = None
branches: Optional[List[Dict[str, Any]]] = None
reflection_history: Optional[List[Dict[str, Any]]] = None
class SessionCreate(BaseModel):
"""Request model for creating a session."""
title: Optional[str] = None
class SessionResponse(BaseModel):
"""Response model for session operations."""
session_id: str
title: str
created_at: str
updated_at: str
message_count: int
# Initialize FastAPI application
app = FastAPI(
title="Advanced LLM Chatbot",
description="Chatbot with Chain-of-Thought, Tree-of-Thought, and Self-Reflection",
version="1.0.0"
)
# Global instances (in production, use dependency injection)
model = None
conversation_manager = None
reasoning_orchestrator = None
session_store = None
active_sessions = {} # In-memory cache of active conversations
def initialize_system(config: Dict[str, Any]):
"""Initialize the chatbot system with configuration.
Args:
config: Configuration dictionary
"""
global model, conversation_manager, reasoning_orchestrator, session_store
# Create language model
model = ModelFactory.create_from_settings(config["model"])
# Initialize components
conversation_manager = ConversationManager(model)
reasoning_orchestrator = ReasoningOrchestrator(model)
session_store = SessionStore(config.get("database_url", "sqlite:///sessions.db"))
@app.on_event("startup")
async def startup_event():
"""Initialize system on server startup."""
# Load configuration from environment or config file
config = {
"model": {
"type": "remote", # or "local"
"model_name": "gpt-3.5-turbo",
"api_key": "your-api-key-here", # Load from environment
"temperature": 0.7,
"max_tokens": 2048
},
"database_url": "sqlite:///sessions.db"
}
initialize_system(config)
@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""Handle chat requests.
Args:
request: ChatRequest with message and options
Returns:
ChatResponse with generated response
"""
# Get or create session
if request.session_id:
session_data = session_store.get_session(request.session_id)
if not session_data:
raise HTTPException(status_code=404, detail="Session not found")
messages = session_data["messages"]
session_id = request.session_id
else:
# Create new session
session_id = session_store.create_session(
user_id="default_user", # In production, use actual user ID
title=f"Chat {datetime.now().strftime('%Y-%m-%d %H:%M')}"
)
messages = []
# Add user message
messages.append(Message(role="user", content=request.message))
# Get context-appropriate messages
context_messages = conversation_manager.get_messages_for_context(messages)
# Process with reasoning
result = await reasoning_orchestrator.process_query(
context_messages,
strategy=request.strategy,
enhance_prompt=request.enhance_prompt,
show_reasoning=request.show_reasoning
)
# Add assistant response to messages
messages.append(Message(role="assistant", content=result["response"]))
# Update session
session_store.update_session(session_id, messages=messages)
# Build response
return ChatResponse(
response=result["response"],
session_id=session_id,
strategy_used=result["strategy_used"],
reasoning_steps=result.get("reasoning_steps"),
branches=result.get("branches"),
reflection_history=result.get("reflection_history")
)
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
"""WebSocket endpoint for streaming chat.
Args:
websocket: WebSocket connection
"""
await websocket.accept()
try:
while True:
# Receive message from client
data = await websocket.receive_json()
message_text = data.get("message")
session_id = data.get("session_id")
# Get or create session
if session_id:
session_data = session_store.get_session(session_id)
if session_data:
messages = session_data["messages"]
else:
session_id = session_store.create_session(user_id="default_user")
messages = []
else:
session_id = session_store.create_session(user_id="default_user")
messages = []
# Add user message
messages.append(Message(role="user", content=message_text))
# Get context
context_messages = conversation_manager.get_messages_for_context(messages)
# Send session ID to client
await websocket.send_json({
"type": "session_id",
"session_id": session_id
})
# Stream response
response_text = ""
async for chunk in reasoning_orchestrator.stream_response(
context_messages,
strategy="simple"
):
response_text += chunk
await websocket.send_json({
"type": "chunk",
"content": chunk
})
# Send completion signal
await websocket.send_json({"type": "complete"})
# Update session with complete response
messages.append(Message(role="assistant", content=response_text))
session_store.update_session(session_id, messages=messages)
except WebSocketDisconnect:
pass
@app.post("/api/sessions", response_model=SessionResponse)
async def create_session(request: SessionCreate):
"""Create a new chat session.
Args:
request: SessionCreate with optional title
Returns:
SessionResponse with session details
"""
session_id = session_store.create_session(
user_id="default_user",
title=request.title
)
session_data = session_store.get_session(session_id)
return SessionResponse(
session_id=session_data["id"],
title=session_data["title"],
created_at=session_data["created_at"],
updated_at=session_data["updated_at"],
message_count=session_data["message_count"]
)
@app.get("/api/sessions", response_model=List[SessionResponse])
async def list_sessions(limit: int = 50, offset: int = 0):
"""List chat sessions.
Args:
limit: Maximum sessions to return
offset: Number of sessions to skip
Returns:
List of SessionResponse objects
"""
sessions = session_store.list_sessions(
user_id="default_user",
limit=limit,
offset=offset
)
return [
SessionResponse(
session_id=s["id"],
title=s["title"],
created_at=s["created_at"],
updated_at=s["updated_at"],
message_count=s["message_count"]
)
for s in sessions
]
@app.get("/api/sessions/{session_id}")
async def get_session(session_id: str):
"""Get a specific session with full message history.
Args:
session_id: Session identifier
Returns:
Complete session data
"""
session_data = session_store.get_session(session_id)
if not session_data:
raise HTTPException(status_code=404, detail="Session not found")
# Convert messages to dictionaries for JSON response
messages_dict = [
{"role": msg.role, "content": msg.content}
for msg in session_data["messages"]
]
return {
"session_id": session_data["id"],
"title": session_data["title"],
"created_at": session_data["created_at"],
"updated_at": session_data["updated_at"],
"messages": messages_dict,
"message_count": session_data["message_count"]
}
@app.delete("/api/sessions/{session_id}")
async def delete_session(session_id: str):
"""Delete a session.
Args:
session_id: Session identifier
Returns:
Success message
"""
success = session_store.delete_session(session_id)
if not success:
raise HTTPException(status_code=404, detail="Session not found")
return {"message": "Session deleted successfully"}
# Serve static files (frontend)
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/")
async def root():
"""Serve the main application page."""
return FileResponse("static/index.html")
The FastAPI server provides both REST endpoints for standard request-response interactions and WebSocket endpoints for streaming responses. The application initializes the language model, conversation manager, reasoning orchestrator, and session store on startup.
The chat endpoint handles complete request-response cycles with full reasoning capabilities. It accepts a message, optional session ID, and configuration options. It retrieves or creates a session, processes the message through the reasoning orchestrator, and returns the response with reasoning details.
The WebSocket endpoint enables real-time streaming of responses. It accepts messages over the WebSocket connection and streams back response chunks as they are generated. This provides a more interactive experience for users, though it only supports simple generation without complex reasoning.
Session management endpoints allow creating, listing, retrieving, and deleting sessions. The create endpoint initializes a new conversation. The list endpoint returns summaries of all sessions for a user. The get endpoint retrieves full session data including all messages. The delete endpoint removes a session from storage.
The server serves static files from a static directory, which will contain the frontend HTML, CSS, and JavaScript. The root endpoint returns the main application page.
IMPLEMENTING THE WEB FRONTEND
The web frontend provides an intuitive, visually appealing interface for interacting with the chatbot. It displays messages, reasoning steps, and session management in a clean, responsive layout.
<!-- static/index.html - Main application page -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Advanced LLM Chatbot</title>
<link rel="stylesheet" href="/static/styles.css">
</head>
<body>
<div class="app-container">
<!-- Sidebar for session management -->
<aside class="sidebar">
<div class="sidebar-header">
<h2>Conversations</h2>
<button id="new-chat-btn" class="btn-primary">New Chat</button>
</div>
<div class="session-list" id="session-list">
<!-- Sessions will be loaded here -->
</div>
</aside>
<!-- Main chat area -->
<main class="chat-container">
<div class="chat-header">
<h1>Advanced LLM Chatbot</h1>
<div class="settings-toggle">
<button id="settings-btn">Settings</button>
</div>
</div>
<!-- Settings panel -->
<div class="settings-panel" id="settings-panel" style="display: none;">
<div class="setting-group">
<label for="strategy-select">Reasoning Strategy:</label>
<select id="strategy-select">
<option value="auto">Auto (Recommended)</option>
<option value="simple">Simple</option>
<option value="cot">Chain-of-Thought</option>
<option value="tot">Tree-of-Thought</option>
<option value="reflection">Self-Reflection</option>
</select>
</div>
<div class="setting-group">
<label>
<input type="checkbox" id="enhance-prompt-check" checked>
Enhance Prompts
</label>
</div>
<div class="setting-group">
<label>
<input type="checkbox" id="show-reasoning-check" checked>
Show Reasoning Steps
</label>
</div>
</div>
<!-- Messages area -->
<div class="messages-container" id="messages-container">
<div class="welcome-message">
<h2>Welcome to Advanced LLM Chatbot</h2>
<p>This chatbot uses advanced reasoning techniques including Chain-of-Thought,
Tree-of-Thought, and Self-Reflection to provide high-quality responses.</p>
<p>Start a conversation by typing a message below.</p>
</div>
</div>
<!-- Input area -->
<div class="input-container">
<textarea
id="message-input"
placeholder="Type your message here..."
rows="3"
></textarea>
<button id="send-btn" class="btn-primary">Send</button>
</div>
</main>
</div>
<script src="/static/app.js"></script>
</body>
</html>
The HTML structure divides the interface into a sidebar for session management and a main area for the chat. The sidebar contains a list of previous conversations and a button to start new chats. The main area includes the chat header, settings panel, message display, and input controls.
The settings panel allows users to configure the reasoning strategy, enable or disable prompt enhancement, and control whether reasoning steps are displayed. These settings affect how the backend processes queries.
The messages container displays the conversation history, including both user and assistant messages. It also shows reasoning steps, branches, and reflection history when available, providing transparency into the system's decision-making process.
The input area provides a textarea for composing messages and a send button to submit them. The textarea automatically expands as users type longer messages.
/* static/styles.css - Application styles */
:root {
--primary-color: #2563eb;
--primary-hover: #1d4ed8;
--background: #f9fafb;
--surface: #ffffff;
--text-primary: #111827;
--text-secondary: #6b7280;
--border: #e5e7eb;
--success: #10b981;
--warning: #f59e0b;
--error: #ef4444;
--shadow: 0 1px 3px rgba(0, 0, 0, 0.1);
--shadow-lg: 0 10px 15px rgba(0, 0, 0, 0.1);
}
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen, Ubuntu, Cantarell, sans-serif;
background-color: var(--background);
color: var(--text-primary);
line-height: 1.6;
}
.app-container {
display: flex;
height: 100vh;
overflow: hidden;
}
/* Sidebar styles */
.sidebar {
width: 280px;
background-color: var(--surface);
border-right: 1px solid var(--border);
display: flex;
flex-direction: column;
}
.sidebar-header {
padding: 20px;
border-bottom: 1px solid var(--border);
}
.sidebar-header h2 {
font-size: 18px;
margin-bottom: 12px;
color: var(--text-primary);
}
.session-list {
flex: 1;
overflow-y: auto;
padding: 10px;
}
.session-item {
padding: 12px;
margin-bottom: 8px;
background-color: var(--background);
border-radius: 8px;
cursor: pointer;
transition: background-color 0.2s;
}
.session-item:hover {
background-color: var(--border);
}
.session-item.active {
background-color: var(--primary-color);
color: white;
}
.session-item-title {
font-weight: 500;
margin-bottom: 4px;
}
.session-item-meta {
font-size: 12px;
color: var(--text-secondary);
}
.session-item.active .session-item-meta {
color: rgba(255, 255, 255, 0.8);
}
/* Main chat area */
.chat-container {
flex: 1;
display: flex;
flex-direction: column;
background-color: var(--surface);
}
.chat-header {
padding: 20px;
border-bottom: 1px solid var(--border);
display: flex;
justify-content: space-between;
align-items: center;
}
.chat-header h1 {
font-size: 24px;
}
.settings-panel {
padding: 20px;
background-color: var(--background);
border-bottom: 1px solid var(--border);
}
.setting-group {
margin-bottom: 15px;
}
.setting-group label {
display: block;
margin-bottom: 5px;
font-weight: 500;
}
.setting-group select {
width: 100%;
padding: 8px;
border: 1px solid var(--border);
border-radius: 6px;
background-color: var(--surface);
}
.setting-group input[type="checkbox"] {
margin-right: 8px;
}
/* Messages area */
.messages-container {
flex: 1;
overflow-y: auto;
padding: 20px;
}
.welcome-message {
text-align: center;
padding: 40px 20px;
color: var(--text-secondary);
}
.welcome-message h2 {
margin-bottom: 15px;
color: var(--text-primary);
}
.message {
margin-bottom: 20px;
display: flex;
flex-direction: column;
}
.message-header {
font-weight: 600;
margin-bottom: 8px;
color: var(--text-secondary);
}
.message-content {
padding: 15px;
border-radius: 8px;
background-color: var(--background);
}
.message.user .message-content {
background-color: var(--primary-color);
color: white;
margin-left: 20%;
}
.message.assistant .message-content {
background-color: var(--background);
margin-right: 20%;
}
.reasoning-section {
margin-top: 15px;
padding: 15px;
background-color: rgba(37, 99, 235, 0.05);
border-left: 3px solid var(--primary-color);
border-radius: 4px;
}
.reasoning-section h4 {
margin-bottom: 10px;
color: var(--primary-color);
}
.reasoning-step {
padding: 8px 0;
border-bottom: 1px solid var(--border);
}
.reasoning-step:last-child {
border-bottom: none;
}
.branch-item {
margin-bottom: 15px;
padding: 12px;
background-color: var(--surface);
border-radius: 6px;
}
.branch-header {
font-weight: 600;
margin-bottom: 8px;
display: flex;
justify-content: space-between;
}
.branch-score {
color: var(--primary-color);
}
/* Input area */
.input-container {
padding: 20px;
border-top: 1px solid var(--border);
display: flex;
gap: 10px;
}
#message-input {
flex: 1;
padding: 12px;
border: 1px solid var(--border);
border-radius: 8px;
resize: vertical;
font-family: inherit;
font-size: 14px;
}
#message-input:focus {
outline: none;
border-color: var(--primary-color);
}
/* Buttons */
.btn-primary {
padding: 10px 20px;
background-color: var(--primary-color);
color: white;
border: none;
border-radius: 8px;
cursor: pointer;
font-weight: 500;
transition: background-color 0.2s;
}
.btn-primary:hover {
background-color: var(--primary-hover);
}
.btn-primary:disabled {
background-color: var(--text-secondary);
cursor: not-allowed;
}
button {
font-family: inherit;
font-size: 14px;
}
/* Loading indicator */
.loading-indicator {
display: inline-block;
width: 20px;
height: 20px;
border: 3px solid var(--border);
border-top-color: var(--primary-color);
border-radius: 50%;
animation: spin 1s linear infinite;
}
@keyframes spin {
to { transform: rotate(360deg); }
}
/* Scrollbar styling */
::-webkit-scrollbar {
width: 8px;
}
::-webkit-scrollbar-track {
background: var(--background);
}
::-webkit-scrollbar-thumb {
background: var(--border);
border-radius: 4px;
}
::-webkit-scrollbar-thumb:hover {
background: var(--text-secondary);
}
The CSS provides a clean, modern design with a professional color scheme. The layout uses flexbox for responsive arrangement of components. The sidebar has a fixed width while the main chat area expands to fill available space.
Message styling differentiates user and assistant messages through positioning and color. User messages appear on the right with a blue background, while assistant messages appear on the left with a light gray background. This visual distinction makes conversations easy to follow.
The reasoning sections use subtle background colors and border accents to highlight them without overwhelming the main content. Each reasoning step or branch is clearly separated for readability.
The input area provides a comfortable typing experience with a resizable textarea and prominent send button. The button changes color on hover to provide visual feedback.
// static/app.js - Frontend JavaScript application
class ChatApp {
constructor() {
this.currentSessionId = null;
this.settings = {
strategy: 'auto',
enhancePrompt: true,
showReasoning: true
};
this.initializeElements();
this.attachEventListeners();
this.loadSessions();
}
initializeElements() {
this.messagesContainer = document.getElementById('messages-container');
this.messageInput = document.getElementById('message-input');
this.sendBtn = document.getElementById('send-btn');
this.sessionList = document.getElementById('session-list');
this.newChatBtn = document.getElementById('new-chat-btn');
this.settingsBtn = document.getElementById('settings-btn');
this.settingsPanel = document.getElementById('settings-panel');
this.strategySelect = document.getElementById('strategy-select');
this.enhancePromptCheck = document.getElementById('enhance-prompt-check');
this.showReasoningCheck = document.getElementById('show-reasoning-check');
}
attachEventListeners() {
this.sendBtn.addEventListener('click', () => this.sendMessage());
this.messageInput.addEventListener('keydown', (e) => {
if (e.key === 'Enter' && !e.shiftKey) {
e.preventDefault();
this.sendMessage();
}
});
this.newChatBtn.addEventListener('click', () => this.createNewChat());
this.settingsBtn.addEventListener('click', () => this.toggleSettings());
this.strategySelect.addEventListener('change', (e) => {
this.settings.strategy = e.target.value;
});
this.enhancePromptCheck.addEventListener('change', (e) => {
this.settings.enhancePrompt = e.target.checked;
});
this.showReasoningCheck.addEventListener('change', (e) => {
this.settings.showReasoning = e.target.checked;
});
}
toggleSettings() {
const panel = this.settingsPanel;
panel.style.display = panel.style.display === 'none' ? 'block' : 'none';
}
async loadSessions() {
try {
const response = await fetch('/api/sessions');
const sessions = await response.json();
this.renderSessionList(sessions);
} catch (error) {
console.error('Failed to load sessions:', error);
}
}
renderSessionList(sessions) {
this.sessionList.innerHTML = '';
sessions.forEach(session => {
const sessionItem = document.createElement('div');
sessionItem.className = 'session-item';
if (session.session_id === this.currentSessionId) {
sessionItem.classList.add('active');
}
sessionItem.innerHTML = `
<div class="session-item-title">${this.escapeHtml(session.title)}</div>
<div class="session-item-meta">
${session.message_count} messages • ${this.formatDate(session.updated_at)}
</div>
`;
sessionItem.addEventListener('click', () => {
this.loadSession(session.session_id);
});
this.sessionList.appendChild(sessionItem);
});
}
async loadSession(sessionId) {
try {
const response = await fetch(`/api/sessions/${sessionId}`);
const sessionData = await response.json();
this.currentSessionId = sessionId;
this.renderMessages(sessionData.messages);
this.loadSessions(); // Refresh to update active state
} catch (error) {
console.error('Failed to load session:', error);
}
}
renderMessages(messages) {
// Clear welcome message
this.messagesContainer.innerHTML = '';
messages.forEach(msg => {
this.addMessageToUI(msg.role, msg.content);
});
this.scrollToBottom();
}
async createNewChat() {
try {
const response = await fetch('/api/sessions', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({})
});
const session = await response.json();
this.currentSessionId = session.session_id;
// Clear messages
this.messagesContainer.innerHTML = `
<div class="welcome-message">
<h2>New Conversation</h2>
<p>Start by typing a message below.</p>
</div>
`;
this.loadSessions();
} catch (error) {
console.error('Failed to create new chat:', error);
}
}
async sendMessage() {
const messageText = this.messageInput.value.trim();
if (!messageText) return;
// Disable input while processing
this.messageInput.disabled = true;
this.sendBtn.disabled = true;
// Add user message to UI
this.addMessageToUI('user', messageText);
this.messageInput.value = '';
try {
const response = await fetch('/api/chat', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({
message: messageText,
session_id: this.currentSessionId,
strategy: this.settings.strategy,
enhance_prompt: this.settings.enhancePrompt,
show_reasoning: this.settings.showReasoning
})
});
const data = await response.json();
// Update session ID if this was a new chat
if (!this.currentSessionId) {
this.currentSessionId = data.session_id;
this.loadSessions();
}
// Add assistant response to UI
this.addMessageToUI('assistant', data.response, {
reasoningSteps: data.reasoning_steps,
branches: data.branches,
reflectionHistory: data.reflection_history,
strategyUsed: data.strategy_used
});
} catch (error) {
console.error('Failed to send message:', error);
this.addMessageToUI('assistant', 'Sorry, an error occurred. Please try again.');
} finally {
// Re-enable input
this.messageInput.disabled = false;
this.sendBtn.disabled = false;
this.messageInput.focus();
}
}
addMessageToUI(role, content, metadata = {}) {
const messageDiv = document.createElement('div');
messageDiv.className = `message ${role}`;
let html = `
<div class="message-header">${role === 'user' ? 'You' : 'Assistant'}</div>
<div class="message-content">${this.formatContent(content)}</div>
`;
// Add reasoning sections if available
if (metadata.reasoningSteps && metadata.reasoningSteps.length > 0) {
html += this.renderReasoningSteps(metadata.reasoningSteps);
}
if (metadata.branches && metadata.branches.length > 0) {
html += this.renderBranches(metadata.branches);
}
if (metadata.reflectionHistory && metadata.reflectionHistory.length > 0) {
html += this.renderReflectionHistory(metadata.reflectionHistory);
}
if (metadata.strategyUsed) {
html += `<div class="strategy-badge">Strategy: ${metadata.strategyUsed}</div>`;
}
messageDiv.innerHTML = html;
this.messagesContainer.appendChild(messageDiv);
this.scrollToBottom();
}
renderReasoningSteps(steps) {
let html = '<div class="reasoning-section"><h4>Reasoning Steps:</h4>';
steps.forEach((step, index) => {
html += `<div class="reasoning-step">${index + 1}. ${this.escapeHtml(step)}</div>`;
});
html += '</div>';
return html;
}
renderBranches(branches) {
let html = '<div class="reasoning-section"><h4>Explored Approaches:</h4>';
branches.forEach((branch, index) => {
html += `
<div class="branch-item">
<div class="branch-header">
<span>Approach ${index + 1}</span>
<span class="branch-score">Score: ${branch.score}/10</span>
</div>
<div>${this.formatContent(branch.content)}</div>
</div>
`;
});
html += '</div>';
return html;
}
renderReflectionHistory(history) {
let html = '<div class="reasoning-section"><h4>Reflection Process:</h4>';
history.forEach((item, index) => {
if (item.critique !== 'Final response accepted') {
html += `
<div class="reflection-item">
<strong>Iteration ${item.iteration + 1}:</strong>
<div>Critique: ${this.escapeHtml(item.critique)}</div>
</div>
`;
}
});
html += '</div>';
return html;
}
formatContent(content) {
// Simple formatting: preserve line breaks
return this.escapeHtml(content).replace(/\n/g, '<br>');
}
escapeHtml(text) {
const div = document.createElement('div');
div.textContent = text;
return div.innerHTML;
}
formatDate(dateString) {
const date = new Date(dateString);
const now = new Date();
const diffMs = now - date;
const diffMins = Math.floor(diffMs / 60000);
if (diffMins < 1) return 'Just now';
if (diffMins < 60) return `${diffMins}m ago`;
const diffHours = Math.floor(diffMins / 60);
if (diffHours < 24) return `${diffHours}h ago`;
const diffDays = Math.floor(diffHours / 24);
if (diffDays < 7) return `${diffDays}d ago`;
return date.toLocaleDateString();
}
scrollToBottom() {
this.messagesContainer.scrollTop = this.messagesContainer.scrollHeight;
}
}
// Initialize application when DOM is ready
document.addEventListener('DOMContentLoaded', () => {
new ChatApp();
});
The JavaScript application manages the frontend state and interactions. The ChatApp class encapsulates all functionality, maintaining the current session, settings, and UI state.
The initialization process sets up references to DOM elements and attaches event listeners for user interactions. The settings are stored in memory and applied to each request.
The sendMessage method handles the core chat interaction. It disables the input while processing, adds the user message to the UI, sends the request to the backend, and displays the response with any reasoning information. Error handling ensures the interface remains responsive even if requests fail.
Session management methods handle creating new chats, loading existing sessions, and displaying the session list. The renderSessionList method creates clickable session items that load the corresponding conversation when selected.
The message rendering methods format content and add reasoning visualizations. The renderReasoningSteps method displays Chain-of-Thought steps as a numbered list. The renderBranches method shows Tree-of-Thought exploration with scores. The renderReflectionHistory method presents the iterative refinement process.
Utility methods handle HTML escaping to prevent XSS vulnerabilities, date formatting for human-readable timestamps, and automatic scrolling to keep the latest messages visible.
COMPLETE RUNNING EXAMPLE
The following is a complete, production-ready implementation that integrates all components into a working system. This example can be deployed and used immediately with appropriate configuration.
# complete_chatbot.py - Complete production-ready chatbot system
import os
import asyncio
import json
import uuid
import re
from abc import ABC, abstractmethod
from typing import List, Dict, Any, AsyncGenerator, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
# Third-party imports
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Depends
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, String, DateTime, Text, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import uvicorn
# Try to import optional dependencies
try:
from llama_cpp import Llama
LLAMA_CPP_AVAILABLE = True
except ImportError:
LLAMA_CPP_AVAILABLE = False
try:
from openai import AsyncOpenAI
import tiktoken
OPENAI_AVAILABLE = True
except ImportError:
OPENAI_AVAILABLE = False
# =============================================================================
# MODELS MODULE - Language model abstractions and implementations
# =============================================================================
@dataclass
class ModelConfig:
"""Configuration for a language model instance."""
model_name: str
temperature: float = 0.7
max_tokens: int = 2048
top_p: float = 0.9
context_window: int = 4096
gpu_layers: Optional[int] = None
api_key: Optional[str] = None
@dataclass
class Message:
"""Represents a single message in a conversation."""
role: str
content: str
def to_dict(self) -> Dict[str, str]:
return {"role": self.role, "content": self.content}
class LanguageModel(ABC):
"""Abstract base class for all language model implementations."""
def __init__(self, config: ModelConfig):
self.config = config
self.model_name = config.model_name
@abstractmethod
async def generate(self, messages: List[Message], **kwargs) -> str:
pass
@abstractmethod
async def stream_generate(self, messages: List[Message], **kwargs) -> AsyncGenerator[str, None]:
pass
@abstractmethod
def count_tokens(self, text: str) -> int:
pass
@abstractmethod
def get_context_window(self) -> int:
pass
class LocalLanguageModel(LanguageModel):
"""Implementation for local models using llama.cpp."""
def __init__(self, config: ModelConfig, model_path: str):
if not LLAMA_CPP_AVAILABLE:
raise ImportError("llama-cpp-python is required for local models")
super().__init__(config)
self.model_path = model_path
n_gpu_layers = config.gpu_layers if config.gpu_layers is not None else -1
self.model = Llama(
model_path=model_path,
n_ctx=config.context_window,
n_gpu_layers=n_gpu_layers,
verbose=False
)
def _format_messages(self, messages: List[Message]) -> str:
prompt_parts = []
for message in messages:
if message.role == "system":
prompt_parts.append(f"System: {message.content}")
elif message.role == "user":
prompt_parts.append(f"User: {message.content}")
elif message.role == "assistant":
prompt_parts.append(f"Assistant: {message.content}")
prompt_parts.append("Assistant:")
return "\n\n".join(prompt_parts)
async def generate(self, messages: List[Message], **kwargs) -> str:
prompt = self._format_messages(messages)
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stop=["User:", "\n\nUser:", "\n\nSystem:"],
echo=False
)
)
return response['choices'][0]['text'].strip()
async def stream_generate(self, messages: List[Message], **kwargs) -> AsyncGenerator[str, None]:
prompt = self._format_messages(messages)
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
stream = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stop=["User:", "\n\nUser:", "\n\nSystem:"],
stream=True,
echo=False
)
for output in stream:
token = output['choices'][0]['text']
yield token
await asyncio.sleep(0.001)
def count_tokens(self, text: str) -> int:
tokens = self.model.tokenize(text.encode('utf-8'))
return len(tokens)
def get_context_window(self) -> int:
return self.config.context_window
class RemoteLanguageModel(LanguageModel):
"""Implementation for remote models via OpenAI API."""
def __init__(self, config: ModelConfig):
if not OPENAI_AVAILABLE:
raise ImportError("openai and tiktoken are required for remote models")
super().__init__(config)
if not config.api_key:
raise ValueError("API key is required for remote models")
self.client = AsyncOpenAI(api_key=config.api_key)
try:
self.encoding = tiktoken.encoding_for_model(config.model_name)
except KeyError:
self.encoding = tiktoken.get_encoding("cl100k_base")
async def generate(self, messages: List[Message], **kwargs) -> str:
api_messages = [msg.to_dict() for msg in messages]
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
response = await self.client.chat.completions.create(
model=self.config.model_name,
messages=api_messages,
temperature=temperature,
max_tokens=max_tokens,
top_p=top_p
)
return response.choices[0].message.content
async def stream_generate(self, messages: List[Message], **kwargs) -> AsyncGenerator[str, None]:
api_messages = [msg.to_dict() for msg in messages]
temperature = kwargs.get('temperature', self.config.temperature)
max_tokens = kwargs.get('max_tokens', self.config.max_tokens)
top_p = kwargs.get('top_p', self.config.top_p)
stream = await self.client.chat.completions.create(
model=self.config.model_name,
messages=api_messages,
temperature=temperature,
max_tokens=max_tokens,
top_p=top_p,
stream=True
)
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
yield chunk.choices[0].delta.content
def count_tokens(self, text: str) -> int:
tokens = self.encoding.encode(text)
return len(tokens)
def get_context_window(self) -> int:
context_windows = {
"gpt-4": 8192,
"gpt-4-32k": 32768,
"gpt-4-turbo-preview": 128000,
"gpt-3.5-turbo": 4096,
"gpt-3.5-turbo-16k": 16384,
}
return context_windows.get(self.config.model_name, self.config.context_window)
class ModelFactory:
"""Factory for creating language model instances."""
@staticmethod
def create_model(model_type: str, config: ModelConfig, model_path: str = None) -> LanguageModel:
if model_type == "local":
if not model_path:
raise ValueError("model_path is required for local models")
return LocalLanguageModel(config, model_path)
elif model_type == "remote":
return RemoteLanguageModel(config)
else:
raise ValueError(f"Unknown model type: {model_type}")
@staticmethod
def create_from_settings(settings: dict) -> LanguageModel:
model_type = settings.get("type", "remote")
config = ModelConfig(
model_name=settings.get("model_name", "gpt-3.5-turbo"),
temperature=settings.get("temperature", 0.7),
max_tokens=settings.get("max_tokens", 2048),
top_p=settings.get("top_p", 0.9),
context_window=settings.get("context_window", 4096),
gpu_layers=settings.get("gpu_layers"),
api_key=settings.get("api_key")
)
model_path = settings.get("model_path")
return ModelFactory.create_model(model_type, config, model_path)
# =============================================================================
# REASONING MODULE - Advanced reasoning implementations
# =============================================================================
class ChainOfThought:
"""Implements Chain-of-Thought reasoning."""
def __init__(self, model: LanguageModel):
self.model = model
self.cot_system_prompt = """You are a helpful assistant that thinks step-by-step.
When answering questions, break down your reasoning into clear steps:
1. Analyze what the question is asking
2. Identify key information needed
3. Work through the problem step-by-step
4. Arrive at a conclusion based on your reasoning
Show your thought process clearly."""
async def reason(self, messages: List[Message], show_reasoning: bool = True) -> Dict[str, Any]:
cot_messages = [Message(role="system", content=self.cot_system_prompt)]
for msg in messages:
if msg.role != "system":
cot_messages.append(msg)
if cot_messages[-1].role == "user":
original_content = cot_messages[-1].content
cot_messages[-1] = Message(
role="user",
content=f"{original_content}\n\nPlease think through this step-by-step."
)
response = await self.model.generate(cot_messages)
parsed_response = self._parse_reasoning(response)
if not show_reasoning:
return {
"reasoning_steps": [],
"final_answer": parsed_response["final_answer"]
}
return parsed_response
def _parse_reasoning(self, response: str) -> Dict[str, Any]:
lines = response.split('\n')
reasoning_steps = []
final_answer_lines = []
in_conclusion = False
for line in lines:
line = line.strip()
if not line:
continue
conclusion_markers = [
"therefore", "thus", "in conclusion", "finally",
"the answer is", "conclusion:"
]
if any(marker in line.lower() for marker in conclusion_markers):
in_conclusion = True
if line and (line[0].isdigit() or line.startswith('-') or line.startswith('*')):
if not in_conclusion:
reasoning_steps.append(line)
else:
final_answer_lines.append(line)
else:
if in_conclusion:
final_answer_lines.append(line)
else:
if reasoning_steps:
reasoning_steps[-1] += " " + line
else:
reasoning_steps.append(line)
if not final_answer_lines:
if reasoning_steps:
final_answer_lines = [reasoning_steps.pop()]
else:
final_answer_lines = [response]
return {
"reasoning_steps": reasoning_steps,
"final_answer": "\n".join(final_answer_lines)
}
class TreeOfThought:
"""Implements Tree-of-Thought reasoning with multiple exploration paths."""
def __init__(self, model: LanguageModel, num_branches: int = 3):
self.model = model
self.num_branches = num_branches
self.tot_system_prompt = """You are a creative problem solver who explores multiple approaches.
When given a problem, consider different angles and strategies for solving it.
Be thorough and analytical in your exploration of each approach."""
async def reason(self, messages: List[Message], selection_strategy: str = "best") -> Dict[str, Any]:
branches = await self._generate_branches(messages)
evaluated_branches = await self._evaluate_branches(branches, messages)
if selection_strategy == "best":
final_answer = self._select_best_branch(evaluated_branches)
elif selection_strategy == "consensus":
final_answer = await self._build_consensus(evaluated_branches, messages)
else:
raise ValueError(f"Unknown selection strategy: {selection_strategy}")
return {
"branches": evaluated_branches,
"final_answer": final_answer,
"strategy": selection_strategy
}
async def _generate_branches(self, messages: List[Message]) -> List[str]:
branch_prompts = self._create_branch_prompts(messages)
tasks = [self.model.generate(prompt_messages) for prompt_messages in branch_prompts]
branches = await asyncio.gather(*tasks)
return branches
def _create_branch_prompts(self, messages: List[Message]) -> List[List[Message]]:
branch_prompts = []
if messages[-1].role != "user":
raise ValueError("Last message must be from user")
user_question = messages[-1].content
approaches = [
"Approach this problem analytically, breaking it into logical components.",
"Think about this creatively, considering unconventional solutions.",
"Consider this from first principles, building up from fundamental concepts.",
"Examine this problem from multiple perspectives before concluding.",
"Use analogies and related examples to reason about this problem."
]
for i in range(min(self.num_branches, len(approaches))):
branch_messages = [Message(role="system", content=self.tot_system_prompt)]
for msg in messages[:-1]:
if msg.role != "system":
branch_messages.append(msg)
modified_question = f"{approaches[i]}\n\n{user_question}"
branch_messages.append(Message(role="user", content=modified_question))
branch_prompts.append(branch_messages)
return branch_prompts
async def _evaluate_branches(self, branches: List[str], original_messages: List[Message]) -> List[Dict[str, Any]]:
evaluated = []
for i, branch in enumerate(branches):
eval_messages = [
Message(role="system", content="You are an expert evaluator of reasoning quality."),
Message(
role="user",
content=f"""Evaluate the following reasoning on a scale of 1-10 based on:
- Logical coherence
- Completeness of analysis
- Correctness of conclusions
- Clarity of explanation
Original question: {original_messages[-1].content}
Reasoning to evaluate:
{branch}
Provide only a numeric score from 1-10."""
)
]
score_text = await self.model.generate(eval_messages, temperature=0.3)
try:
numbers = re.findall(r'\d+', score_text)
score = int(numbers[0]) if numbers else 5
score = max(1, min(10, score))
except (ValueError, IndexError):
score = 5
evaluated.append({
"branch_id": i,
"content": branch,
"score": score
})
return evaluated
def _select_best_branch(self, evaluated_branches: List[Dict[str, Any]]) -> str:
best_branch = max(evaluated_branches, key=lambda b: b["score"])
return best_branch["content"]
async def _build_consensus(self, evaluated_branches: List[Dict[str, Any]], original_messages: List[Message]) -> str:
branches_text = "\n\n---\n\n".join([
f"Approach {b['branch_id'] + 1} (Score: {b['score']}/10):\n{b['content']}"
for b in evaluated_branches
])
synthesis_messages = [
Message(role="system", content="You synthesize insights from multiple reasoning paths."),
Message(
role="user",
content=f"""Given these different reasoning approaches to the question:
"{original_messages[-1].content}"
{branches_text}
Synthesize the best insights from all approaches into a single, coherent answer.
Focus on points where multiple approaches agree, and note any important unique insights."""
)
]
consensus = await self.model.generate(synthesis_messages)
return consensus
class SelfReflection:
"""Implements self-reflection and iterative refinement."""
def __init__(self, model: LanguageModel, max_iterations: int = 2):
self.model = model
self.max_iterations = max_iterations
self.reflection_system_prompt = """You are a critical evaluator who identifies weaknesses in responses.
When reviewing an answer, look for:
- Factual errors or inconsistencies
- Unclear explanations
- Missing important information
- Logical flaws
- Ways to improve clarity and completeness"""
self.refinement_system_prompt = """You are an expert at improving responses based on feedback.
Take the original response and the critique, then create an improved version that addresses all issues raised."""
async def reflect_and_refine(self, messages: List[Message], initial_response: Optional[str] = None) -> Dict[str, Any]:
if initial_response is None:
current_response = await self.model.generate(messages)
else:
current_response = initial_response
reflection_history = []
for iteration in range(self.max_iterations):
critique = await self._generate_critique(messages, current_response)
reflection_history.append({
"iteration": iteration,
"response": current_response,
"critique": critique
})
if self._is_satisfactory(critique):
break
refined_response = await self._refine_response(messages, current_response, critique)
current_response = refined_response
reflection_history.append({
"iteration": len(reflection_history),
"response": current_response,
"critique": "Final response accepted"
})
return {
"reflection_history": reflection_history,
"final_response": current_response,
"iterations_used": len(reflection_history) - 1
}
async def _generate_critique(self, original_messages: List[Message], response: str) -> str:
critique_messages = [
Message(role="system", content=self.reflection_system_prompt),
Message(
role="user",
content=f"""Original question: {original_messages[-1].content}
Response to evaluate:
{response}
Please provide a critical evaluation. If the response is excellent, say "SATISFACTORY".
Otherwise, identify specific issues and suggest improvements."""
)
]
critique = await self.model.generate(critique_messages, temperature=0.3)
return critique
def _is_satisfactory(self, critique: str) -> bool:
satisfaction_markers = [
"satisfactory", "excellent", "no issues", "well done", "no improvements needed"
]
critique_lower = critique.lower()
return any(marker in critique_lower for marker in satisfaction_markers)
async def _refine_response(self, original_messages: List[Message], current_response: str, critique: str) -> str:
refinement_messages = [
Message(role="system", content=self.refinement_system_prompt),
Message(
role="user",
content=f"""Original question: {original_messages[-1].content}
Current response:
{current_response}
Critique and suggestions for improvement:
{critique}
Please provide an improved response that addresses all the issues raised in the critique."""
)
]
refined = await self.model.generate(refinement_messages)
return refined
class PromptEnhancer:
"""Enhances user prompts for better language model responses."""
def __init__(self, model: LanguageModel):
self.model = model
self.enhancement_system_prompt = """You are an expert at crafting effective prompts for language models.
When given a user's question or request, analyze it and create an enhanced version that:
1. Clarifies any ambiguity in the original question
2. Adds relevant context that helps answer the question
3. Structures the request for optimal model understanding
4. Specifies the desired format or detail level if unclear
5. Breaks complex requests into clear sub-questions if needed
Provide the enhanced prompt(s) in a clear, structured format."""
async def enhance_prompt(self, user_message: str, conversation_context: List[Message] = None) -> Dict[str, Any]:
analysis = await self._analyze_prompt(user_message, conversation_context)
enhanced = await self._generate_enhanced_prompt(user_message, analysis, conversation_context)
return {
"original_prompt": user_message,
"analysis": analysis,
"enhanced_prompt": enhanced["primary"],
"alternative_prompts": enhanced.get("alternatives", []),
"enhancement_strategy": enhanced["strategy"]
}
async def _analyze_prompt(self, user_message: str, context: List[Message] = None) -> Dict[str, Any]:
context_summary = ""
if context:
context_messages = [msg.content for msg in context[-3:]]
context_summary = f"\n\nRecent conversation context:\n" + "\n".join(context_messages)
analysis_messages = [
Message(role="system", content="You analyze prompts to identify intent, ambiguities, and requirements."),
Message(
role="user",
content=f"""Analyze this user message:{context_summary}
User message: {user_message}
Identify:
1. Primary intent (what the user wants to achieve)
2. Any ambiguities or unclear aspects
3. Missing context that would help answer better
4. Complexity level (simple, moderate, complex)
5. Optimal response format (explanation, step-by-step, list, etc.)
Provide your analysis in a structured format."""
)
]
analysis_text = await self.model.generate(analysis_messages, temperature=0.3)
return self._parse_analysis(analysis_text)
def _parse_analysis(self, analysis_text: str) -> Dict[str, Any]:
analysis = {
"intent": "",
"ambiguities": [],
"missing_context": [],
"complexity": "moderate",
"response_format": "explanation"
}
lines = analysis_text.split('\n')
current_section = None
for line in lines:
line = line.strip()
if not line:
continue
lower_line = line.lower()
if "intent" in lower_line or "purpose" in lower_line:
current_section = "intent"
elif "ambiguit" in lower_line or "unclear" in lower_line:
current_section = "ambiguities"
elif "missing" in lower_line or "context" in lower_line:
current_section = "missing_context"
elif "complexity" in lower_line:
current_section = "complexity"
elif "format" in lower_line:
current_section = "response_format"
elif current_section:
if current_section == "intent":
analysis["intent"] += " " + line
elif current_section in ["ambiguities", "missing_context"]:
if line.startswith('-') or line.startswith('*') or line[0].isdigit():
analysis[current_section].append(line.lstrip('-*0123456789. '))
elif current_section == "complexity":
if "simple" in lower_line:
analysis["complexity"] = "simple"
elif "complex" in lower_line:
analysis["complexity"] = "complex"
elif current_section == "response_format":
analysis["response_format"] = line.lstrip('-*0123456789. ')
return analysis
async def _generate_enhanced_prompt(self, original_prompt: str, analysis: Dict[str, Any], context: List[Message] = None) -> Dict[str, Any]:
enhancement_messages = [
Message(role="system", content=self.enhancement_system_prompt),
Message(
role="user",
content=f"""Original prompt: {original_prompt}
Analysis:
- Intent: {analysis['intent']}
- Ambiguities: {', '.join(analysis['ambiguities']) if analysis['ambiguities'] else 'None'}
- Missing context: {', '.join(analysis['missing_context']) if analysis['missing_context'] else 'None'}
- Complexity: {analysis['complexity']}
- Desired format: {analysis['response_format']}
Create an enhanced version of this prompt that addresses the ambiguities, adds necessary context,
and structures the request optimally. If the request is complex, consider breaking it into sub-questions."""
)
]
enhanced_text = await self.model.generate(enhancement_messages)
if analysis['complexity'] == 'complex':
strategy = "decomposition"
alternatives = self._extract_sub_prompts(enhanced_text)
else:
strategy = "clarification"
alternatives = []
return {
"primary": enhanced_text,
"alternatives": alternatives,
"strategy": strategy
}
def _extract_sub_prompts(self, enhanced_text: str) -> List[str]:
sub_prompts = []
lines = enhanced_text.split('\n')
for line in lines:
line = line.strip()
if line and (line[0].isdigit() or line.startswith('-') or line.startswith('*')) and '?' in line:
cleaned = line.lstrip('-*0123456789. ')
if cleaned:
sub_prompts.append(cleaned)
return sub_prompts
class ReasoningOrchestrator:
"""Orchestrates different reasoning strategies."""
def __init__(self, model: LanguageModel):
self.model = model
self.cot = ChainOfThought(model)
self.tot = TreeOfThought(model)
self.reflection = SelfReflection(model)
self.enhancer = PromptEnhancer(model)
async def process_query(self, messages: List[Message], strategy: str = "auto", enhance_prompt: bool = True, show_reasoning: bool = True) -> Dict[str, Any]:
result = {
"strategy_used": strategy,
"prompt_enhanced": enhance_prompt,
"reasoning_visible": show_reasoning
}
if enhance_prompt and messages:
enhancement = await self.enhancer.enhance_prompt(
messages[-1].content,
messages[:-1] if len(messages) > 1 else None
)
enhanced_messages = messages[:-1] + [Message(role="user", content=enhancement["enhanced_prompt"])]
result["prompt_enhancement"] = enhancement
else:
enhanced_messages = messages
result["prompt_enhancement"] = None
if strategy == "auto":
strategy = self._select_strategy(enhanced_messages)
result["strategy_used"] = strategy
if strategy == "cot":
reasoning_result = await self.cot.reason(enhanced_messages, show_reasoning=show_reasoning)
result["response"] = reasoning_result["final_answer"]
result["reasoning_steps"] = reasoning_result.get("reasoning_steps", [])
elif strategy == "tot":
reasoning_result = await self.tot.reason(enhanced_messages, selection_strategy="best")
result["response"] = reasoning_result["final_answer"]
result["branches"] = reasoning_result.get("branches", []) if show_reasoning else []
elif strategy == "reflection":
reasoning_result = await self.reflection.reflect_and_refine(enhanced_messages)
result["response"] = reasoning_result["final_response"]
result["reflection_history"] = reasoning_result.get("reflection_history", []) if show_reasoning else []
elif strategy == "simple":
response = await self.model.generate(enhanced_messages)
result["response"] = response
else:
raise ValueError(f"Unknown strategy: {strategy}")
return result
def _select_strategy(self, messages: List[Message]) -> str:
if not messages:
return "simple"
last_message = messages[-1].content.lower()
tot_indicators = ["compare", "evaluate", "which is better", "pros and cons", "different approaches", "alternatives", "options"]
if any(indicator in last_message for indicator in tot_indicators):
return "tot"
cot_indicators = ["how", "why", "explain", "step by step", "calculate", "solve", "prove", "derive"]
if any(indicator in last_message for indicator in cot_indicators):
return "cot"
reflection_indicators = ["accurate", "precise", "correct", "verify", "check", "important", "critical"]
if any(indicator in last_message for indicator in reflection_indicators):
return "reflection"
return "cot"
async def stream_response(self, messages: List[Message], strategy: str = "simple", enhance_prompt: bool = False):
if strategy != "simple":
raise ValueError("Only 'simple' strategy supports streaming")
if enhance_prompt and messages:
enhancement = await self.enhancer.enhance_prompt(
messages[-1].content,
messages[:-1] if len(messages) > 1 else None
)
enhanced_messages = messages[:-1] + [Message(role="user", content=enhancement["enhanced_prompt"])]
else:
enhanced_messages = messages
async for chunk in self.model.stream_generate(enhanced_messages):
yield chunk
# =============================================================================
# CONVERSATION MODULE - Conversation management with summarization
# =============================================================================
class ConversationManager:
"""Manages conversation history with intelligent summarization."""
def __init__(self, model: LanguageModel, max_context_tokens: Optional[int] = None, summary_trigger_ratio: float = 0.75):
self.model = model
self.max_context_tokens = max_context_tokens or model.get_context_window()
self.summary_trigger_ratio = summary_trigger_ratio
self.reserved_tokens = 1024
self.effective_max_tokens = self.max_context_tokens - self.reserved_tokens
self.summarization_prompt = """Summarize the following conversation, preserving:
1. Key facts and information discussed
2. Important decisions or conclusions reached
3. The overall context and topic
4. Any unresolved questions or ongoing discussions
Be concise but comprehensive. The summary will be used to maintain context in an ongoing conversation."""
async def get_messages_for_context(self, messages: List[Message], system_message: Optional[Message] = None) -> List[Message]:
total_tokens = 0
if system_message:
total_tokens += self.model.count_tokens(system_message.content)
for msg in messages:
total_tokens += self.model.count_tokens(msg.content)
if total_tokens <= self.effective_max_tokens * self.summary_trigger_ratio:
result = []
if system_message:
result.append(system_message)
result.extend(messages)
return result
return await self._summarize_and_compress(messages, system_message)
async def _summarize_and_compress(self, messages: List[Message], system_message: Optional[Message]) -> List[Message]:
recent_message_count = 10
if len(messages) <= recent_message_count:
result = []
if system_message:
result.append(system_message)
result.extend(messages)
return result
old_messages = messages[:-recent_message_count]
recent_messages = messages[-recent_message_count:]
summary = await self._generate_summary(old_messages)
result = []
if system_message:
result.append(system_message)
result.append(Message(role="system", content=f"Previous conversation summary:\n{summary}"))
result.extend(recent_messages)
return result
async def _generate_summary(self, messages: List[Message]) -> str:
conversation_text = self._format_messages_for_summary(messages)
summary_messages = [
Message(role="system", content=self.summarization_prompt),
Message(role="user", content=conversation_text)
]
summary = await self.model.generate(summary_messages, temperature=0.3, max_tokens=500)
return summary
def _format_messages_for_summary(self, messages: List[Message]) -> str:
formatted_parts = []
for msg in messages:
role_label = msg.role.capitalize()
formatted_parts.append(f"{role_label}: {msg.content}")
return "\n\n".join(formatted_parts)
# =============================================================================
# STORAGE MODULE - Session persistence
# =============================================================================
Base = declarative_base()
class SessionModel(Base):
"""Database model for conversation sessions."""
__tablename__ = 'sessions'
id = Column(String, primary_key=True)
user_id = Column(String, nullable=False, index=True)
title = Column(String, nullable=False)
created_at = Column(DateTime, nullable=False)
updated_at = Column(DateTime, nullable=False)
messages = Column(Text, nullable=False)
metadata = Column(Text)
message_count = Column(Integer, default=0)
class SessionStore:
"""Manages persistent storage of conversation sessions."""
def __init__(self, database_url: str = "sqlite:///sessions.db"):
self.engine = create_engine(database_url)
Base.metadata.create_all(self.engine)
self.SessionLocal = sessionmaker(bind=self.engine)
def create_session(self, user_id: str, title: Optional[str] = None, initial_messages: Optional[List[Message]] = None, metadata: Optional[Dict[str, Any]] = None) -> str:
session_id = str(uuid.uuid4())
now = datetime.utcnow()
messages = initial_messages or []
messages_json = json.dumps([self._message_to_dict(m) for m in messages])
metadata_json = json.dumps(metadata or {})
session = SessionModel(
id=session_id,
user_id=user_id,
title=title or f"Conversation {now.strftime('%Y-%m-%d %H:%M')}",
created_at=now,
updated_at=now,
messages=messages_json,
metadata=metadata_json,
message_count=len(messages)
)
db = self.SessionLocal()
try:
db.add(session)
db.commit()
finally:
db.close()
return session_id
def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
db = self.SessionLocal()
try:
session = db.query(SessionModel).filter(SessionModel.id == session_id).first()
if not session:
return None
return self._session_to_dict(session)
finally:
db.close()
def update_session(self, session_id: str, messages: Optional[List[Message]] = None, title: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> bool:
db = self.SessionLocal()
try:
session = db.query(SessionModel).filter(SessionModel.id == session_id).first()
if not session:
return False
if messages is not None:
session.messages = json.dumps([self._message_to_dict(m) for m in messages])
session.message_count = len(messages)
if title is not None:
session.title = title
if metadata is not None:
session.metadata = json.dumps(metadata)
session.updated_at = datetime.utcnow()
db.commit()
return True
finally:
db.close()
def list_sessions(self, user_id: str, limit: int = 50, offset: int = 0) -> List[Dict[str, Any]]:
db = self.SessionLocal()
try:
sessions = db.query(SessionModel).filter(
SessionModel.user_id == user_id
).order_by(SessionModel.updated_at.desc()).limit(limit).offset(offset).all()
return [self._session_to_summary(s) for s in sessions]
finally:
db.close()
def delete_session(self, session_id: str) -> bool:
db = self.SessionLocal()
try:
session = db.query(SessionModel).filter(SessionModel.id == session_id).first()
if not session:
return False
db.delete(session)
db.commit()
return True
finally:
db.close()
def _message_to_dict(self, message: Message) -> Dict[str, str]:
return {"role": message.role, "content": message.content}
def _dict_to_message(self, data: Dict[str, str]) -> Message:
return Message(role=data["role"], content=data["content"])
def _session_to_dict(self, session: SessionModel) -> Dict[str, Any]:
messages_data = json.loads(session.messages)
messages = [self._dict_to_message(m) for m in messages_data]
metadata = json.loads(session.metadata) if session.metadata else {}
return {
"id": session.id,
"user_id": session.user_id,
"title": session.title,
"created_at": session.created_at.isoformat(),
"updated_at": session.updated_at.isoformat(),
"messages": messages,
"metadata": metadata,
"message_count": session.message_count
}
def _session_to_summary(self, session: SessionModel) -> Dict[str, Any]:
return {
"id": session.id,
"title": session.title,
"created_at": session.created_at.isoformat(),
"updated_at": session.updated_at.isoformat(),
"message_count": session.message_count
}
# =============================================================================
# WEB SERVER - FastAPI application
# =============================================================================
class ChatRequest(BaseModel):
message: str
session_id: Optional[str] = None
strategy: str = "auto"
enhance_prompt: bool = True
show_reasoning: bool = True
class ChatResponse(BaseModel):
response: str
session_id: str
strategy_used: str
reasoning_steps: Optional[List[str]] = None
branches: Optional[List[Dict[str, Any]]] = None
reflection_history: Optional[List[Dict[str, Any]]] = None
class SessionCreate(BaseModel):
title: Optional[str] = None
class SessionResponse(BaseModel):
session_id: str
title: str
created_at: str
updated_at: str
message_count: int
app = FastAPI(
title="Advanced LLM Chatbot",
description="Chatbot with Chain-of-Thought, Tree-of-Thought, and Self-Reflection",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
model = None
conversation_manager = None
reasoning_orchestrator = None
session_store = None
def initialize_system(config: Dict[str, Any]):
global model, conversation_manager, reasoning_orchestrator, session_store
model = ModelFactory.create_from_settings(config["model"])
conversation_manager = ConversationManager(model)
reasoning_orchestrator = ReasoningOrchestrator(model)
session_store = SessionStore(config.get("database_url", "sqlite:///sessions.db"))
@app.on_event("startup")
async def startup_event():
config = {
"model": {
"type": os.getenv("MODEL_TYPE", "remote"),
"model_name": os.getenv("MODEL_NAME", "gpt-3.5-turbo"),
"api_key": os.getenv("OPENAI_API_KEY"),
"model_path": os.getenv("MODEL_PATH"),
"temperature": float(os.getenv("TEMPERATURE", "0.7")),
"max_tokens": int(os.getenv("MAX_TOKENS", "2048")),
"gpu_layers": int(os.getenv("GPU_LAYERS", "-1")) if os.getenv("GPU_LAYERS") else None
},
"database_url": os.getenv("DATABASE_URL", "sqlite:///sessions.db")
}
initialize_system(config)
@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
if request.session_id:
session_data = session_store.get_session(request.session_id)
if not session_data:
raise HTTPException(status_code=404, detail="Session not found")
messages = session_data["messages"]
session_id = request.session_id
else:
session_id = session_store.create_session(
user_id="default_user",
title=f"Chat {datetime.now().strftime('%Y-%m-%d %H:%M')}"
)
messages = []
messages.append(Message(role="user", content=request.message))
context_messages = await conversation_manager.get_messages_for_context(messages)
result = await reasoning_orchestrator.process_query(
context_messages,
strategy=request.strategy,
enhance_prompt=request.enhance_prompt,
show_reasoning=request.show_reasoning
)
messages.append(Message(role="assistant", content=result["response"]))
session_store.update_session(session_id, messages=messages)
return ChatResponse(
response=result["response"],
session_id=session_id,
strategy_used=result["strategy_used"],
reasoning_steps=result.get("reasoning_steps"),
branches=result.get("branches"),
reflection_history=result.get("reflection_history")
)
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
message_text = data.get("message")
session_id = data.get("session_id")
if session_id:
session_data = session_store.get_session(session_id)
if session_data:
messages = session_data["messages"]
else:
session_id = session_store.create_session(user_id="default_user")
messages = []
else:
session_id = session_store.create_session(user_id="default_user")
messages = []
messages.append(Message(role="user", content=message_text))
context_messages = await conversation_manager.get_messages_for_context(messages)
await websocket.send_json({"type": "session_id", "session_id": session_id})
response_text = ""
async for chunk in reasoning_orchestrator.stream_response(context_messages, strategy="simple"):
response_text += chunk
await websocket.send_json({"type": "chunk", "content": chunk})
await websocket.send_json({"type": "complete"})
messages.append(Message(role="assistant", content=response_text))
session_store.update_session(session_id, messages=messages)
except WebSocketDisconnect:
pass
except Exception as e:
print(f"WebSocket error: {e}")
try:
await websocket.close()
except:
pass
@app.post("/api/sessions", response_model=SessionResponse)
async def create_session(request: SessionCreate):
session_id = session_store.create_session(user_id="default_user", title=request.title)
session_data = session_store.get_session(session_id)
return SessionResponse(
session_id=session_data["id"],
title=session_data["title"],
created_at=session_data["created_at"],
updated_at=session_data["updated_at"],
message_count=session_data["message_count"]
)
@app.get("/api/sessions", response_model=List[SessionResponse])
async def list_sessions(limit: int = 50, offset: int = 0):
sessions = session_store.list_sessions(user_id="default_user", limit=limit, offset=offset)
return [
SessionResponse(
session_id=s["id"],
title=s["title"],
created_at=s["created_at"],
updated_at=s["updated_at"],
message_count=s["message_count"]
)
for s in sessions
]
@app.get("/api/sessions/{session_id}")
async def get_session(session_id: str):
session_data = session_store.get_session(session_id)
if not session_data:
raise HTTPException(status_code=404, detail="Session not found")
messages_dict = [{"role": msg.role, "content": msg.content} for msg in session_data["messages"]]
return {
"session_id": session_data["id"],
"title": session_data["title"],
"created_at": session_data["created_at"],
"updated_at": session_data["updated_at"],
"messages": messages_dict,
"message_count": session_data["message_count"]
}
@app.delete("/api/sessions/{session_id}")
async def delete_session(session_id: str):
success = session_store.delete_session(session_id)
if not success:
raise HTTPException(status_code=404, detail="Session not found")
return {"message": "Session deleted successfully"}
@app.get("/")
async def root():
return JSONResponse(content={
"message": "Advanced LLM Chatbot API",
"version": "1.0.0",
"endpoints": {
"chat": "/api/chat",
"websocket": "/ws/chat",
"sessions": "/api/sessions"
}
})
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
This complete implementation provides a fully functional chatbot system with all the features described throughout the tutorial. It can be deployed immediately by installing the required dependencies and setting appropriate environment variables for configuration.
To use the system, install the dependencies with pip install fastapi uvicorn sqlalchemy aiosqlite pydantic python-dotenv. For remote models, also install openai and tiktoken. For local models, install llama-cpp-python with appropriate GPU support flags.
Set environment variables for configuration. For remote models, set OPENAI_API_KEY to your API key and MODEL_NAME to the desired model. For local models, set MODEL_TYPE to local, MODEL_PATH to the path of your GGUF model file, and optionally GPU_LAYERS to control GPU offloading.
Run the server with python complete_chatbot.py. The web interface will be available at http://localhost:8000. The system provides full Chain-of-Thought, Tree-of-Thought, and Self-Reflection capabilities, with automatic prompt enhancement, conversation summarization, and session persistence.
The implementation follows clean architecture principles with clear separation of concerns. The models module handles language model abstraction. The reasoning module implements advanced reasoning techniques. The conversation module manages chat history. The storage module handles persistence. The web server ties everything together with a REST and WebSocket API.
The system supports both local and remote language models with a unified interface. Local models can leverage GPU acceleration on NVIDIA CUDA, AMD ROCm, Intel, and Apple Metal platforms through llama.cpp. Remote models access cloud APIs for powerful models without local computational requirements.
The web interface provides an intuitive user experience with session management, configurable reasoning strategies, and transparent display of reasoning processes. Users can see how the system arrives at its conclusions, building trust and understanding of the AI's decision-making.
This tutorial has covered the complete architecture and implementation of an advanced LLM chatbot system. The combination of Chain-of-Thought, Tree-of-Thought, and Self-Reflection reasoning techniques produces higher quality responses than standard chatbots. The prompt enhancement system optimizes user inputs for better model understanding. The conversation management ensures context is maintained efficiently. The session storage provides persistence across interactions. The web interface makes all these capabilities accessible through an elegant, user-friendly design.
