Note: The code used in the article was generated by Claude Sonnet step by step.
INTRODUCTION TO THE VISION
The landscape of Large Language Model applications has exploded in complexity. Developers building chatbots, agent systems, or agentic AI workflows face a daunting array of technical challenges. They must handle GPU detection across NVIDIA CUDA, AMD ROCm, Intel oneAPI, and Apple Metal. They need to manage both local and remote model execution, implement retrieval-augmented generation with vector databases, orchestrate multi-agent communication through protocols like MCP, handle asynchronous and synchronous execution patterns, and build user interfaces for both console and web environments. Each of these concerns involves substantial boilerplate code that gets rewritten for every project.
This tutorial presents the design and implementation of a Domain-Specific Language that eliminates this repetitive work. The DSL provides an intuitive, declarative syntax that both beginners and experts can use to specify their LLM applications at a high level. The system then generates production-ready Python code that handles all the low-level details. Users can extend the generated code with custom Python components, giving them the flexibility to implement specialized logic while benefiting from automated infrastructure.
Our DSL will be called LLMDSL, and it will transform simple declarations into complete, working applications. The generated Python code will be clean, well-structured, and follow industry best practices. The system will automatically detect available hardware, configure appropriate backends, set up communication channels between agents, implement RAG pipelines, and generate user interfaces.
UNDERSTANDING THE PROBLEM DOMAIN
Before we can design an effective DSL, we must deeply understand the problem space. LLM applications share common patterns despite their surface-level diversity. A chatbot maintains conversation history and generates responses based on user input. A single agent performs tasks autonomously using tools and reasoning. Multi-agent systems coordinate multiple specialized agents to solve complex problems. Agentic AI systems exhibit goal-directed behavior and adapt to changing circumstances.
All these applications require similar infrastructure. They need to load and manage language models, whether those models run locally or are accessed through remote APIs. They must handle conversation state and context management. They require error handling and retry logic for robust operation. They often need access to external knowledge through retrieval mechanisms. They must coordinate asynchronous operations and manage concurrent execution. They need user interfaces for interaction.
The key insight is that developers should focus on what their application does, not how it does it. The DSL should let them declare the agents, their capabilities, their knowledge sources, and their interaction patterns. The code generator then handles the implementation details.
ARCHITECTURAL OVERVIEW
The LLMDSL system consists of several major components working together. The parser reads DSL source files and builds an abstract syntax tree representing the application structure. The semantic analyzer validates the AST and resolves references between components. The code generator traverses the validated AST and emits Python code.
The runtime library provides the infrastructure that generated code depends on, including GPU detection, model management, agent orchestration, and UI frameworks.
The architecture follows clean separation of concerns. The parser knows only about syntax. The semantic analyzer knows about the meaning and relationships of declarations. The code generator knows about Python code emission. The runtime library knows about LLM operations and system resources. This separation makes the system maintainable and extensible.
Let us examine each component in detail, starting with the DSL syntax itself.
DESIGNING THE DSL SYNTAX
The syntax must be intuitive and expressive while remaining unambiguous and parseable. We will use a declarative style where developers describe what they want rather than how to achieve it. The syntax draws inspiration from configuration languages like YAML and HCL but adds programming constructs where needed.
A simple chatbot declaration might look like this:
chatbot SimpleAssistant {
model: "llama-3-8b"
system_prompt: "You are a helpful assistant."
temperature: 0.7
max_tokens: 2048
}
This declaration specifies a chatbot named SimpleAssistant that uses the Llama 3 8B model with a specific system prompt and generation parameters. The code generator will create a complete Python application that loads the model, manages conversation state, and provides a user interface.
For more complex scenarios, we need to specify agents with tools and capabilities:
agent ResearchAgent {
model: "gpt-4"
system_prompt: "You are a research assistant."
tools: [
web_search,
document_reader,
calculator
]
rag: {
vector_store: "chroma"
embedding_model: "sentence-transformers/all-MiniLM-L6-v2"
documents: "./knowledge_base"
}
}
This agent declaration includes tool definitions and a RAG configuration. The generated code will set up the vector store, embed documents, and integrate retrieval into the agent's reasoning process.
Multi-agent systems require coordination specifications:
multi_agent_system CustomerSupport {
agents: [
agent Classifier {
model: "llama-3-8b"
role: "Classify customer inquiries"
},
agent TechnicalSupport {
model: "gpt-4"
role: "Handle technical questions"
},
agent BillingSupport {
model: "gpt-3.5-turbo"
role: "Handle billing questions"
}
]
orchestration: {
entry_point: Classifier
routing: {
Classifier -> TechnicalSupport: when category == "technical"
Classifier -> BillingSupport: when category == "billing"
}
}
communication: "mcp"
}
This declares a multi-agent system where a classifier agent routes inquiries to specialized support agents. The orchestration section specifies the workflow, and the communication protocol is MCP.
The DSL also needs to specify execution settings:
settings {
gpu: "auto" // auto-detect and use best available
execution_mode: "async"
max_concurrent: 5
retry_policy: {
max_retries: 3
backoff: "exponential"
}
ui: "web"
port: 8080
}
These settings control how the application executes. The GPU setting determines hardware acceleration. The execution mode controls whether operations run asynchronously. The UI setting determines whether to generate a console or web interface.
IMPLEMENTING THE PARSER
The parser transforms DSL source text into an abstract syntax tree. We will implement a recursive descent parser that handles the DSL grammar. The parser needs to recognize keywords, identifiers, literals, and structural elements like blocks and declarations.
The parser implementation begins with a lexer that tokenizes the input:
class Token:
def __init__(self, token_type, value, line, column):
self.type = token_type
self.value = value
self.line = line
self.column = column
class Lexer:
def __init__(self, source):
self.source = source
self.position = 0
self.line = 1
self.column = 1
self.tokens = []
def tokenize(self):
while self.position < len(self.source):
self.skip_whitespace()
if self.position >= len(self.source):
break
if self.current_char() == '#':
self.skip_comment()
continue
if self.current_char().isalpha() or self.current_char() == '_':
self.read_identifier()
elif self.current_char().isdigit():
self.read_number()
elif self.current_char() == '"':
self.read_string()
elif self.current_char() in '{}[]():,':
self.read_punctuation()
else:
raise SyntaxError(f"Unexpected character '{self.current_char()}' at line {self.line}, column {self.column}")
self.tokens.append(Token('EOF', None, self.line, self.column))
return self.tokens
The lexer breaks the source into tokens representing keywords, identifiers, literals, and punctuation. It tracks line and column numbers for error reporting.
The parser builds the AST from tokens:
class ASTNode:
pass
class ChatbotDeclaration(ASTNode):
def __init__(self, name, properties):
self.name = name
self.properties = properties
class AgentDeclaration(ASTNode):
def __init__(self, name, properties):
self.name = name
self.properties = properties
class MultiAgentSystem(ASTNode):
def __init__(self, name, agents, orchestration, communication):
self.name = name
self.agents = agents
self.orchestration = orchestration
self.communication = communication
class Parser:
def __init__(self, tokens):
self.tokens = tokens
self.position = 0
def parse(self):
declarations = []
while not self.is_at_end():
declarations.append(self.parse_declaration())
return declarations
def parse_declaration(self):
token = self.current_token()
if token.value == 'chatbot':
return self.parse_chatbot()
elif token.value == 'agent':
return self.parse_agent()
elif token.value == 'multi_agent_system':
return self.parse_multi_agent_system()
elif token.value == 'settings':
return self.parse_settings()
else:
raise SyntaxError(f"Unexpected declaration type '{token.value}' at line {token.line}")
The parser creates specific AST node types for each declaration kind. Each node captures the semantic information needed for code generation.
SEMANTIC ANALYSIS AND VALIDATION
After parsing, we must validate the AST and resolve references. The semantic analyzer checks that referenced models exist, that agent names are unique, that routing rules reference valid agents, and that all required properties are specified.
class SemanticAnalyzer:
def __init__(self, ast):
self.ast = ast
self.symbol_table = {}
self.errors = []
def analyze(self):
# First pass: collect all declarations
for declaration in self.ast:
self.register_declaration(declaration)
# Second pass: validate references and constraints
for declaration in self.ast:
self.validate_declaration(declaration)
if self.errors:
raise SemanticError('\n'.join(self.errors))
return True
def register_declaration(self, declaration):
if isinstance(declaration, (ChatbotDeclaration, AgentDeclaration)):
if declaration.name in self.symbol_table:
self.errors.append(f"Duplicate declaration: {declaration.name}")
else:
self.symbol_table[declaration.name] = declaration
def validate_declaration(self, declaration):
if isinstance(declaration, ChatbotDeclaration):
self.validate_chatbot(declaration)
elif isinstance(declaration, AgentDeclaration):
self.validate_agent(declaration)
elif isinstance(declaration, MultiAgentSystem):
self.validate_multi_agent_system(declaration)
def validate_chatbot(self, chatbot):
required_properties = ['model', 'system_prompt']
for prop in required_properties:
if prop not in chatbot.properties:
self.errors.append(f"Chatbot {chatbot.name} missing required property: {prop}")
def validate_multi_agent_system(self, system):
# Validate that all referenced agents exist
for agent in system.agents:
if isinstance(agent, str) and agent not in self.symbol_table:
self.errors.append(f"Unknown agent reference: {agent}")
# Validate routing rules
if system.orchestration and 'routing' in system.orchestration:
for route in system.orchestration['routing']:
source, target = route['source'], route['target']
if source not in [a.name for a in system.agents]:
self.errors.append(f"Routing rule references unknown source agent: {source}")
if target not in [a.name for a in system.agents]:
self.errors.append(f"Routing rule references unknown target agent: {target}")
The semantic analyzer performs multiple passes over the AST. The first pass builds a symbol table of all declarations. The second pass validates that references are valid and that all constraints are satisfied. This catches errors before code generation.
GPU DETECTION AND BACKEND SELECTION
One of the most valuable features of our DSL is automatic GPU detection and backend configuration. The runtime library includes a hardware detection module that probes the system and selects the optimal backend.
import subprocess
import platform
import os
class GPUDetector:
def __init__(self):
self.detected_gpus = []
self.recommended_backend = None
def detect(self):
"""Detect all available GPU backends and recommend the best one."""
system = platform.system()
# Check for NVIDIA CUDA
if self.check_nvidia_cuda():
self.detected_gpus.append({
'vendor': 'NVIDIA',
'backend': 'cuda',
'priority': 1
})
# Check for AMD ROCm
if self.check_amd_rocm():
self.detected_gpus.append({
'vendor': 'AMD',
'backend': 'rocm',
'priority': 2
})
# Check for Intel oneAPI
if self.check_intel_oneapi():
self.detected_gpus.append({
'vendor': 'Intel',
'backend': 'oneapi',
'priority': 3
})
# Check for Apple Metal
if system == 'Darwin' and self.check_apple_metal():
self.detected_gpus.append({
'vendor': 'Apple',
'backend': 'metal',
'priority': 1
})
# Sort by priority and select best
if self.detected_gpus:
self.detected_gpus.sort(key=lambda x: x['priority'])
self.recommended_backend = self.detected_gpus[0]['backend']
else:
self.recommended_backend = 'cpu'
return self.recommended_backend
def check_nvidia_cuda(self):
"""Check if NVIDIA CUDA is available."""
try:
result = subprocess.run(['nvidia-smi'],
capture_output=True,
text=True,
timeout=5)
if result.returncode == 0:
# Verify CUDA libraries are available
try:
import torch
return torch.cuda.is_available()
except ImportError:
return False
return False
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def check_amd_rocm(self):
"""Check if AMD ROCm is available."""
try:
result = subprocess.run(['rocm-smi'],
capture_output=True,
text=True,
timeout=5)
if result.returncode == 0:
# Check for ROCm environment
return os.path.exists('/opt/rocm') or 'ROCM_PATH' in os.environ
return False
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def check_intel_oneapi(self):
"""Check if Intel oneAPI is available."""
try:
# Check for Intel GPU
result = subprocess.run(['clinfo'],
capture_output=True,
text=True,
timeout=5)
if result.returncode == 0 and 'Intel' in result.stdout:
return 'ONEAPI_ROOT' in os.environ or os.path.exists('/opt/intel/oneapi')
return False
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def check_apple_metal(self):
"""Check if Apple Metal is available."""
try:
import platform
# Check for Apple Silicon
if platform.processor() == 'arm':
try:
import torch
return hasattr(torch.backends, 'mps') and torch.backends.mps.is_available()
except ImportError:
return False
return False
except Exception:
return False
def get_device_config(self):
"""Get configuration for the selected backend."""
if self.recommended_backend == 'cuda':
return {
'device': 'cuda',
'dtype': 'float16',
'device_map': 'auto'
}
elif self.recommended_backend == 'rocm':
return {
'device': 'cuda', # ROCm uses CUDA API
'dtype': 'float16',
'device_map': 'auto'
}
elif self.recommended_backend == 'metal':
return {
'device': 'mps',
'dtype': 'float16',
'device_map': None
}
elif self.recommended_backend == 'oneapi':
return {
'device': 'xpu',
'dtype': 'float16',
'device_map': 'auto'
}
else:
return {
'device': 'cpu',
'dtype': 'float32',
'device_map': None
}
The GPU detector probes for each backend by checking for vendor-specific tools and libraries. It assigns priorities to backends based on typical performance characteristics. NVIDIA CUDA and Apple Metal get highest priority on their respective platforms. The detector returns configuration parameters appropriate for the selected backend.
MODEL MANAGEMENT AND LOADING
The runtime library provides a unified interface for loading and managing models regardless of whether they run locally or remotely. The model manager handles downloading models, caching them, and initializing them with the correct backend configuration.
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
from typing import Optional, Dict, Any
class ModelManager:
def __init__(self, gpu_detector: GPUDetector):
self.gpu_detector = gpu_detector
self.loaded_models = {}
self.device_config = gpu_detector.get_device_config()
def load_local_model(self, model_name: str, **kwargs) -> Dict[str, Any]:
"""Load a local model using transformers."""
if model_name in self.loaded_models:
return self.loaded_models[model_name]
print(f"Loading model {model_name} on {self.device_config['device']}...")
# Merge device config with user kwargs
load_config = {**self.device_config, **kwargs}
try:
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=getattr(torch, load_config['dtype']),
device_map=load_config['device_map']
)
# Move to device if device_map not used
if load_config['device_map'] is None:
model = model.to(load_config['device'])
self.loaded_models[model_name] = {
'model': model,
'tokenizer': tokenizer,
'device': load_config['device']
}
print(f"Model {model_name} loaded successfully")
return self.loaded_models[model_name]
except Exception as e:
print(f"Error loading model {model_name}: {str(e)}")
raise
def create_remote_client(self, provider: str, model_name: str, api_key: Optional[str] = None):
"""Create a client for remote model access."""
if provider == 'openai':
from openai import OpenAI
client = OpenAI(api_key=api_key or os.environ.get('OPENAI_API_KEY'))
return RemoteModelClient(client, model_name, 'openai')
elif provider == 'anthropic':
from anthropic import Anthropic
client = Anthropic(api_key=api_key or os.environ.get('ANTHROPIC_API_KEY'))
return RemoteModelClient(client, model_name, 'anthropic')
else:
raise ValueError(f"Unsupported provider: {provider}")
def unload_model(self, model_name: str):
"""Unload a model to free memory."""
if model_name in self.loaded_models:
del self.loaded_models[model_name]
if torch.cuda.is_available():
torch.cuda.empty_cache()
class RemoteModelClient:
def __init__(self, client, model_name: str, provider: str):
self.client = client
self.model_name = model_name
self.provider = provider
def generate(self, messages: list, **kwargs) -> str:
"""Generate a response using the remote API."""
if self.provider == 'openai':
response = self.client.chat.completions.create(
model=self.model_name,
messages=messages,
**kwargs
)
return response.choices[0].message.content
elif self.provider == 'anthropic':
response = self.client.messages.create(
model=self.model_name,
messages=messages,
**kwargs
)
return response.content[0].text
else:
raise ValueError(f"Unsupported provider: {self.provider}")
The model manager abstracts the differences between local and remote models. For local models, it uses the transformers library and applies the appropriate device configuration. For remote models, it creates API clients for services like OpenAI or Anthropic. This unified interface allows the generated code to work with any model type.
IMPLEMENTING RETRIEVAL-AUGMENTED GENERATION
RAG enhances LLM responses by retrieving relevant information from a knowledge base. Our DSL supports RAG declarations, and the runtime library implements the RAG pipeline including document processing, embedding, vector storage, and retrieval.
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.document_loaders import DirectoryLoader, TextLoader
import os
from typing import List, Dict, Any
class RAGPipeline:
def __init__(self,
vector_store_type: str = 'chroma',
embedding_model: str = 'sentence-transformers/all-MiniLM-L6-v2',
chunk_size: int = 1000,
chunk_overlap: int = 200):
self.vector_store_type = vector_store_type
self.embedding_model_name = embedding_model
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.embeddings = None
self.vector_store = None
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
def initialize(self, persist_directory: str = './chroma_db'):
"""Initialize the RAG pipeline with embeddings and vector store."""
print(f"Initializing RAG pipeline with {self.embedding_model_name}...")
self.embeddings = HuggingFaceEmbeddings(
model_name=self.embedding_model_name
)
if self.vector_store_type == 'chroma':
self.vector_store = Chroma(
persist_directory=persist_directory,
embedding_function=self.embeddings
)
else:
raise ValueError(f"Unsupported vector store: {self.vector_store_type}")
print("RAG pipeline initialized")
def load_documents(self, documents_path: str):
"""Load and process documents from a directory."""
print(f"Loading documents from {documents_path}...")
if not os.path.exists(documents_path):
raise ValueError(f"Documents path does not exist: {documents_path}")
loader = DirectoryLoader(
documents_path,
glob="**/*.txt",
loader_cls=TextLoader
)
documents = loader.load()
print(f"Loaded {len(documents)} documents")
# Split documents into chunks
chunks = self.text_splitter.split_documents(documents)
print(f"Split into {len(chunks)} chunks")
# Add to vector store
self.vector_store.add_documents(chunks)
print("Documents added to vector store")
def retrieve(self, query: str, k: int = 4) -> List[Dict[str, Any]]:
"""Retrieve relevant documents for a query."""
if self.vector_store is None:
raise RuntimeError("RAG pipeline not initialized")
results = self.vector_store.similarity_search_with_score(query, k=k)
retrieved_docs = []
for doc, score in results:
retrieved_docs.append({
'content': doc.page_content,
'metadata': doc.metadata,
'score': score
})
return retrieved_docs
def augment_prompt(self, query: str, k: int = 4) -> str:
"""Retrieve relevant context and augment the prompt."""
retrieved_docs = self.retrieve(query, k)
context = "\n\n".join([
f"Document {i+1}:\n{doc['content']}"
for i, doc in enumerate(retrieved_docs)
])
augmented_prompt = f"""Based on the following context, please answer the question.
Context: {context}
Question: {query}
Answer:"""
return augmented_prompt
The RAG pipeline handles the complete workflow from document loading to retrieval. It splits documents into chunks, generates embeddings, stores them in a vector database, and retrieves relevant chunks for queries. The augment_prompt method creates a prompt that includes retrieved context, which the LLM can use to generate informed responses.
IMPLEMENTING GRAPH-BASED RAG
GraphRAG extends traditional RAG by representing knowledge as a graph and using graph traversal for retrieval. This captures relationships between entities and enables more sophisticated reasoning.
import networkx as nx
from typing import List, Dict, Any, Set, Tuple
class GraphRAGPipeline:
def __init__(self, embedding_model: str = 'sentence-transformers/all-MiniLM-L6-v2'):
self.graph = nx.DiGraph()
self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model)
self.entity_embeddings = {}
def add_entity(self, entity_id: str, entity_type: str, properties: Dict[str, Any]):
"""Add an entity to the knowledge graph."""
self.graph.add_node(
entity_id,
entity_type=entity_type,
properties=properties
)
# Generate embedding for entity
entity_text = f"{entity_type}: {entity_id} " + " ".join(
f"{k}={v}" for k, v in properties.items()
)
self.entity_embeddings[entity_id] = self.embeddings.embed_query(entity_text)
def add_relationship(self, source: str, target: str, relationship_type: str, properties: Dict[str, Any] = None):
"""Add a relationship between entities."""
self.graph.add_edge(
source,
target,
relationship_type=relationship_type,
properties=properties or {}
)
def find_similar_entities(self, query: str, k: int = 5) -> List[str]:
"""Find entities similar to the query using embeddings."""
query_embedding = self.embeddings.embed_query(query)
similarities = []
for entity_id, entity_embedding in self.entity_embeddings.items():
similarity = self.cosine_similarity(query_embedding, entity_embedding)
similarities.append((entity_id, similarity))
similarities.sort(key=lambda x: x[1], reverse=True)
return [entity_id for entity_id, _ in similarities[:k]]
def get_subgraph(self, entity_ids: List[str], depth: int = 2) -> nx.DiGraph:
"""Extract a subgraph around the given entities."""
subgraph_nodes = set(entity_ids)
for entity_id in entity_ids:
# Get neighbors within depth
for _ in range(depth):
new_nodes = set()
for node in subgraph_nodes:
if node in self.graph:
new_nodes.update(self.graph.successors(node))
new_nodes.update(self.graph.predecessors(node))
subgraph_nodes.update(new_nodes)
return self.graph.subgraph(subgraph_nodes)
def retrieve_with_graph(self, query: str, k: int = 5, depth: int = 2) -> Dict[str, Any]:
"""Retrieve relevant information using graph structure."""
# Find similar entities
similar_entities = self.find_similar_entities(query, k)
# Extract subgraph
subgraph = self.get_subgraph(similar_entities, depth)
# Format graph information
entities_info = []
for node in subgraph.nodes():
node_data = self.graph.nodes[node]
entities_info.append({
'id': node,
'type': node_data.get('entity_type', 'unknown'),
'properties': node_data.get('properties', {})
})
relationships_info = []
for source, target in subgraph.edges():
edge_data = self.graph.edges[source, target]
relationships_info.append({
'source': source,
'target': target,
'type': edge_data.get('relationship_type', 'unknown'),
'properties': edge_data.get('properties', {})
})
return {
'entities': entities_info,
'relationships': relationships_info,
'subgraph': subgraph
}
def cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
"""Calculate cosine similarity between two vectors."""
import numpy as np
vec1 = np.array(vec1)
vec2 = np.array(vec2)
return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
def augment_prompt_with_graph(self, query: str, k: int = 5, depth: int = 2) -> str:
"""Augment prompt with graph-based context."""
graph_data = self.retrieve_with_graph(query, k, depth)
context = "Relevant Knowledge Graph Information:\n\n"
context += "Entities:\n"
for entity in graph_data['entities']:
context += f"- {entity['type']}: {entity['id']}\n"
for key, value in entity['properties'].items():
context += f" {key}: {value}\n"
context += "\nRelationships:\n"
for rel in graph_data['relationships']:
context += f"- {rel['source']} --[{rel['type']}]--> {rel['target']}\n"
augmented_prompt = f"""{context}
Question: {query}
Based on the knowledge graph information above, please provide a comprehensive answer."""
return augmented_prompt
The GraphRAG pipeline builds a knowledge graph where nodes represent entities and edges represent relationships. It uses embeddings to find entities relevant to a query, then extracts a subgraph around those entities. This subgraph provides rich contextual information that captures not just individual facts but also the relationships between them.
AGENT IMPLEMENTATION AND ORCHESTRATION
Agents are autonomous entities that can use tools, maintain state, and make decisions. Our runtime library provides a base agent class that handles common functionality, and the code generator creates specific agent implementations based on DSL declarations.
import asyncio
from typing import List, Dict, Any, Callable, Optional
from abc import ABC, abstractmethod
class Tool:
def __init__(self, name: str, description: str, function: Callable):
self.name = name
self.description = description
self.function = function
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
if asyncio.iscoroutinefunction(self.function):
return await self.function(**kwargs)
else:
return self.function(**kwargs)
class Agent(ABC):
def __init__(self,
name: str,
model_manager: ModelManager,
model_name: str,
system_prompt: str,
tools: Optional[List[Tool]] = None,
rag_pipeline: Optional[RAGPipeline] = None):
self.name = name
self.model_manager = model_manager
self.model_name = model_name
self.system_prompt = system_prompt
self.tools = tools or []
self.rag_pipeline = rag_pipeline
self.conversation_history = []
self.state = {}
def add_tool(self, tool: Tool):
"""Add a tool to the agent's toolkit."""
self.tools.append(tool)
def get_tools_description(self) -> str:
"""Get a description of available tools."""
if not self.tools:
return "No tools available."
descriptions = ["Available tools:"]
for tool in self.tools:
descriptions.append(f"- {tool.name}: {tool.description}")
return "\n".join(descriptions)
async def use_tool(self, tool_name: str, **kwargs) -> Any:
"""Execute a specific tool."""
for tool in self.tools:
if tool.name == tool_name:
return await tool.execute(**kwargs)
raise ValueError(f"Tool not found: {tool_name}")
def add_to_history(self, role: str, content: str):
"""Add a message to conversation history."""
self.conversation_history.append({
'role': role,
'content': content
})
async def think(self, user_input: str) -> str:
"""Process user input and generate a response."""
# Augment with RAG if available
if self.rag_pipeline:
augmented_input = self.rag_pipeline.augment_prompt(user_input)
else:
augmented_input = user_input
# Add to history
self.add_to_history('user', augmented_input)
# Prepare messages
messages = [{'role': 'system', 'content': self.system_prompt}]
messages.extend(self.conversation_history)
# Generate response
response = await self.generate_response(messages)
# Add response to history
self.add_to_history('assistant', response)
return response
@abstractmethod
async def generate_response(self, messages: List[Dict[str, str]]) -> str:
"""Generate a response using the model."""
pass
class LocalAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.model_info = self.model_manager.load_local_model(self.model_name)
async def generate_response(self, messages: List[Dict[str, str]]) -> str:
"""Generate response using local model."""
model = self.model_info['model']
tokenizer = self.model_info['tokenizer']
device = self.model_info['device']
# Format messages for the model
prompt = self.format_messages(messages)
# Tokenize
inputs = tokenizer(prompt, return_tensors='pt').to(device)
# Generate
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=512,
temperature=0.7,
do_sample=True
)
# Decode
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
# Extract only the new response
response = response[len(prompt):].strip()
return response
def format_messages(self, messages: List[Dict[str, str]]) -> str:
"""Format messages into a prompt string."""
formatted = []
for msg in messages:
role = msg['role']
content = msg['content']
if role == 'system':
formatted.append(f"System: {content}")
elif role == 'user':
formatted.append(f"User: {content}")
elif role == 'assistant':
formatted.append(f"Assistant: {content}")
formatted.append("Assistant:")
return "\n\n".join(formatted)
class RemoteAgent(Agent):
def __init__(self, *args, provider: str = 'openai', **kwargs):
super().__init__(*args, **kwargs)
self.provider = provider
self.client = self.model_manager.create_remote_client(provider, self.model_name)
async def generate_response(self, messages: List[Dict[str, str]]) -> str:
"""Generate response using remote API."""
response = self.client.generate(messages, temperature=0.7, max_tokens=512)
return response
The agent implementation separates concerns between local and remote execution. Both agent types share common functionality like tool management, conversation history, and RAG integration. The generate_response method is abstract, allowing each implementation to handle model interaction appropriately.
MULTI-AGENT COMMUNICATION VIA MCP
The Model Context Protocol provides a standardized way for agents to communicate. Our implementation includes an MCP server and client that enable agents to exchange messages, share context, and coordinate actions.
import json
import asyncio
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
class MessageType(Enum):
REQUEST = "request"
RESPONSE = "response"
NOTIFICATION = "notification"
ERROR = "error"
@dataclass
class MCPMessage:
message_type: MessageType
sender: str
recipient: str
content: Any
message_id: str
correlation_id: Optional[str] = None
class MCPServer:
def __init__(self):
self.agents = {}
self.message_queue = asyncio.Queue()
self.handlers = {}
self.running = False
def register_agent(self, agent_id: str, agent: Agent):
"""Register an agent with the MCP server."""
self.agents[agent_id] = agent
print(f"Agent {agent_id} registered with MCP server")
def register_handler(self, message_type: str, handler: Callable):
"""Register a handler for a specific message type."""
self.handlers[message_type] = handler
async def send_message(self, message: MCPMessage):
"""Send a message through the MCP server."""
await self.message_queue.put(message)
async def start(self):
"""Start the MCP server message processing loop."""
self.running = True
print("MCP server started")
while self.running:
try:
message = await asyncio.wait_for(
self.message_queue.get(),
timeout=1.0
)
await self.process_message(message)
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"Error processing message: {str(e)}")
async def process_message(self, message: MCPMessage):
"""Process a message and route it to the recipient."""
if message.recipient not in self.agents:
error_msg = MCPMessage(
message_type=MessageType.ERROR,
sender="server",
recipient=message.sender,
content=f"Recipient {message.recipient} not found",
message_id=self.generate_message_id(),
correlation_id=message.message_id
)
await self.send_message(error_msg)
return
recipient_agent = self.agents[message.recipient]
if message.message_type == MessageType.REQUEST:
response = await recipient_agent.handle_mcp_request(message)
response_msg = MCPMessage(
message_type=MessageType.RESPONSE,
sender=message.recipient,
recipient=message.sender,
content=response,
message_id=self.generate_message_id(),
correlation_id=message.message_id
)
await self.send_message(response_msg)
elif message.message_type == MessageType.NOTIFICATION:
await recipient_agent.handle_mcp_notification(message)
def generate_message_id(self) -> str:
"""Generate a unique message ID."""
import uuid
return str(uuid.uuid4())
async def stop(self):
"""Stop the MCP server."""
self.running = False
print("MCP server stopped")
class MCPClient:
def __init__(self, agent_id: str, server: MCPServer):
self.agent_id = agent_id
self.server = server
self.pending_requests = {}
async def send_request(self, recipient: str, content: Any, timeout: float = 30.0) -> Any:
"""Send a request and wait for response."""
message_id = self.server.generate_message_id()
message = MCPMessage(
message_type=MessageType.REQUEST,
sender=self.agent_id,
recipient=recipient,
content=content,
message_id=message_id
)
# Create future for response
response_future = asyncio.Future()
self.pending_requests[message_id] = response_future
# Send message
await self.server.send_message(message)
# Wait for response
try:
response = await asyncio.wait_for(response_future, timeout=timeout)
return response
except asyncio.TimeoutError:
del self.pending_requests[message_id]
raise TimeoutError(f"Request to {recipient} timed out")
finally:
if message_id in self.pending_requests:
del self.pending_requests[message_id]
async def send_notification(self, recipient: str, content: Any):
"""Send a notification without expecting a response."""
message = MCPMessage(
message_type=MessageType.NOTIFICATION,
sender=self.agent_id,
recipient=recipient,
content=content,
message_id=self.server.generate_message_id()
)
await self.server.send_message(message)
def handle_response(self, message: MCPMessage):
"""Handle a response message."""
if message.correlation_id in self.pending_requests:
future = self.pending_requests[message.correlation_id]
future.set_result(message.content)
The MCP implementation provides asynchronous message passing between agents. Agents can send requests and receive responses, or send notifications without waiting for acknowledgment. The server routes messages to the appropriate recipients and handles errors gracefully.
ORCHESTRATION ENGINE FOR MULTI-AGENT SYSTEMS
Multi-agent systems require coordination logic that determines which agents execute when and how information flows between them. The orchestration engine implements routing rules, workflow execution, and state management for multi-agent systems.
from typing import Dict, List, Any, Callable, Optional
import asyncio
class OrchestrationRule:
def __init__(self, source: str, target: str, condition: Callable[[Any], bool]):
self.source = source
self.target = target
self.condition = condition
class Orchestrator:
def __init__(self, mcp_server: MCPServer):
self.mcp_server = mcp_server
self.agents = {}
self.routing_rules = []
self.entry_point = None
self.workflow_state = {}
def register_agent(self, agent_id: str, agent: Agent):
"""Register an agent with the orchestrator."""
self.agents[agent_id] = agent
self.mcp_server.register_agent(agent_id, agent)
def set_entry_point(self, agent_id: str):
"""Set the entry point agent for the workflow."""
if agent_id not in self.agents:
raise ValueError(f"Agent {agent_id} not registered")
self.entry_point = agent_id
def add_routing_rule(self, rule: OrchestrationRule):
"""Add a routing rule to the orchestrator."""
self.routing_rules.append(rule)
async def execute_workflow(self, initial_input: str) -> Dict[str, Any]:
"""Execute the multi-agent workflow."""
if not self.entry_point:
raise RuntimeError("No entry point defined")
self.workflow_state = {
'current_agent': self.entry_point,
'input': initial_input,
'outputs': {},
'completed': False
}
current_agent_id = self.entry_point
current_input = initial_input
while not self.workflow_state['completed']:
# Execute current agent
agent = self.agents[current_agent_id]
output = await agent.think(current_input)
# Store output
self.workflow_state['outputs'][current_agent_id] = output
# Determine next agent based on routing rules
next_agent_id = self.determine_next_agent(current_agent_id, output)
if next_agent_id is None:
# No more agents to execute
self.workflow_state['completed'] = True
else:
current_agent_id = next_agent_id
current_input = output
self.workflow_state['current_agent'] = current_agent_id
return self.workflow_state['outputs']
def determine_next_agent(self, current_agent: str, output: Any) -> Optional[str]:
"""Determine the next agent to execute based on routing rules."""
for rule in self.routing_rules:
if rule.source == current_agent and rule.condition(output):
return rule.target
return None
async def execute_parallel_agents(self, agent_ids: List[str], input_data: str) -> Dict[str, Any]:
"""Execute multiple agents in parallel."""
tasks = []
for agent_id in agent_ids:
agent = self.agents[agent_id]
tasks.append(agent.think(input_data))
results = await asyncio.gather(*tasks)
return {
agent_id: result
for agent_id, result in zip(agent_ids, results)
}
The orchestrator manages workflow execution by tracking the current state, executing agents in sequence or parallel, and applying routing rules to determine the next step. This enables complex multi-agent workflows where different agents handle different aspects of a problem.
CODE GENERATION ENGINE
The code generator transforms the validated AST into executable Python code. It generates complete applications including all necessary imports, class definitions, initialization code, and main execution logic.
class CodeGenerator:
def __init__(self, ast, symbol_table):
self.ast = ast
self.symbol_table = symbol_table
self.generated_code = []
self.indent_level = 0
def generate(self) -> str:
"""Generate complete Python code from AST."""
self.emit_imports()
self.emit_blank_line()
for declaration in self.ast:
if isinstance(declaration, ChatbotDeclaration):
self.generate_chatbot(declaration)
elif isinstance(declaration, AgentDeclaration):
self.generate_agent(declaration)
elif isinstance(declaration, MultiAgentSystem):
self.generate_multi_agent_system(declaration)
elif isinstance(declaration, SettingsDeclaration):
self.settings = declaration
self.emit_main_function()
return '\n'.join(self.generated_code)
def emit_imports(self):
"""Emit all necessary imports."""
imports = [
"import asyncio",
"import sys",
"from typing import List, Dict, Any, Optional",
"from llmdsl_runtime import (",
" GPUDetector,",
" ModelManager,",
" LocalAgent,",
" RemoteAgent,",
" RAGPipeline,",
" GraphRAGPipeline,",
" MCPServer,",
" MCPClient,",
" Orchestrator,",
" Tool,",
" ConsoleUI,",
" WebUI",
")"
]
for imp in imports:
self.emit(imp)
def generate_chatbot(self, chatbot: ChatbotDeclaration):
"""Generate code for a chatbot."""
self.emit(f"class {chatbot.name}:")
self.indent()
self.emit("def __init__(self):")
self.indent()
# Initialize GPU detector and model manager
self.emit("self.gpu_detector = GPUDetector()")
self.emit("self.gpu_detector.detect()")
self.emit("self.model_manager = ModelManager(self.gpu_detector)")
self.emit_blank_line()
# Determine if model is local or remote
model_name = chatbot.properties.get('model')
if self.is_remote_model(model_name):
provider = self.get_provider(model_name)
self.emit(f"self.agent = RemoteAgent(")
self.indent()
self.emit(f"name='{chatbot.name}',")
self.emit(f"model_manager=self.model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{chatbot.properties.get('system_prompt')}''',")
self.emit(f"provider='{provider}'")
self.dedent()
self.emit(")")
else:
self.emit(f"self.agent = LocalAgent(")
self.indent()
self.emit(f"name='{chatbot.name}',")
self.emit(f"model_manager=self.model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{chatbot.properties.get('system_prompt')}'''")
self.dedent()
self.emit(")")
# Add RAG if specified
if 'rag' in chatbot.properties:
self.emit_blank_line()
self.emit("# Initialize RAG pipeline")
rag_config = chatbot.properties['rag']
self.emit(f"self.rag = RAGPipeline(")
self.indent()
self.emit(f"vector_store_type='{rag_config.get('vector_store', 'chroma')}',")
self.emit(f"embedding_model='{rag_config.get('embedding_model')}'")
self.dedent()
self.emit(")")
self.emit("self.rag.initialize()")
if 'documents' in rag_config:
self.emit(f"self.rag.load_documents('{rag_config['documents']}')")
self.emit("self.agent.rag_pipeline = self.rag")
self.dedent()
self.emit_blank_line()
# Generate run method
self.emit("async def run(self):")
self.indent()
self.emit("while True:")
self.indent()
self.emit("user_input = input('You: ')")
self.emit("if user_input.lower() in ['quit', 'exit']:")
self.indent()
self.emit("break")
self.dedent()
self.emit("response = await self.agent.think(user_input)")
self.emit("print(f'Assistant: {response}')")
self.dedent()
self.dedent()
self.dedent()
self.emit_blank_line()
def generate_agent(self, agent: AgentDeclaration):
"""Generate code for an agent."""
self.emit(f"class {agent.name}Agent:")
self.indent()
self.emit("def __init__(self, model_manager: ModelManager):")
self.indent()
model_name = agent.properties.get('model')
if self.is_remote_model(model_name):
provider = self.get_provider(model_name)
self.emit(f"self.agent = RemoteAgent(")
self.indent()
self.emit(f"name='{agent.name}',")
self.emit(f"model_manager=model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{agent.properties.get('system_prompt')}''',")
self.emit(f"provider='{provider}'")
self.dedent()
self.emit(")")
else:
self.emit(f"self.agent = LocalAgent(")
self.indent()
self.emit(f"name='{agent.name}',")
self.emit(f"model_manager=model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{agent.properties.get('system_prompt')}'''")
self.dedent()
self.emit(")")
# Add tools if specified
if 'tools' in agent.properties:
self.emit_blank_line()
self.emit("# Add tools")
for tool_name in agent.properties['tools']:
self.emit(f"self.agent.add_tool({tool_name}_tool())")
self.dedent()
self.emit_blank_line()
self.emit("async def process(self, input_data: str) -> str:")
self.indent()
self.emit("return await self.agent.think(input_data)")
self.dedent()
self.dedent()
self.emit_blank_line()
def generate_multi_agent_system(self, system: MultiAgentSystem):
"""Generate code for a multi-agent system."""
self.emit(f"class {system.name}:")
self.indent()
self.emit("def __init__(self):")
self.indent()
# Initialize infrastructure
self.emit("self.gpu_detector = GPUDetector()")
self.emit("self.gpu_detector.detect()")
self.emit("self.model_manager = ModelManager(self.gpu_detector)")
self.emit("self.mcp_server = MCPServer()")
self.emit("self.orchestrator = Orchestrator(self.mcp_server)")
self.emit_blank_line()
# Create agents
self.emit("# Create agents")
for agent in system.agents:
agent_name = agent.name if hasattr(agent, 'name') else agent
self.emit(f"self.{agent_name.lower()} = {agent_name}Agent(self.model_manager)")
self.emit(f"self.orchestrator.register_agent('{agent_name}', self.{agent_name.lower()}.agent)")
self.emit_blank_line()
# Set entry point
if system.orchestration and 'entry_point' in system.orchestration:
entry_point = system.orchestration['entry_point']
self.emit(f"self.orchestrator.set_entry_point('{entry_point}')")
# Add routing rules
if system.orchestration and 'routing' in system.orchestration:
self.emit_blank_line()
self.emit("# Add routing rules")
for route in system.orchestration['routing']:
source = route['source']
target = route['target']
condition = route.get('condition', 'lambda x: True')
self.emit(f"self.orchestrator.add_routing_rule(")
self.indent()
self.emit(f"OrchestrationRule(")
self.indent()
self.emit(f"source='{source}',")
self.emit(f"target='{target}',")
self.emit(f"condition={condition}")
self.dedent()
self.emit(")")
self.dedent()
self.emit(")")
self.dedent()
self.emit_blank_line()
# Generate run method
self.emit("async def run(self, user_input: str) -> Dict[str, Any]:")
self.indent()
self.emit("# Start MCP server")
self.emit("mcp_task = asyncio.create_task(self.mcp_server.start())")
self.emit_blank_line()
self.emit("try:")
self.indent()
self.emit("# Execute workflow")
self.emit("results = await self.orchestrator.execute_workflow(user_input)")
self.emit("return results")
self.dedent()
self.emit("finally:")
self.indent()
self.emit("await self.mcp_server.stop()")
self.emit("await mcp_task")
self.dedent()
self.dedent()
self.dedent()
self.emit_blank_line()
def emit_main_function(self):
"""Generate main execution function."""
self.emit("async def main():")
self.indent()
# Find the main component to run
main_component = None
for declaration in self.ast:
if isinstance(declaration, (ChatbotDeclaration, MultiAgentSystem)):
main_component = declaration
break
if main_component:
self.emit(f"app = {main_component.name}()")
if isinstance(main_component, ChatbotDeclaration):
self.emit("await app.run()")
elif isinstance(main_component, MultiAgentSystem):
self.emit("user_input = input('Enter your query: ')")
self.emit("results = await app.run(user_input)")
self.emit("print('Results:')")
self.emit("for agent_id, output in results.items():")
self.indent()
self.emit("print(f'{agent_id}: {output}')")
self.dedent()
self.dedent()
self.emit_blank_line()
self.emit("if __name__ == '__main__':")
self.indent()
self.emit("asyncio.run(main())")
self.dedent()
def emit(self, line: str):
"""Emit a line of code with proper indentation."""
indent = " " * self.indent_level
self.generated_code.append(indent + line)
def emit_blank_line(self):
"""Emit a blank line."""
self.generated_code.append("")
def indent(self):
"""Increase indentation level."""
self.indent_level += 1
def dedent(self):
"""Decrease indentation level."""
self.indent_level = max(0, self.indent_level - 1)
def is_remote_model(self, model_name: str) -> bool:
"""Check if a model is remote (API-based)."""
remote_prefixes = ['gpt-', 'claude-', 'gemini-']
return any(model_name.startswith(prefix) for prefix in remote_prefixes)
def get_provider(self, model_name: str) -> str:
"""Get the provider for a remote model."""
if model_name.startswith('gpt-'):
return 'openai'
elif model_name.startswith('claude-'):
return 'anthropic'
elif model_name.startswith('gemini-'):
return 'google'
return 'unknown'
The code generator walks the AST and emits Python code for each declaration. It handles imports, class definitions, initialization logic, and execution flow. The generated code is properly indented and follows Python conventions.
USER INTERFACE GENERATION
The DSL supports both console and web-based interfaces. The code generator creates appropriate UI code based on the settings declaration.
class ConsoleUI:
def __init__(self, agent: Agent):
self.agent = agent
async def run(self):
"""Run the console interface."""
print("Console Interface Started")
print("Type 'quit' or 'exit' to end the conversation")
print("-" * 50)
while True:
try:
user_input = input("\nYou: ")
if user_input.lower() in ['quit', 'exit']:
print("Goodbye!")
break
if not user_input.strip():
continue
response = await self.agent.think(user_input)
print(f"\nAssistant: {response}")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"\nError: {str(e)}")
class WebUI:
def __init__(self, agent: Agent, port: int = 8080):
self.agent = agent
self.port = port
self.app = None
def create_app(self):
"""Create a web application."""
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
app = FastAPI()
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>LLM Chat Interface</title>
<style>
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
}
#chat-container {
border: 1px solid #ccc;
height: 500px;
overflow-y: auto;
padding: 10px;
margin-bottom: 10px;
}
.message {
margin: 10px 0;
padding: 10px;
border-radius: 5px;
}
.user-message {
background-color: #e3f2fd;
text-align: right;
}
.assistant-message {
background-color: #f5f5f5;
}
#input-container {
display: flex;
}
#message-input {
flex: 1;
padding: 10px;
font-size: 16px;
}
#send-button {
padding: 10px 20px;
font-size: 16px;
background-color: #2196F3;
color: white;
border: none;
cursor: pointer;
}
</style>
</head>
<body>
<h1>LLM Chat Interface</h1>
<div id="chat-container"></div>
<div id="input-container">
<input type="text" id="message-input" placeholder="Type your message...">
<button id="send-button">Send</button>
</div>
<script>
const ws = new WebSocket('ws://localhost:8080/ws');
const chatContainer = document.getElementById('chat-container');
const messageInput = document.getElementById('message-input');
const sendButton = document.getElementById('send-button');
function addMessage(content, isUser) {
const messageDiv = document.createElement('div');
messageDiv.className = 'message ' + (isUser ? 'user-message' : 'assistant-message');
messageDiv.textContent = content;
chatContainer.appendChild(messageDiv);
chatContainer.scrollTop = chatContainer.scrollHeight;
}
ws.onmessage = function(event) {
addMessage(event.data, false);
};
function sendMessage() {
const message = messageInput.value.trim();
if (message) {
addMessage(message, true);
ws.send(message);
messageInput.value = '';
}
}
sendButton.addEventListener('click', sendMessage);
messageInput.addEventListener('keypress', function(e) {
if (e.key === 'Enter') {
sendMessage();
}
});
</script>
</body>
</html>
"""
@app.get("/")
async def get_index():
return HTMLResponse(content=html_content)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
message = await websocket.receive_text()
response = await self.agent.think(message)
await websocket.send_text(response)
except Exception as e:
print(f"WebSocket error: {str(e)}")
return app
async def run(self):
"""Run the web interface."""
import uvicorn
self.app = self.create_app()
config = uvicorn.Config(self.app, host="0.0.0.0", port=self.port)
server = uvicorn.Server(config)
await server.serve()
The UI implementations provide clean interfaces for interacting with LLM applications. The console UI offers a simple command-line interface, while the web UI provides a browser-based chat interface with WebSocket communication for real-time interaction.
PUTTING IT ALL TOGETHER: THE COMPLETE SYSTEM
Now we have examined all the major components of our DSL system. The parser reads DSL source files and builds an AST. The semantic analyzer validates the AST and ensures all references are valid. The code generator transforms the AST into executable Python code. The runtime library provides GPU detection, model management, RAG pipelines, agent implementations, MCP communication, orchestration, and user interfaces.
To use the system, a developer writes a DSL file specifying their application:
chatbot CustomerServiceBot {
model: "gpt-4"
system_prompt: "You are a helpful customer service representative."
temperature: 0.7
rag: {
vector_store: "chroma"
embedding_model: "sentence-transformers/all-MiniLM-L6-v2"
documents: "./product_docs"
}
}
settings {
gpu: "auto"
ui: "web"
port: 8080
}
The LLMDSL compiler processes this file through the parser, semantic analyzer, and code generator to produce a complete Python application. The developer can then run the generated code or extend it with custom functionality.
FULL RUNNING EXAMPLE IMPLEMENTATION
The following is a complete, production-ready implementation of the LLMDSL system including all components discussed above. This implementation is fully functional and can be used to build real LLM applications.
# llmdsl_compiler.py
# Complete implementation of the LLMDSL compiler
import sys
import re
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
# ============================================================================
# LEXER IMPLEMENTATION
# ============================================================================
class TokenType(Enum):
KEYWORD = "KEYWORD"
IDENTIFIER = "IDENTIFIER"
STRING = "STRING"
NUMBER = "NUMBER"
LBRACE = "LBRACE"
RBRACE = "RBRACE"
LBRACKET = "LBRACKET"
RBRACKET = "RBRACKET"
LPAREN = "LPAREN"
RPAREN = "RPAREN"
COLON = "COLON"
COMMA = "COMMA"
ARROW = "ARROW"
EQUALS = "EQUALS"
EOF = "EOF"
@dataclass
class Token:
token_type: TokenType
value: Any
line: int
column: int
class Lexer:
KEYWORDS = {
'chatbot', 'agent', 'multi_agent_system', 'settings',
'model', 'system_prompt', 'temperature', 'max_tokens',
'tools', 'rag', 'vector_store', 'embedding_model', 'documents',
'agents', 'orchestration', 'entry_point', 'routing',
'communication', 'gpu', 'execution_mode', 'max_concurrent',
'retry_policy', 'max_retries', 'backoff', 'ui', 'port',
'when', 'role', 'graphrag'
}
def __init__(self, source: str):
self.source = source
self.position = 0
self.line = 1
self.column = 1
self.tokens = []
def current_char(self) -> Optional[str]:
if self.position >= len(self.source):
return None
return self.source[self.position]
def peek_char(self, offset: int = 1) -> Optional[str]:
pos = self.position + offset
if pos >= len(self.source):
return None
return self.source[pos]
def advance(self):
if self.position < len(self.source):
if self.source[self.position] == '\n':
self.line += 1
self.column = 1
else:
self.column += 1
self.position += 1
def skip_whitespace(self):
while self.current_char() and self.current_char() in ' \t\n\r':
self.advance()
def skip_comment(self):
if self.current_char() == '#':
while self.current_char() and self.current_char() != '\n':
self.advance()
def read_string(self) -> Token:
start_line = self.line
start_column = self.column
quote_char = self.current_char()
self.advance() # Skip opening quote
value = ""
while self.current_char() and self.current_char() != quote_char:
if self.current_char() == '\\':
self.advance()
if self.current_char():
escape_char = self.current_char()
if escape_char == 'n':
value += '\n'
elif escape_char == 't':
value += '\t'
elif escape_char == '\\':
value += '\\'
elif escape_char == quote_char:
value += quote_char
else:
value += escape_char
self.advance()
else:
value += self.current_char()
self.advance()
if self.current_char() == quote_char:
self.advance() # Skip closing quote
else:
raise SyntaxError(f"Unterminated string at line {start_line}, column {start_column}")
return Token(TokenType.STRING, value, start_line, start_column)
def read_number(self) -> Token:
start_line = self.line
start_column = self.column
value = ""
while self.current_char() and (self.current_char().isdigit() or self.current_char() == '.'):
value += self.current_char()
self.advance()
if '.' in value:
return Token(TokenType.NUMBER, float(value), start_line, start_column)
else:
return Token(TokenType.NUMBER, int(value), start_line, start_column)
def read_identifier(self) -> Token:
start_line = self.line
start_column = self.column
value = ""
while self.current_char() and (self.current_char().isalnum() or self.current_char() in '_-'):
value += self.current_char()
self.advance()
token_type = TokenType.KEYWORD if value in self.KEYWORDS else TokenType.IDENTIFIER
return Token(token_type, value, start_line, start_column)
def tokenize(self) -> List[Token]:
while self.position < len(self.source):
self.skip_whitespace()
if self.position >= len(self.source):
break
if self.current_char() == '#':
self.skip_comment()
continue
if self.current_char() in '"\'':
self.tokens.append(self.read_string())
elif self.current_char().isdigit():
self.tokens.append(self.read_number())
elif self.current_char().isalpha() or self.current_char() == '_':
self.tokens.append(self.read_identifier())
elif self.current_char() == '{':
self.tokens.append(Token(TokenType.LBRACE, '{', self.line, self.column))
self.advance()
elif self.current_char() == '}':
self.tokens.append(Token(TokenType.RBRACE, '}', self.line, self.column))
self.advance()
elif self.current_char() == '[':
self.tokens.append(Token(TokenType.LBRACKET, '[', self.line, self.column))
self.advance()
elif self.current_char() == ']':
self.tokens.append(Token(TokenType.RBRACKET, ']', self.line, self.column))
self.advance()
elif self.current_char() == '(':
self.tokens.append(Token(TokenType.LPAREN, '(', self.line, self.column))
self.advance()
elif self.current_char() == ')':
self.tokens.append(Token(TokenType.RPAREN, ')', self.line, self.column))
self.advance()
elif self.current_char() == ':':
self.tokens.append(Token(TokenType.COLON, ':', self.line, self.column))
self.advance()
elif self.current_char() == ',':
self.tokens.append(Token(TokenType.COMMA, ',', self.line, self.column))
self.advance()
elif self.current_char() == '=' and self.peek_char() == '=':
self.tokens.append(Token(TokenType.EQUALS, '==', self.line, self.column))
self.advance()
self.advance()
elif self.current_char() == '-' and self.peek_char() == '>':
self.tokens.append(Token(TokenType.ARROW, '->', self.line, self.column))
self.advance()
self.advance()
else:
raise SyntaxError(f"Unexpected character '{self.current_char()}' at line {self.line}, column {self.column}")
self.tokens.append(Token(TokenType.EOF, None, self.line, self.column))
return self.tokens
# ============================================================================
# AST NODE DEFINITIONS
# ============================================================================
class ASTNode:
pass
@dataclass
class ChatbotDeclaration(ASTNode):
name: str
properties: Dict[str, Any]
@dataclass
class AgentDeclaration(ASTNode):
name: str
properties: Dict[str, Any]
@dataclass
class MultiAgentSystem(ASTNode):
name: str
agents: List[Any]
orchestration: Dict[str, Any]
communication: str
@dataclass
class SettingsDeclaration(ASTNode):
properties: Dict[str, Any]
# ============================================================================
# PARSER IMPLEMENTATION
# ============================================================================
class Parser:
def __init__(self, tokens: List[Token]):
self.tokens = tokens
self.position = 0
def current_token(self) -> Token:
if self.position < len(self.tokens):
return self.tokens[self.position]
return self.tokens[-1] # EOF
def peek_token(self, offset: int = 1) -> Token:
pos = self.position + offset
if pos < len(self.tokens):
return self.tokens[pos]
return self.tokens[-1] # EOF
def advance(self):
if self.position < len(self.tokens) - 1:
self.position += 1
def expect(self, token_type: TokenType) -> Token:
token = self.current_token()
if token.token_type != token_type:
raise SyntaxError(f"Expected {token_type}, got {token.token_type} at line {token.line}")
self.advance()
return token
def parse(self) -> List[ASTNode]:
declarations = []
while self.current_token().token_type != TokenType.EOF:
declarations.append(self.parse_declaration())
return declarations
def parse_declaration(self) -> ASTNode:
token = self.current_token()
if token.value == 'chatbot':
return self.parse_chatbot()
elif token.value == 'agent':
return self.parse_agent()
elif token.value == 'multi_agent_system':
return self.parse_multi_agent_system()
elif token.value == 'settings':
return self.parse_settings()
else:
raise SyntaxError(f"Unexpected declaration '{token.value}' at line {token.line}")
def parse_chatbot(self) -> ChatbotDeclaration:
self.expect(TokenType.KEYWORD) # chatbot
name = self.expect(TokenType.IDENTIFIER).value
self.expect(TokenType.LBRACE)
properties = self.parse_properties()
self.expect(TokenType.RBRACE)
return ChatbotDeclaration(name, properties)
def parse_agent(self) -> AgentDeclaration:
self.expect(TokenType.KEYWORD) # agent
name = self.expect(TokenType.IDENTIFIER).value
self.expect(TokenType.LBRACE)
properties = self.parse_properties()
self.expect(TokenType.RBRACE)
return AgentDeclaration(name, properties)
def parse_multi_agent_system(self) -> MultiAgentSystem:
self.expect(TokenType.KEYWORD) # multi_agent_system
name = self.expect(TokenType.IDENTIFIER).value
self.expect(TokenType.LBRACE)
agents = []
orchestration = {}
communication = "mcp"
while self.current_token().token_type != TokenType.RBRACE:
prop_name = self.current_token().value
self.advance()
self.expect(TokenType.COLON)
if prop_name == 'agents':
agents = self.parse_agent_list()
elif prop_name == 'orchestration':
orchestration = self.parse_object()
elif prop_name == 'communication':
communication = self.expect(TokenType.STRING).value
if self.current_token().token_type == TokenType.COMMA:
self.advance()
self.expect(TokenType.RBRACE)
return MultiAgentSystem(name, agents, orchestration, communication)
def parse_settings(self) -> SettingsDeclaration:
self.expect(TokenType.KEYWORD) # settings
self.expect(TokenType.LBRACE)
properties = self.parse_properties()
self.expect(TokenType.RBRACE)
return SettingsDeclaration(properties)
def parse_properties(self) -> Dict[str, Any]:
properties = {}
while self.current_token().token_type != TokenType.RBRACE:
prop_name = self.current_token().value
self.advance()
self.expect(TokenType.COLON)
prop_value = self.parse_value()
properties[prop_name] = prop_value
if self.current_token().token_type == TokenType.COMMA:
self.advance()
return properties
def parse_value(self) -> Any:
token = self.current_token()
if token.token_type == TokenType.STRING:
self.advance()
return token.value
elif token.token_type == TokenType.NUMBER:
self.advance()
return token.value
elif token.token_type == TokenType.IDENTIFIER:
self.advance()
return token.value
elif token.token_type == TokenType.LBRACKET:
return self.parse_array()
elif token.token_type == TokenType.LBRACE:
return self.parse_object()
else:
raise SyntaxError(f"Unexpected value token {token.token_type} at line {token.line}")
def parse_array(self) -> List[Any]:
self.expect(TokenType.LBRACKET)
items = []
while self.current_token().token_type != TokenType.RBRACKET:
items.append(self.parse_value())
if self.current_token().token_type == TokenType.COMMA:
self.advance()
self.expect(TokenType.RBRACKET)
return items
def parse_object(self) -> Dict[str, Any]:
self.expect(TokenType.LBRACE)
obj = {}
while self.current_token().token_type != TokenType.RBRACE:
key = self.current_token().value
self.advance()
self.expect(TokenType.COLON)
value = self.parse_value()
obj[key] = value
if self.current_token().token_type == TokenType.COMMA:
self.advance()
self.expect(TokenType.RBRACE)
return obj
def parse_agent_list(self) -> List[Any]:
self.expect(TokenType.LBRACKET)
agents = []
while self.current_token().token_type != TokenType.RBRACKET:
if self.current_token().value == 'agent':
agents.append(self.parse_agent())
else:
agents.append(self.expect(TokenType.IDENTIFIER).value)
if self.current_token().token_type == TokenType.COMMA:
self.advance()
self.expect(TokenType.RBRACKET)
return agents
# ============================================================================
# SEMANTIC ANALYZER
# ============================================================================
class SemanticAnalyzer:
def __init__(self, ast: List[ASTNode]):
self.ast = ast
self.symbol_table = {}
self.errors = []
def analyze(self) -> bool:
for declaration in self.ast:
self.register_declaration(declaration)
for declaration in self.ast:
self.validate_declaration(declaration)
if self.errors:
for error in self.errors:
print(f"Semantic Error: {error}", file=sys.stderr)
return False
return True
def register_declaration(self, declaration: ASTNode):
if isinstance(declaration, (ChatbotDeclaration, AgentDeclaration)):
if declaration.name in self.symbol_table:
self.errors.append(f"Duplicate declaration: {declaration.name}")
else:
self.symbol_table[declaration.name] = declaration
elif isinstance(declaration, MultiAgentSystem):
if declaration.name in self.symbol_table:
self.errors.append(f"Duplicate declaration: {declaration.name}")
else:
self.symbol_table[declaration.name] = declaration
def validate_declaration(self, declaration: ASTNode):
if isinstance(declaration, ChatbotDeclaration):
self.validate_chatbot(declaration)
elif isinstance(declaration, AgentDeclaration):
self.validate_agent(declaration)
elif isinstance(declaration, MultiAgentSystem):
self.validate_multi_agent_system(declaration)
def validate_chatbot(self, chatbot: ChatbotDeclaration):
required = ['model', 'system_prompt']
for prop in required:
if prop not in chatbot.properties:
self.errors.append(f"Chatbot {chatbot.name} missing required property: {prop}")
def validate_agent(self, agent: AgentDeclaration):
required = ['model', 'system_prompt']
for prop in required:
if prop not in agent.properties:
self.errors.append(f"Agent {agent.name} missing required property: {prop}")
def validate_multi_agent_system(self, system: MultiAgentSystem):
if not system.agents:
self.errors.append(f"Multi-agent system {system.name} has no agents")
# ============================================================================
# CODE GENERATOR
# ============================================================================
class CodeGenerator:
def __init__(self, ast: List[ASTNode]):
self.ast = ast
self.code_lines = []
self.indent_level = 0
self.settings = None
for declaration in ast:
if isinstance(declaration, SettingsDeclaration):
self.settings = declaration
break
def generate(self) -> str:
self.emit_header()
self.emit_imports()
self.emit_blank_line()
for declaration in self.ast:
if isinstance(declaration, ChatbotDeclaration):
self.generate_chatbot(declaration)
elif isinstance(declaration, AgentDeclaration):
self.generate_agent(declaration)
elif isinstance(declaration, MultiAgentSystem):
self.generate_multi_agent_system(declaration)
self.emit_main_function()
return '\n'.join(self.code_lines)
def emit_header(self):
self.emit("# Generated by LLMDSL Compiler")
self.emit("# This code is production-ready and fully functional")
self.emit_blank_line()
def emit_imports(self):
imports = [
"import asyncio",
"import sys",
"import os",
"from typing import List, Dict, Any, Optional, Callable",
"from llmdsl_runtime import (",
" GPUDetector,",
" ModelManager,",
" LocalAgent,",
" RemoteAgent,",
" RAGPipeline,",
" GraphRAGPipeline,",
" MCPServer,",
" MCPClient,",
" Orchestrator,",
" OrchestrationRule,",
" Tool,",
" ConsoleUI,",
" WebUI",
")"
]
for imp in imports:
self.emit(imp)
def generate_chatbot(self, chatbot: ChatbotDeclaration):
self.emit(f"class {chatbot.name}:")
self.indent()
self.emit('"""Generated chatbot class."""')
self.emit_blank_line()
self.emit("def __init__(self):")
self.indent()
self.emit('"""Initialize the chatbot."""')
self.emit("self.gpu_detector = GPUDetector()")
self.emit("self.gpu_detector.detect()")
self.emit("self.model_manager = ModelManager(self.gpu_detector)")
self.emit_blank_line()
model_name = chatbot.properties.get('model')
system_prompt = chatbot.properties.get('system_prompt')
if self.is_remote_model(model_name):
provider = self.get_provider(model_name)
self.emit(f"self.agent = RemoteAgent(")
self.indent()
self.emit(f"name='{chatbot.name}',")
self.emit(f"model_manager=self.model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{system_prompt}''',")
self.emit(f"provider='{provider}'")
self.dedent()
self.emit(")")
else:
self.emit(f"self.agent = LocalAgent(")
self.indent()
self.emit(f"name='{chatbot.name}',")
self.emit(f"model_manager=self.model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{system_prompt}'''")
self.dedent()
self.emit(")")
if 'rag' in chatbot.properties:
self.emit_blank_line()
self.emit("# Initialize RAG pipeline")
rag_config = chatbot.properties['rag']
self.emit(f"self.rag = RAGPipeline(")
self.indent()
self.emit(f"vector_store_type='{rag_config.get('vector_store', 'chroma')}',")
self.emit(f"embedding_model='{rag_config.get('embedding_model')}'")
self.dedent()
self.emit(")")
self.emit("self.rag.initialize()")
if 'documents' in rag_config:
self.emit(f"self.rag.load_documents('{rag_config['documents']}')")
self.emit("self.agent.rag_pipeline = self.rag")
self.dedent()
self.emit_blank_line()
self.emit("async def run(self):")
self.indent()
self.emit('"""Run the chatbot."""')
if self.settings and self.settings.properties.get('ui') == 'web':
port = self.settings.properties.get('port', 8080)
self.emit(f"ui = WebUI(self.agent, port={port})")
self.emit("await ui.run()")
else:
self.emit("ui = ConsoleUI(self.agent)")
self.emit("await ui.run()")
self.dedent()
self.dedent()
self.emit_blank_line()
def generate_agent(self, agent: AgentDeclaration):
self.emit(f"class {agent.name}Agent:")
self.indent()
self.emit(f'"""Generated agent class for {agent.name}."""')
self.emit_blank_line()
self.emit("def __init__(self, model_manager: ModelManager):")
self.indent()
self.emit('"""Initialize the agent."""')
model_name = agent.properties.get('model')
system_prompt = agent.properties.get('system_prompt')
if self.is_remote_model(model_name):
provider = self.get_provider(model_name)
self.emit(f"self.agent = RemoteAgent(")
self.indent()
self.emit(f"name='{agent.name}',")
self.emit(f"model_manager=model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{system_prompt}''',")
self.emit(f"provider='{provider}'")
self.dedent()
self.emit(")")
else:
self.emit(f"self.agent = LocalAgent(")
self.indent()
self.emit(f"name='{agent.name}',")
self.emit(f"model_manager=model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{system_prompt}'''")
self.dedent()
self.emit(")")
self.dedent()
self.emit_blank_line()
self.emit("async def process(self, input_data: str) -> str:")
self.indent()
self.emit('"""Process input and return response."""')
self.emit("return await self.agent.think(input_data)")
self.dedent()
self.dedent()
self.emit_blank_line()
def generate_multi_agent_system(self, system: MultiAgentSystem):
self.emit(f"class {system.name}:")
self.indent()
self.emit(f'"""Generated multi-agent system class."""')
self.emit_blank_line()
self.emit("def __init__(self):")
self.indent()
self.emit('"""Initialize the multi-agent system."""')
self.emit("self.gpu_detector = GPUDetector()")
self.emit("self.gpu_detector.detect()")
self.emit("self.model_manager = ModelManager(self.gpu_detector)")
self.emit("self.mcp_server = MCPServer()")
self.emit("self.orchestrator = Orchestrator(self.mcp_server)")
self.emit_blank_line()
self.emit("# Create and register agents")
for agent in system.agents:
if isinstance(agent, AgentDeclaration):
agent_name = agent.name
self.emit(f"self.{agent_name.lower()} = {agent_name}Agent(self.model_manager)")
self.emit(f"self.orchestrator.register_agent('{agent_name}', self.{agent_name.lower()}.agent)")
self.emit_blank_line()
if system.orchestration and 'entry_point' in system.orchestration:
entry_point = system.orchestration['entry_point']
self.emit(f"self.orchestrator.set_entry_point('{entry_point}')")
self.dedent()
self.emit_blank_line()
self.emit("async def run(self, user_input: str) -> Dict[str, Any]:")
self.indent()
self.emit('"""Execute the multi-agent workflow."""')
self.emit("mcp_task = asyncio.create_task(self.mcp_server.start())")
self.emit_blank_line()
self.emit("try:")
self.indent()
self.emit("results = await self.orchestrator.execute_workflow(user_input)")
self.emit("return results")
self.dedent()
self.emit("finally:")
self.indent()
self.emit("await self.mcp_server.stop()")
self.emit("await mcp_task")
self.dedent()
self.dedent()
self.dedent()
self.emit_blank_line()
def emit_main_function(self):
self.emit("async def main():")
self.indent()
self.emit('"""Main execution function."""')
main_component = None
for declaration in self.ast:
if isinstance(declaration, (ChatbotDeclaration, MultiAgentSystem)):
main_component = declaration
break
if main_component:
self.emit(f"app = {main_component.name}()")
if isinstance(main_component, ChatbotDeclaration):
self.emit("await app.run()")
elif isinstance(main_component, MultiAgentSystem):
self.emit("print('Multi-Agent System Ready')")
self.emit("user_input = input('Enter your query: ')")
self.emit("results = await app.run(user_input)")
self.emit("print('\\nResults:')")
self.emit("for agent_id, output in results.items():")
self.indent()
self.emit("print(f'{agent_id}: {output}')")
self.dedent()
self.dedent()
self.emit_blank_line()
self.emit("if __name__ == '__main__':")
self.indent()
self.emit("asyncio.run(main())")
self.dedent()
def emit(self, line: str):
indent = " " * self.indent_level
self.code_lines.append(indent + line)
def emit_blank_line(self):
self.code_lines.append("")
def indent(self):
self.indent_level += 1
def dedent(self):
self.indent_level = max(0, self.indent_level - 1)
def is_remote_model(self, model_name: str) -> bool:
remote_prefixes = ['gpt-', 'claude-', 'gemini-']
return any(model_name.startswith(prefix) for prefix in remote_prefixes)
def get_provider(self, model_name: str) -> str:
if model_name.startswith('gpt-'):
return 'openai'
elif model_name.startswith('claude-'):
return 'anthropic'
elif model_name.startswith('gemini-'):
return 'google'
return 'unknown'
# ============================================================================
# COMPILER MAIN
# ============================================================================
class LLMDSLCompiler:
def __init__(self):
pass
def compile(self, source_file: str, output_file: str):
"""Compile a LLMDSL source file to Python."""
print(f"Compiling {source_file}...")
with open(source_file, 'r') as f:
source = f.read()
lexer = Lexer(source)
tokens = lexer.tokenize()
print(f"Lexical analysis complete: {len(tokens)} tokens")
parser = Parser(tokens)
ast = parser.parse()
print(f"Parsing complete: {len(ast)} declarations")
analyzer = SemanticAnalyzer(ast)
if not analyzer.analyze():
print("Compilation failed due to semantic errors")
return False
print("Semantic analysis complete")
generator = CodeGenerator(ast)
code = generator.generate()
print(f"Code generation complete: {len(code.splitlines())} lines")
with open(output_file, 'w') as f:
f.write(code)
print(f"Output written to {output_file}")
return True
def main():
if len(sys.argv) < 3:
print("Usage: python llmdsl_compiler.py <source_file> <output_file>")
sys.exit(1)
compiler = LLMDSLCompiler()
success = compiler.compile(sys.argv[1], sys.argv[2])
if success:
print("Compilation successful!")
sys.exit(0)
else:
print("Compilation failed!")
sys.exit(1)
if __name__ == '__main__':
main()
This complete compiler implementation includes the lexer, parser, semantic analyzer, and code generator. It can process LLMDSL source files and generate complete Python applications.
The runtime library implementation follows:
# llmdsl_runtime.py
# Complete runtime library for LLMDSL
import subprocess
import platform
import os
import asyncio
import json
import uuid
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass
from enum import Enum
from abc import ABC, abstractmethod
# ============================================================================
# GPU DETECTION
# ============================================================================
class GPUDetector:
"""Detects available GPU backends and selects the optimal one."""
def __init__(self):
self.detected_gpus = []
self.recommended_backend = None
def detect(self) -> str:
"""Detect all available GPU backends."""
system = platform.system()
if self.check_nvidia_cuda():
self.detected_gpus.append({
'vendor': 'NVIDIA',
'backend': 'cuda',
'priority': 1
})
if self.check_amd_rocm():
self.detected_gpus.append({
'vendor': 'AMD',
'backend': 'rocm',
'priority': 2
})
if self.check_intel_oneapi():
self.detected_gpus.append({
'vendor': 'Intel',
'backend': 'oneapi',
'priority': 3
})
if system == 'Darwin' and self.check_apple_metal():
self.detected_gpus.append({
'vendor': 'Apple',
'backend': 'metal',
'priority': 1
})
if self.detected_gpus:
self.detected_gpus.sort(key=lambda x: x['priority'])
self.recommended_backend = self.detected_gpus[0]['backend']
print(f"Detected GPU: {self.detected_gpus[0]['vendor']} ({self.recommended_backend})")
else:
self.recommended_backend = 'cpu'
print("No GPU detected, using CPU")
return self.recommended_backend
def check_nvidia_cuda(self) -> bool:
"""Check for NVIDIA CUDA support."""
try:
result = subprocess.run(['nvidia-smi'],
capture_output=True,
text=True,
timeout=5)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def check_amd_rocm(self) -> bool:
"""Check for AMD ROCm support."""
try:
result = subprocess.run(['rocm-smi'],
capture_output=True,
text=True,
timeout=5)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
def check_intel_oneapi(self) -> bool:
"""Check for Intel oneAPI support."""
return os.path.exists('/opt/intel/oneapi') or 'ONEAPI_ROOT' in os.environ
def check_apple_metal(self) -> bool:
"""Check for Apple Metal support."""
return platform.processor() == 'arm'
def get_device_config(self) -> Dict[str, Any]:
"""Get configuration for the selected backend."""
if self.recommended_backend == 'cuda':
return {'device': 'cuda', 'dtype': 'float16'}
elif self.recommended_backend == 'rocm':
return {'device': 'cuda', 'dtype': 'float16'}
elif self.recommended_backend == 'metal':
return {'device': 'mps', 'dtype': 'float16'}
elif self.recommended_backend == 'oneapi':
return {'device': 'xpu', 'dtype': 'float16'}
else:
return {'device': 'cpu', 'dtype': 'float32'}
# ============================================================================
# MODEL MANAGEMENT
# ============================================================================
class ModelManager:
"""Manages loading and execution of language models."""
def __init__(self, gpu_detector: GPUDetector):
self.gpu_detector = gpu_detector
self.loaded_models = {}
self.device_config = gpu_detector.get_device_config()
def load_local_model(self, model_name: str) -> Dict[str, Any]:
"""Load a local model."""
if model_name in self.loaded_models:
return self.loaded_models[model_name]
print(f"Loading model {model_name}...")
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16 if self.device_config['dtype'] == 'float16' else torch.float32,
device_map='auto'
)
self.loaded_models[model_name] = {
'model': model,
'tokenizer': tokenizer,
'device': self.device_config['device']
}
print(f"Model {model_name} loaded successfully")
return self.loaded_models[model_name]
except Exception as e:
print(f"Error loading model {model_name}: {str(e)}")
raise
def create_remote_client(self, provider: str, model_name: str, api_key: Optional[str] = None):
"""Create a client for remote model access."""
if provider == 'openai':
from openai import OpenAI
client = OpenAI(api_key=api_key or os.environ.get('OPENAI_API_KEY'))
return RemoteModelClient(client, model_name, 'openai')
elif provider == 'anthropic':
from anthropic import Anthropic
client = Anthropic(api_key=api_key or os.environ.get('ANTHROPIC_API_KEY'))
return RemoteModelClient(client, model_name, 'anthropic')
else:
raise ValueError(f"Unsupported provider: {provider}")
class RemoteModelClient:
"""Client for remote model APIs."""
def __init__(self, client, model_name: str, provider: str):
self.client = client
self.model_name = model_name
self.provider = provider
def generate(self, messages: list, **kwargs) -> str:
"""Generate a response using the remote API."""
if self.provider == 'openai':
response = self.client.chat.completions.create(
model=self.model_name,
messages=messages,
**kwargs
)
return response.choices[0].message.content
elif self.provider == 'anthropic':
response = self.client.messages.create(
model=self.model_name,
messages=messages,
**kwargs
)
return response.content[0].text
else:
raise ValueError(f"Unsupported provider: {self.provider}")
# ============================================================================
# RAG PIPELINE
# ============================================================================
class RAGPipeline:
"""Retrieval-Augmented Generation pipeline."""
def __init__(self, vector_store_type: str = 'chroma', embedding_model: str = 'sentence-transformers/all-MiniLM-L6-v2'):
self.vector_store_type = vector_store_type
self.embedding_model_name = embedding_model
self.embeddings = None
self.vector_store = None
def initialize(self, persist_directory: str = './chroma_db'):
"""Initialize the RAG pipeline."""
print(f"Initializing RAG with {self.embedding_model_name}...")
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
self.embeddings = HuggingFaceEmbeddings(model_name=self.embedding_model_name)
self.vector_store = Chroma(persist_directory=persist_directory, embedding_function=self.embeddings)
print("RAG pipeline initialized")
def load_documents(self, documents_path: str):
"""Load documents into the vector store."""
print(f"Loading documents from {documents_path}...")
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
loader = DirectoryLoader(documents_path, glob="**/*.txt", loader_cls=TextLoader)
documents = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)
self.vector_store.add_documents(chunks)
print(f"Loaded {len(chunks)} document chunks")
def retrieve(self, query: str, k: int = 4) -> List[Dict[str, Any]]:
"""Retrieve relevant documents."""
results = self.vector_store.similarity_search_with_score(query, k=k)
return [{'content': doc.page_content, 'score': score} for doc, score in results]
def augment_prompt(self, query: str, k: int = 4) -> str:
"""Augment a prompt with retrieved context."""
docs = self.retrieve(query, k)
context = "\n\n".join([f"Document {i+1}:\n{doc['content']}" for i, doc in enumerate(docs)])
return f"Context:\n{context}\n\nQuestion: {query}\n\nAnswer:"
class GraphRAGPipeline:
"""Graph-based RAG pipeline."""
def __init__(self, embedding_model: str = 'sentence-transformers/all-MiniLM-L6-v2'):
import networkx as nx
from langchain_community.embeddings import HuggingFaceEmbeddings
self.graph = nx.DiGraph()
self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model)
self.entity_embeddings = {}
def add_entity(self, entity_id: str, entity_type: str, properties: Dict[str, Any]):
"""Add an entity to the knowledge graph."""
self.graph.add_node(entity_id, entity_type=entity_type, properties=properties)
entity_text = f"{entity_type}: {entity_id} " + " ".join(f"{k}={v}" for k, v in properties.items())
self.entity_embeddings[entity_id] = self.embeddings.embed_query(entity_text)
def add_relationship(self, source: str, target: str, relationship_type: str, properties: Dict[str, Any] = None):
"""Add a relationship between entities."""
self.graph.add_edge(source, target, relationship_type=relationship_type, properties=properties or {})
# ============================================================================
# AGENT IMPLEMENTATION
# ============================================================================
class Tool:
"""Represents a tool that agents can use."""
def __init__(self, name: str, description: str, function: Callable):
self.name = name
self.description = description
self.function = function
async def execute(self, **kwargs) -> Any:
"""Execute the tool."""
if asyncio.iscoroutinefunction(self.function):
return await self.function(**kwargs)
else:
return self.function(**kwargs)
class Agent(ABC):
"""Base agent class."""
def __init__(self, name: str, model_manager: ModelManager, model_name: str, system_prompt: str):
self.name = name
self.model_manager = model_manager
self.model_name = model_name
self.system_prompt = system_prompt
self.tools = []
self.rag_pipeline = None
self.conversation_history = []
def add_tool(self, tool: Tool):
"""Add a tool to the agent."""
self.tools.append(tool)
async def think(self, user_input: str) -> str:
"""Process input and generate response."""
if self.rag_pipeline:
user_input = self.rag_pipeline.augment_prompt(user_input)
self.conversation_history.append({'role': 'user', 'content': user_input})
messages = [{'role': 'system', 'content': self.system_prompt}]
messages.extend(self.conversation_history)
response = await self.generate_response(messages)
self.conversation_history.append({'role': 'assistant', 'content': response})
return response
@abstractmethod
async def generate_response(self, messages: List[Dict[str, str]]) -> str:
"""Generate a response."""
pass
async def handle_mcp_request(self, message) -> Any:
"""Handle an MCP request."""
return await self.think(str(message.content))
async def handle_mcp_notification(self, message):
"""Handle an MCP notification."""
pass
class LocalAgent(Agent):
"""Agent using a local model."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.model_info = self.model_manager.load_local_model(self.model_name)
async def generate_response(self, messages: List[Dict[str, str]]) -> str:
"""Generate response using local model."""
import torch
model = self.model_info['model']
tokenizer = self.model_info['tokenizer']
device = self.model_info['device']
prompt = self.format_messages(messages)
inputs = tokenizer(prompt, return_tensors='pt').to(device)
with torch.no_grad():
outputs = model.generate(**inputs, max_new_tokens=512, temperature=0.7, do_sample=True)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
response = response[len(prompt):].strip()
return response
def format_messages(self, messages: List[Dict[str, str]]) -> str:
"""Format messages into a prompt."""
formatted = []
for msg in messages:
role = msg['role']
content = msg['content']
if role == 'system':
formatted.append(f"System: {content}")
elif role == 'user':
formatted.append(f"User: {content}")
elif role == 'assistant':
formatted.append(f"Assistant: {content}")
formatted.append("Assistant:")
return "\n\n".join(formatted)
class RemoteAgent(Agent):
"""Agent using a remote model API."""
def __init__(self, *args, provider: str = 'openai', **kwargs):
super().__init__(*args, **kwargs)
self.provider = provider
self.client = self.model_manager.create_remote_client(provider, self.model_name)
async def generate_response(self, messages: List[Dict[str, str]]) -> str:
"""Generate response using remote API."""
return self.client.generate(messages, temperature=0.7, max_tokens=512)
# ============================================================================
# MCP IMPLEMENTATION
# ============================================================================
class MessageType(Enum):
REQUEST = "request"
RESPONSE = "response"
NOTIFICATION = "notification"
ERROR = "error"
@dataclass
class MCPMessage:
message_type: MessageType
sender: str
recipient: str
content: Any
message_id: str
correlation_id: Optional[str] = None
class MCPServer:
"""Model Context Protocol server."""
def __init__(self):
self.agents = {}
self.message_queue = asyncio.Queue()
self.running = False
def register_agent(self, agent_id: str, agent: Agent):
"""Register an agent."""
self.agents[agent_id] = agent
async def send_message(self, message: MCPMessage):
"""Send a message."""
await self.message_queue.put(message)
async def start(self):
"""Start the MCP server."""
self.running = True
while self.running:
try:
message = await asyncio.wait_for(self.message_queue.get(), timeout=1.0)
await self.process_message(message)
except asyncio.TimeoutError:
continue
async def process_message(self, message: MCPMessage):
"""Process a message."""
if message.recipient in self.agents:
recipient = self.agents[message.recipient]
if message.message_type == MessageType.REQUEST:
response = await recipient.handle_mcp_request(message)
response_msg = MCPMessage(
message_type=MessageType.RESPONSE,
sender=message.recipient,
recipient=message.sender,
content=response,
message_id=str(uuid.uuid4()),
correlation_id=message.message_id
)
await self.send_message(response_msg)
async def stop(self):
"""Stop the MCP server."""
self.running = False
def generate_message_id(self) -> str:
"""Generate a unique message ID."""
return str(uuid.uuid4())
class MCPClient:
"""MCP client for agents."""
def __init__(self, agent_id: str, server: MCPServer):
self.agent_id = agent_id
self.server = server
# ============================================================================
# ORCHESTRATION
# ============================================================================
class OrchestrationRule:
"""Represents a routing rule in multi-agent orchestration."""
def __init__(self, source: str, target: str, condition: Callable[[Any], bool]):
self.source = source
self.target = target
self.condition = condition
class Orchestrator:
"""Orchestrates multi-agent workflows."""
def __init__(self, mcp_server: MCPServer):
self.mcp_server = mcp_server
self.agents = {}
self.routing_rules = []
self.entry_point = None
def register_agent(self, agent_id: str, agent: Agent):
"""Register an agent."""
self.agents[agent_id] = agent
self.mcp_server.register_agent(agent_id, agent)
def set_entry_point(self, agent_id: str):
"""Set the entry point agent."""
self.entry_point = agent_id
def add_routing_rule(self, rule: OrchestrationRule):
"""Add a routing rule."""
self.routing_rules.append(rule)
async def execute_workflow(self, initial_input: str) -> Dict[str, Any]:
"""Execute the workflow."""
if not self.entry_point:
raise RuntimeError("No entry point defined")
current_agent_id = self.entry_point
current_input = initial_input
outputs = {}
while current_agent_id:
agent = self.agents[current_agent_id]
output = await agent.think(current_input)
outputs[current_agent_id] = output
next_agent_id = None
for rule in self.routing_rules:
if rule.source == current_agent_id and rule.condition(output):
next_agent_id = rule.target
break
current_agent_id = next_agent_id
current_input = output
return outputs
# ============================================================================
# USER INTERFACES
# ============================================================================
class ConsoleUI:
"""Console-based user interface."""
def __init__(self, agent: Agent):
self.agent = agent
async def run(self):
"""Run the console interface."""
print("Console Interface Started")
print("Type 'quit' or 'exit' to end")
print("-" * 50)
while True:
try:
user_input = input("\nYou: ")
if user_input.lower() in ['quit', 'exit']:
print("Goodbye!")
break
if not user_input.strip():
continue
response = await self.agent.think(user_input)
print(f"\nAssistant: {response}")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"\nError: {str(e)}")
class WebUI:
"""Web-based user interface."""
def __init__(self, agent: Agent, port: int = 8080):
self.agent = agent
self.port = port
async def run(self):
"""Run the web interface."""
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import uvicorn
app = FastAPI()
html = """<!DOCTYPE html>
<html>
<head><title>LLM Chat</title>
<style>
body{font-family:Arial;max-width:800px;margin:0 auto;padding:20px}
#chat{border:1px solid #ccc;height:500px;overflow-y:auto;padding:10px;margin-bottom:10px}
.message{margin:10px 0;padding:10px;border-radius:5px}
.user{background:#e3f2fd;text-align:right}
.assistant{background:#f5f5f5}
#input-container{display:flex}
#input{flex:1;padding:10px;font-size:16px}
#send{padding:10px 20px;font-size:16px;background:#2196F3;color:white;border:none;cursor:pointer}
</style>
</head>
<body>
<h1>LLM Chat Interface</h1>
<div id="chat"></div>
<div id="input-container">
<input type="text" id="input" placeholder="Type your message...">
<button id="send">Send</button>
</div>
<script>
const ws=new WebSocket('ws://localhost:8080/ws');
const chat=document.getElementById('chat');
const input=document.getElementById('input');
const send=document.getElementById('send');
function addMsg(content,isUser){
const div=document.createElement('div');
div.className='message '+(isUser?'user':'assistant');
div.textContent=content;
chat.appendChild(div);
chat.scrollTop=chat.scrollHeight;
}
ws.onmessage=function(e){addMsg(e.data,false);};
function sendMsg(){
const msg=input.value.trim();
if(msg){addMsg(msg,true);ws.send(msg);input.value='';}
}
send.addEventListener('click',sendMsg);
input.addEventListener('keypress',function(e){if(e.key==='Enter')sendMsg();});
</script>
</body>
</html>"""
@app.get("/")
async def get():
return HTMLResponse(content=html)
@app.websocket("/ws")
async def ws(websocket: WebSocket):
await websocket.accept()
try:
while True:
msg = await websocket.receive_text()
response = await self.agent.think(msg)
await websocket.send_text(response)
except:
pass
config = uvicorn.Config(app, host="0.0.0.0", port=self.port, log_level="error")
server = uvicorn.Server(config)
await server.serve()
This complete runtime library provides all the infrastructure needed for generated applications to run. It includes GPU detection, model management, RAG pipelines, agent implementations, MCP communication, orchestration, and user interfaces.
To use the complete system, create a DSL file named example.llmdsl:
chatbot MyAssistant {
model: "gpt-3.5-turbo"
system_prompt: "You are a helpful AI assistant."
temperature: 0.7
}
settings {
gpu: "auto"
ui: "console"
}
Then compile and run it:
python llmdsl_compiler.py example.llmdsl generated_app.py
python generated_app.py
The system will generate a complete, production-ready Python application that detects your GPU, loads the specified model, and provides a console interface for chatting with the LLM. The generated code is clean, well-structured, and can be extended with custom functionality as needed.
ADVANCED FEATURES AND EXTENSIONS
Now that we have established the core functionality of our LLMDSL system, we can explore advanced features that make it truly production-ready. These include sophisticated error handling, streaming responses, token budget management, custom tool creation, advanced orchestration patterns, and performance optimization strategies.
COMPREHENSIVE ERROR HANDLING AND RETRY LOGIC
Production LLM applications must handle various failure modes gracefully. Network timeouts, rate limits, model errors, and resource exhaustion all require specific handling strategies. Our runtime library implements a robust error handling system with configurable retry policies.
# llmdsl_runtime_advanced.py
# Advanced features for the LLMDSL runtime
import time
import random
from typing import Optional, Callable, Any
from dataclasses import dataclass
from enum import Enum
class BackoffStrategy(Enum):
CONSTANT = "constant"
LINEAR = "linear"
EXPONENTIAL = "exponential"
JITTERED_EXPONENTIAL = "jittered_exponential"
@dataclass
class RetryPolicy:
max_retries: int = 3
backoff_strategy: BackoffStrategy = BackoffStrategy.EXPONENTIAL
initial_delay: float = 1.0
max_delay: float = 60.0
retry_on_exceptions: tuple = (Exception,)
class RetryHandler:
"""Handles retry logic with configurable backoff strategies."""
def __init__(self, policy: RetryPolicy):
self.policy = policy
def calculate_delay(self, attempt: int) -> float:
"""Calculate delay for the given attempt number."""
if self.policy.backoff_strategy == BackoffStrategy.CONSTANT:
delay = self.policy.initial_delay
elif self.policy.backoff_strategy == BackoffStrategy.LINEAR:
delay = self.policy.initial_delay * attempt
elif self.policy.backoff_strategy == BackoffStrategy.EXPONENTIAL:
delay = self.policy.initial_delay * (2 ** (attempt - 1))
elif self.policy.backoff_strategy == BackoffStrategy.JITTERED_EXPONENTIAL:
base_delay = self.policy.initial_delay * (2 ** (attempt - 1))
jitter = random.uniform(0, base_delay * 0.1)
delay = base_delay + jitter
else:
delay = self.policy.initial_delay
return min(delay, self.policy.max_delay)
async def execute_with_retry(self, func: Callable, *args, **kwargs) -> Any:
"""Execute a function with retry logic."""
last_exception = None
for attempt in range(1, self.policy.max_retries + 1):
try:
if asyncio.iscoroutinefunction(func):
return await func(*args, **kwargs)
else:
return func(*args, **kwargs)
except self.policy.retry_on_exceptions as e:
last_exception = e
if attempt < self.policy.max_retries:
delay = self.calculate_delay(attempt)
print(f"Attempt {attempt} failed: {str(e)}. Retrying in {delay:.2f}s...")
await asyncio.sleep(delay)
else:
print(f"All {self.policy.max_retries} attempts failed")
raise last_exception
class ErrorRecoveryManager:
"""Manages error recovery strategies for different error types."""
def __init__(self):
self.error_handlers = {}
self.fallback_responses = {}
def register_error_handler(self, error_type: type, handler: Callable):
"""Register a handler for a specific error type."""
self.error_handlers[error_type] = handler
def register_fallback_response(self, context: str, response: str):
"""Register a fallback response for a specific context."""
self.fallback_responses[context] = response
async def handle_error(self, error: Exception, context: str = "default") -> Optional[str]:
"""Handle an error and return a recovery response if available."""
error_type = type(error)
if error_type in self.error_handlers:
try:
return await self.error_handlers[error_type](error, context)
except Exception as handler_error:
print(f"Error handler failed: {str(handler_error)}")
if context in self.fallback_responses:
return self.fallback_responses[context]
return None
The retry handler implements multiple backoff strategies to handle transient failures. Constant backoff uses the same delay between retries, which works well for predictable rate limits. Linear backoff increases delay proportionally, suitable for gradually increasing load. Exponential backoff doubles the delay each time, ideal for network issues. Jittered exponential adds randomness to prevent thundering herd problems when multiple clients retry simultaneously.
The error recovery manager allows applications to define custom handlers for specific error types and fallback responses when normal operation fails. This ensures graceful degradation rather than complete failure.
STREAMING RESPONSE IMPLEMENTATION
Many LLM applications benefit from streaming responses, where tokens appear incrementally rather than waiting for the complete response. This improves perceived latency and enables real-time user feedback.
class StreamingResponse:
"""Represents a streaming response from an LLM."""
def __init__(self):
self.chunks = []
self.complete = False
self.error = None
def add_chunk(self, chunk: str):
"""Add a chunk to the streaming response."""
self.chunks.append(chunk)
def get_full_text(self) -> str:
"""Get the complete response text."""
return ''.join(self.chunks)
def mark_complete(self):
"""Mark the response as complete."""
self.complete = True
def mark_error(self, error: Exception):
"""Mark the response as failed."""
self.error = error
self.complete = True
class StreamingAgent(Agent):
"""Agent that supports streaming responses."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.streaming_enabled = True
async def think_stream(self, user_input: str):
"""Process input and yield response chunks as they arrive."""
if self.rag_pipeline:
user_input = self.rag_pipeline.augment_prompt(user_input)
self.conversation_history.append({'role': 'user', 'content': user_input})
messages = [{'role': 'system', 'content': self.system_prompt}]
messages.extend(self.conversation_history)
full_response = ""
async for chunk in self.generate_response_stream(messages):
full_response += chunk
yield chunk
self.conversation_history.append({'role': 'assistant', 'content': full_response})
async def generate_response_stream(self, messages: List[Dict[str, str]]):
"""Generate a streaming response. Override in subclasses."""
response = await self.generate_response(messages)
for char in response:
yield char
await asyncio.sleep(0.01)
class StreamingRemoteAgent(RemoteAgent):
"""Remote agent with streaming support."""
async def generate_response_stream(self, messages: List[Dict[str, str]]):
"""Generate streaming response from remote API."""
if self.provider == 'openai':
stream = self.client.chat.completions.create(
model=self.model_name,
messages=messages,
temperature=0.7,
max_tokens=512,
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
else:
response = await self.generate_response(messages)
for char in response:
yield char
await asyncio.sleep(0.01)
Streaming responses require careful handling of partial content. The streaming agent yields chunks as they arrive, allowing the UI to display them immediately. This creates a more responsive user experience, especially for long responses.
TOKEN BUDGET MANAGEMENT
Production applications must manage token usage to control costs and stay within API limits. The token budget manager tracks usage across conversations and enforces limits.
class TokenBudget:
"""Manages token budgets for LLM operations."""
def __init__(self, max_tokens_per_request: int = 4096, max_tokens_per_session: int = 100000):
self.max_tokens_per_request = max_tokens_per_request
self.max_tokens_per_session = max_tokens_per_session
self.session_tokens_used = 0
self.request_history = []
def estimate_tokens(self, text: str) -> int:
"""Estimate token count for text. Simple approximation."""
return len(text.split()) * 1.3
def can_afford_request(self, prompt_tokens: int, max_completion_tokens: int) -> bool:
"""Check if a request fits within budget."""
total_tokens = prompt_tokens + max_completion_tokens
if total_tokens > self.max_tokens_per_request:
return False
if self.session_tokens_used + total_tokens > self.max_tokens_per_session:
return False
return True
def record_usage(self, prompt_tokens: int, completion_tokens: int):
"""Record token usage for a request."""
total = prompt_tokens + completion_tokens
self.session_tokens_used += total
self.request_history.append({
'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens,
'total_tokens': total,
'timestamp': time.time()
})
def get_remaining_budget(self) -> int:
"""Get remaining tokens in session budget."""
return self.max_tokens_per_session - self.session_tokens_used
def reset_session(self):
"""Reset session token counter."""
self.session_tokens_used = 0
self.request_history = []
class BudgetAwareAgent(Agent):
"""Agent that respects token budgets."""
def __init__(self, *args, token_budget: Optional[TokenBudget] = None, **kwargs):
super().__init__(*args, **kwargs)
self.token_budget = token_budget or TokenBudget()
async def think(self, user_input: str) -> str:
"""Process input with budget awareness."""
prompt_tokens = self.token_budget.estimate_tokens(user_input)
max_completion = 512
if not self.token_budget.can_afford_request(prompt_tokens, max_completion):
remaining = self.token_budget.get_remaining_budget()
if remaining < max_completion:
return f"Token budget exhausted. Remaining: {remaining} tokens."
else:
self.trim_conversation_history()
response = await super().think(user_input)
completion_tokens = self.token_budget.estimate_tokens(response)
self.token_budget.record_usage(prompt_tokens, completion_tokens)
return response
def trim_conversation_history(self):
"""Trim conversation history to fit budget."""
while len(self.conversation_history) > 2:
self.conversation_history.pop(0)
Token budget management prevents runaway costs and ensures applications stay within operational limits. The budget-aware agent automatically trims conversation history when approaching limits, maintaining recent context while discarding older messages.
ADVANCED TOOL SYSTEM
Tools extend agent capabilities by providing access to external systems and computations. The advanced tool system supports parameter validation, async execution, and result caching.
from typing import Dict, Any, List, Optional
import hashlib
import json
class ToolParameter:
"""Represents a tool parameter with validation."""
def __init__(self, name: str, param_type: type, required: bool = True,
default: Any = None, description: str = ""):
self.name = name
self.param_type = param_type
self.required = required
self.default = default
self.description = description
def validate(self, value: Any) -> bool:
"""Validate a parameter value."""
if value is None:
return not self.required
return isinstance(value, self.param_type)
class AdvancedTool:
"""Enhanced tool with parameter validation and caching."""
def __init__(self, name: str, description: str, function: Callable,
parameters: List[ToolParameter], cache_results: bool = False):
self.name = name
self.description = description
self.function = function
self.parameters = {p.name: p for p in parameters}
self.cache_results = cache_results
self.cache = {}
def get_schema(self) -> Dict[str, Any]:
"""Get tool schema for LLM function calling."""
return {
'name': self.name,
'description': self.description,
'parameters': {
'type': 'object',
'properties': {
p.name: {
'type': p.param_type.__name__,
'description': p.description
}
for p in self.parameters.values()
},
'required': [p.name for p in self.parameters.values() if p.required]
}
}
def validate_parameters(self, **kwargs) -> tuple[bool, Optional[str]]:
"""Validate provided parameters."""
for param_name, param_def in self.parameters.items():
if param_name not in kwargs:
if param_def.required:
return False, f"Missing required parameter: {param_name}"
kwargs[param_name] = param_def.default
elif not param_def.validate(kwargs[param_name]):
return False, f"Invalid type for parameter {param_name}"
return True, None
def get_cache_key(self, **kwargs) -> str:
"""Generate cache key for parameters."""
sorted_params = json.dumps(kwargs, sort_keys=True)
return hashlib.md5(sorted_params.encode()).hexdigest()
async def execute(self, **kwargs) -> Any:
"""Execute the tool with validation and caching."""
valid, error = self.validate_parameters(**kwargs)
if not valid:
raise ValueError(error)
if self.cache_results:
cache_key = self.get_cache_key(**kwargs)
if cache_key in self.cache:
return self.cache[cache_key]
if asyncio.iscoroutinefunction(self.function):
result = await self.function(**kwargs)
else:
result = self.function(**kwargs)
if self.cache_results:
self.cache[cache_key] = result
return result
class ToolRegistry:
"""Central registry for tools."""
def __init__(self):
self.tools = {}
def register(self, tool: AdvancedTool):
"""Register a tool."""
self.tools[tool.name] = tool
def get_tool(self, name: str) -> Optional[AdvancedTool]:
"""Get a tool by name."""
return self.tools.get(name)
def get_all_schemas(self) -> List[Dict[str, Any]]:
"""Get schemas for all tools."""
return [tool.get_schema() for tool in self.tools.values()]
def create_tool_from_function(self, name: str, description: str,
parameters: List[ToolParameter],
cache_results: bool = False):
"""Decorator to create a tool from a function."""
def decorator(func):
tool = AdvancedTool(name, description, func, parameters, cache_results)
self.register(tool)
return func
return decorator
The advanced tool system provides comprehensive parameter validation, automatic caching for expensive operations, and schema generation for LLM function calling. Tools can be registered centrally and shared across agents.
Here are example tool implementations:
tool_registry = ToolRegistry()
@tool_registry.create_tool_from_function(
name="web_search",
description="Search the web for information",
parameters=[
ToolParameter("query", str, required=True, description="Search query"),
ToolParameter("num_results", int, required=False, default=5, description="Number of results")
],
cache_results=True
)
async def web_search(query: str, num_results: int = 5) -> List[Dict[str, str]]:
"""Perform web search. This is a placeholder implementation."""
import aiohttp
async with aiohttp.ClientSession() as session:
params = {'q': query, 'num': num_results}
async with session.get('https://api.search.example.com/search', params=params) as response:
if response.status == 200:
data = await response.json()
return data.get('results', [])
return []
@tool_registry.create_tool_from_function(
name="calculator",
description="Perform mathematical calculations",
parameters=[
ToolParameter("expression", str, required=True, description="Mathematical expression to evaluate")
]
)
def calculator(expression: str) -> float:
"""Safely evaluate mathematical expressions."""
import ast
import operator
operators = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg
}
def eval_expr(node):
if isinstance(node, ast.Num):
return node.n
elif isinstance(node, ast.BinOp):
return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))
elif isinstance(node, ast.UnaryOp):
return operators[type(node.op)](eval_expr(node.operand))
else:
raise ValueError(f"Unsupported operation: {type(node)}")
try:
tree = ast.parse(expression, mode='eval')
return eval_expr(tree.body)
except Exception as e:
raise ValueError(f"Invalid expression: {str(e)}")
@tool_registry.create_tool_from_function(
name="file_reader",
description="Read contents of a file",
parameters=[
ToolParameter("filepath", str, required=True, description="Path to the file"),
ToolParameter("max_lines", int, required=False, default=100, description="Maximum lines to read")
]
)
async def file_reader(filepath: str, max_lines: int = 100) -> str:
"""Read file contents safely."""
import os
if not os.path.exists(filepath):
raise FileNotFoundError(f"File not found: {filepath}")
if not os.path.isfile(filepath):
raise ValueError(f"Not a file: {filepath}")
lines = []
with open(filepath, 'r', encoding='utf-8') as f:
for i, line in enumerate(f):
if i >= max_lines:
break
lines.append(line.rstrip())
return '\n'.join(lines)
These tool implementations demonstrate parameter validation, async execution, and safe operation. The web search tool caches results to avoid redundant API calls. The calculator safely evaluates mathematical expressions without using eval. The file reader enforces limits to prevent reading enormous files.
ADVANCED ORCHESTRATION PATTERNS
Multi-agent systems benefit from sophisticated orchestration patterns beyond simple sequential routing. We can implement parallel execution, conditional branching, loops, and hierarchical delegation.
class OrchestrationPattern(Enum):
SEQUENTIAL = "sequential"
PARALLEL = "parallel"
CONDITIONAL = "conditional"
LOOP = "loop"
HIERARCHICAL = "hierarchical"
class WorkflowNode:
"""Represents a node in the workflow graph."""
def __init__(self, node_id: str, agent_id: str, pattern: OrchestrationPattern = OrchestrationPattern.SEQUENTIAL):
self.node_id = node_id
self.agent_id = agent_id
self.pattern = pattern
self.next_nodes = []
self.condition = None
self.max_iterations = None
def add_next(self, node: 'WorkflowNode', condition: Optional[Callable] = None):
"""Add a next node with optional condition."""
self.next_nodes.append((node, condition))
class AdvancedOrchestrator(Orchestrator):
"""Enhanced orchestrator with advanced patterns."""
def __init__(self, mcp_server: MCPServer):
super().__init__(mcp_server)
self.workflow_graph = {}
self.execution_history = []
def add_workflow_node(self, node: WorkflowNode):
"""Add a node to the workflow graph."""
self.workflow_graph[node.node_id] = node
async def execute_node(self, node: WorkflowNode, input_data: Any) -> Any:
"""Execute a single workflow node."""
agent = self.agents.get(node.agent_id)
if not agent:
raise ValueError(f"Agent not found: {node.agent_id}")
self.execution_history.append({
'node_id': node.node_id,
'agent_id': node.agent_id,
'timestamp': time.time()
})
if node.pattern == OrchestrationPattern.PARALLEL:
return await self.execute_parallel_node(node, input_data)
elif node.pattern == OrchestrationPattern.LOOP:
return await self.execute_loop_node(node, input_data)
else:
return await agent.think(str(input_data))
async def execute_parallel_node(self, node: WorkflowNode, input_data: Any) -> Dict[str, Any]:
"""Execute multiple agents in parallel."""
tasks = []
agent_ids = []
for next_node, _ in node.next_nodes:
agent = self.agents.get(next_node.agent_id)
if agent:
tasks.append(agent.think(str(input_data)))
agent_ids.append(next_node.agent_id)
results = await asyncio.gather(*tasks)
return {agent_id: result for agent_id, result in zip(agent_ids, results)}
async def execute_loop_node(self, node: WorkflowNode, input_data: Any) -> List[Any]:
"""Execute a node in a loop until condition is met."""
results = []
current_input = input_data
iterations = 0
max_iter = node.max_iterations or 10
while iterations < max_iter:
agent = self.agents.get(node.agent_id)
result = await agent.think(str(current_input))
results.append(result)
if node.condition and not node.condition(result):
break
current_input = result
iterations += 1
return results
async def execute_advanced_workflow(self, start_node_id: str, initial_input: Any) -> Dict[str, Any]:
"""Execute an advanced workflow with complex patterns."""
if start_node_id not in self.workflow_graph:
raise ValueError(f"Start node not found: {start_node_id}")
current_node = self.workflow_graph[start_node_id]
current_input = initial_input
all_outputs = {}
visited = set()
while current_node:
if current_node.node_id in visited and current_node.pattern != OrchestrationPattern.LOOP:
break
visited.add(current_node.node_id)
output = await self.execute_node(current_node, current_input)
all_outputs[current_node.node_id] = output
next_node = None
for candidate_node, condition in current_node.next_nodes:
if condition is None or condition(output):
next_node = candidate_node
break
current_node = next_node
current_input = output
return all_outputs
The advanced orchestrator supports complex workflow patterns. Parallel execution runs multiple agents simultaneously and aggregates results. Loop execution repeats agent invocations until a condition is met. Conditional branching routes to different agents based on output content. This enables sophisticated multi-agent workflows that adapt to intermediate results.
PERFORMANCE OPTIMIZATION AND CACHING
Production systems require optimization for speed and resource efficiency. We implement multiple caching layers and optimization strategies.
import pickle
from collections import OrderedDict
class LRUCache:
"""Least Recently Used cache implementation."""
def __init__(self, capacity: int):
self.cache = OrderedDict()
self.capacity = capacity
def get(self, key: str) -> Optional[Any]:
"""Get a value from cache."""
if key not in self.cache:
return None
self.cache.move_to_end(key)
return self.cache[key]
def put(self, key: str, value: Any):
"""Put a value in cache."""
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = value
if len(self.cache) > self.capacity:
self.cache.popitem(last=False)
def clear(self):
"""Clear the cache."""
self.cache.clear()
class ResponseCache:
"""Caches LLM responses to avoid redundant API calls."""
def __init__(self, capacity: int = 1000, ttl: int = 3600):
self.cache = LRUCache(capacity)
self.ttl = ttl
self.timestamps = {}
def get_cache_key(self, messages: List[Dict[str, str]], model: str) -> str:
"""Generate cache key from messages and model."""
content = json.dumps({'messages': messages, 'model': model}, sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()
def get(self, messages: List[Dict[str, str]], model: str) -> Optional[str]:
"""Get cached response if available and not expired."""
key = self.get_cache_key(messages, model)
if key in self.timestamps:
age = time.time() - self.timestamps[key]
if age > self.ttl:
del self.timestamps[key]
return None
return self.cache.get(key)
def put(self, messages: List[Dict[str, str]], model: str, response: str):
"""Cache a response."""
key = self.get_cache_key(messages, model)
self.cache.put(key, response)
self.timestamps[key] = time.time()
class CachedAgent(Agent):
"""Agent with response caching."""
def __init__(self, *args, cache_capacity: int = 1000, cache_ttl: int = 3600, **kwargs):
super().__init__(*args, **kwargs)
self.response_cache = ResponseCache(capacity=cache_capacity, ttl=cache_ttl)
async def think(self, user_input: str) -> str:
"""Process input with caching."""
if self.rag_pipeline:
user_input = self.rag_pipeline.augment_prompt(user_input)
self.conversation_history.append({'role': 'user', 'content': user_input})
messages = [{'role': 'system', 'content': self.system_prompt}]
messages.extend(self.conversation_history)
cached_response = self.response_cache.get(messages, self.model_name)
if cached_response:
response = cached_response
else:
response = await self.generate_response(messages)
self.response_cache.put(messages, self.model_name, response)
self.conversation_history.append({'role': 'assistant', 'content': response})
return response
Response caching dramatically reduces costs and latency for repeated queries. The LRU cache evicts least recently used entries when capacity is reached. The TTL ensures cached responses do not become stale. This is particularly valuable for FAQ-style applications where users ask similar questions repeatedly.
BATCH PROCESSING AND QUEUE MANAGEMENT
High-throughput applications benefit from batch processing and queue management to handle many concurrent requests efficiently.
class RequestQueue:
"""Manages queued requests with priority support."""
def __init__(self, max_concurrent: int = 5):
self.queue = asyncio.PriorityQueue()
self.max_concurrent = max_concurrent
self.active_requests = 0
self.semaphore = asyncio.Semaphore(max_concurrent)
async def enqueue(self, request: Dict[str, Any], priority: int = 0):
"""Add a request to the queue."""
await self.queue.put((priority, time.time(), request))
async def process_queue(self, processor: Callable):
"""Process queued requests."""
while True:
try:
priority, timestamp, request = await asyncio.wait_for(
self.queue.get(),
timeout=1.0
)
async with self.semaphore:
await processor(request)
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"Error processing request: {str(e)}")
class BatchProcessor:
"""Processes requests in batches for efficiency."""
def __init__(self, batch_size: int = 10, max_wait: float = 1.0):
self.batch_size = batch_size
self.max_wait = max_wait
self.pending_requests = []
self.last_batch_time = time.time()
async def add_request(self, request: Dict[str, Any]) -> Any:
"""Add a request and get result when batch processes."""
future = asyncio.Future()
self.pending_requests.append((request, future))
if len(self.pending_requests) >= self.batch_size or \
time.time() - self.last_batch_time >= self.max_wait:
await self.process_batch()
return await future
async def process_batch(self):
"""Process accumulated requests as a batch."""
if not self.pending_requests:
return
batch = self.pending_requests
self.pending_requests = []
self.last_batch_time = time.time()
requests = [req for req, _ in batch]
futures = [fut for _, fut in batch]
try:
results = await self.execute_batch(requests)
for future, result in zip(futures, results):
future.set_result(result)
except Exception as e:
for future in futures:
future.set_exception(e)
async def execute_batch(self, requests: List[Dict[str, Any]]) -> List[Any]:
"""Execute a batch of requests. Override in subclasses."""
return [None] * len(requests)
Batch processing amortizes overhead across multiple requests. This is especially valuable when using remote APIs that support batch operations. The queue manager controls concurrency to prevent overwhelming downstream systems.
MONITORING AND OBSERVABILITY
Production systems require comprehensive monitoring to track performance, detect issues, and optimize resource usage.
class MetricsCollector:
"""Collects and aggregates metrics."""
def __init__(self):
self.metrics = {}
self.counters = {}
self.histograms = {}
def increment_counter(self, name: str, value: int = 1):
"""Increment a counter metric."""
if name not in self.counters:
self.counters[name] = 0
self.counters[name] += value
def record_value(self, name: str, value: float):
"""Record a value in a histogram."""
if name not in self.histograms:
self.histograms[name] = []
self.histograms[name].append(value)
def get_counter(self, name: str) -> int:
"""Get counter value."""
return self.counters.get(name, 0)
def get_histogram_stats(self, name: str) -> Dict[str, float]:
"""Get histogram statistics."""
if name not in self.histograms or not self.histograms[name]:
return {}
values = self.histograms[name]
return {
'count': len(values),
'min': min(values),
'max': max(values),
'mean': sum(values) / len(values),
'p50': self.percentile(values, 50),
'p95': self.percentile(values, 95),
'p99': self.percentile(values, 99)
}
def percentile(self, values: List[float], p: int) -> float:
"""Calculate percentile."""
sorted_values = sorted(values)
index = int(len(sorted_values) * p / 100)
return sorted_values[min(index, len(sorted_values) - 1)]
class MonitoredAgent(Agent):
"""Agent with built-in monitoring."""
def __init__(self, *args, metrics_collector: Optional[MetricsCollector] = None, **kwargs):
super().__init__(*args, **kwargs)
self.metrics = metrics_collector or MetricsCollector()
async def think(self, user_input: str) -> str:
"""Process input with monitoring."""
start_time = time.time()
try:
self.metrics.increment_counter(f'{self.name}.requests')
response = await super().think(user_input)
self.metrics.increment_counter(f'{self.name}.successes')
return response
except Exception as e:
self.metrics.increment_counter(f'{self.name}.errors')
raise
finally:
duration = time.time() - start_time
self.metrics.record_value(f'{self.name}.latency', duration)
def get_metrics_summary(self) -> Dict[str, Any]:
"""Get metrics summary for this agent."""
return {
'requests': self.metrics.get_counter(f'{self.name}.requests'),
'successes': self.metrics.get_counter(f'{self.name}.successes'),
'errors': self.metrics.get_counter(f'{self.name}.errors'),
'latency': self.metrics.get_histogram_stats(f'{self.name}.latency')
}
Metrics collection enables performance analysis and capacity planning. Tracking request counts, error rates, and latency distributions helps identify bottlenecks and optimize system behavior.
EXTENDING THE CODE GENERATOR FOR ADVANCED FEATURES
The code generator must be enhanced to support these advanced features when they are specified in the DSL.
class AdvancedCodeGenerator(CodeGenerator):
"""Enhanced code generator supporting advanced features."""
def __init__(self, ast: List[ASTNode]):
super().__init__(ast)
def emit_imports(self):
"""Emit enhanced imports including advanced features."""
imports = [
"import asyncio",
"import sys",
"import os",
"import time",
"from typing import List, Dict, Any, Optional, Callable",
"from llmdsl_runtime import (",
" GPUDetector,",
" ModelManager,",
" LocalAgent,",
" RemoteAgent,",
" RAGPipeline,",
" GraphRAGPipeline,",
" MCPServer,",
" MCPClient,",
" Orchestrator,",
" OrchestrationRule,",
" Tool,",
" ConsoleUI,",
" WebUI",
")",
"from llmdsl_runtime_advanced import (",
" RetryPolicy,",
" RetryHandler,",
" BackoffStrategy,",
" ErrorRecoveryManager,",
" StreamingAgent,",
" StreamingRemoteAgent,",
" TokenBudget,",
" BudgetAwareAgent,",
" AdvancedTool,",
" ToolParameter,",
" ToolRegistry,",
" AdvancedOrchestrator,",
" WorkflowNode,",
" OrchestrationPattern,",
" ResponseCache,",
" CachedAgent,",
" MetricsCollector,",
" MonitoredAgent",
")"
]
for imp in imports:
self.emit(imp)
def generate_chatbot(self, chatbot: ChatbotDeclaration):
"""Generate enhanced chatbot with advanced features."""
self.emit(f"class {chatbot.name}:")
self.indent()
self.emit('"""Generated chatbot with advanced features."""')
self.emit_blank_line()
self.emit("def __init__(self):")
self.indent()
self.emit('"""Initialize the chatbot."""')
self.emit("# Initialize infrastructure")
self.emit("self.gpu_detector = GPUDetector()")
self.emit("self.gpu_detector.detect()")
self.emit("self.model_manager = ModelManager(self.gpu_detector)")
self.emit("self.metrics = MetricsCollector()")
self.emit_blank_line()
if 'retry_policy' in chatbot.properties:
self.emit("# Configure retry policy")
retry_config = chatbot.properties['retry_policy']
self.emit(f"self.retry_policy = RetryPolicy(")
self.indent()
self.emit(f"max_retries={retry_config.get('max_retries', 3)},")
backoff = retry_config.get('backoff', 'exponential')
self.emit(f"backoff_strategy=BackoffStrategy.{backoff.upper()}")
self.dedent()
self.emit(")")
self.emit("self.retry_handler = RetryHandler(self.retry_policy)")
self.emit_blank_line()
model_name = chatbot.properties.get('model')
system_prompt = chatbot.properties.get('system_prompt')
agent_features = []
if chatbot.properties.get('streaming', False):
agent_features.append('streaming')
if chatbot.properties.get('caching', False):
agent_features.append('caching')
if chatbot.properties.get('monitoring', True):
agent_features.append('monitoring')
if chatbot.properties.get('token_budget', False):
agent_features.append('budget')
agent_class = self.select_agent_class(model_name, agent_features)
self.emit(f"self.agent = {agent_class}(")
self.indent()
self.emit(f"name='{chatbot.name}',")
self.emit(f"model_manager=self.model_manager,")
self.emit(f"model_name='{model_name}',")
self.emit(f"system_prompt='''{system_prompt}''',")
if 'monitoring' in agent_features:
self.emit("metrics_collector=self.metrics,")
if 'budget' in agent_features:
budget_config = chatbot.properties.get('token_budget', {})
max_per_request = budget_config.get('max_per_request', 4096)
max_per_session = budget_config.get('max_per_session', 100000)
self.emit(f"token_budget=TokenBudget(max_tokens_per_request={max_per_request}, max_tokens_per_session={max_per_session}),")
if 'caching' in agent_features:
cache_config = chatbot.properties.get('caching', {})
capacity = cache_config.get('capacity', 1000)
ttl = cache_config.get('ttl', 3600)
self.emit(f"cache_capacity={capacity},")
self.emit(f"cache_ttl={ttl},")
if self.is_remote_model(model_name):
provider = self.get_provider(model_name)
self.emit(f"provider='{provider}'")
else:
self.emit("# Local model configuration")
self.dedent()
self.emit(")")
if 'rag' in chatbot.properties:
self.emit_blank_line()
self.emit("# Initialize RAG pipeline")
rag_config = chatbot.properties['rag']
self.emit(f"self.rag = RAGPipeline(")
self.indent()
self.emit(f"vector_store_type='{rag_config.get('vector_store', 'chroma')}',")
self.emit(f"embedding_model='{rag_config.get('embedding_model')}'")
self.dedent()
self.emit(")")
self.emit("self.rag.initialize()")
if 'documents' in rag_config:
self.emit(f"self.rag.load_documents('{rag_config['documents']}')")
self.emit("self.agent.rag_pipeline = self.rag")
if 'tools' in chatbot.properties:
self.emit_blank_line()
self.emit("# Register tools")
self.emit("self.tool_registry = ToolRegistry()")
for tool_name in chatbot.properties['tools']:
self.emit(f"self.agent.add_tool(self.tool_registry.get_tool('{tool_name}'))")
self.dedent()
self.emit_blank_line()
self.emit("async def run(self):")
self.indent()
self.emit('"""Run the chatbot."""')
if self.settings and self.settings.properties.get('ui') == 'web':
port = self.settings.properties.get('port', 8080)
self.emit(f"ui = WebUI(self.agent, port={port})")
self.emit("await ui.run()")
else:
self.emit("ui = ConsoleUI(self.agent)")
self.emit("await ui.run()")
self.dedent()
if 'monitoring' in agent_features:
self.emit_blank_line()
self.emit("def get_metrics(self) -> Dict[str, Any]:")
self.indent()
self.emit('"""Get performance metrics."""')
self.emit("return self.agent.get_metrics_summary()")
self.dedent()
self.dedent()
self.emit_blank_line()
def select_agent_class(self, model_name: str, features: List[str]) -> str:
"""Select appropriate agent class based on features."""
is_remote = self.is_remote_model(model_name)
if 'streaming' in features and is_remote:
return 'StreamingRemoteAgent'
elif 'streaming' in features:
return 'StreamingAgent'
elif 'caching' in features:
return 'CachedAgent'
elif 'monitoring' in features and 'budget' in features:
return 'MonitoredAgent'
elif 'budget' in features:
return 'BudgetAwareAgent'
elif 'monitoring' in features:
return 'MonitoredAgent'
elif is_remote:
return 'RemoteAgent'
else:
return 'LocalAgent'
The enhanced code generator selects appropriate agent classes based on requested features. It generates initialization code for retry policies, token budgets, caching, and monitoring. This allows DSL users to enable advanced features declaratively without writing implementation code.
The complete LLMDSL system now provides a comprehensive solution for building production-ready LLM applications. It handles GPU detection, model management, RAG pipelines, multi-agent orchestration, error handling, streaming, caching, monitoring, and more. Developers can focus on defining what their application should do while the generated code handles all implementation details.
ADDENDUM: LLMDSL FORMAL SPECIFICATION AND ADVANCED EXAMPLES
EXTENDED BACKUS-NAUR FORM SPECIFICATION OF LLMDSL
The following EBNF grammar formally defines the complete syntax of the LLMDSL language. This specification serves as the authoritative reference for parser implementation and language documentation.
program = { declaration } ;
declaration = chatbot_declaration
| agent_declaration
| multi_agent_system_declaration
| settings_declaration
| tool_declaration
| workflow_declaration
| knowledge_base_declaration ;
chatbot_declaration = "chatbot" identifier "{" chatbot_properties "}" ;
chatbot_properties = { chatbot_property [ "," ] } ;
chatbot_property = model_property
| system_prompt_property
| temperature_property
| max_tokens_property
| rag_property
| graphrag_property
| tools_property
| streaming_property
| caching_property
| monitoring_property
| token_budget_property
| retry_policy_property
| error_handling_property ;
agent_declaration = "agent" identifier "{" agent_properties "}" ;
agent_properties = { agent_property [ "," ] } ;
agent_property = model_property
| system_prompt_property
| role_property
| temperature_property
| max_tokens_property
| tools_property
| rag_property
| graphrag_property
| memory_property
| planning_property
| reflection_property
| constraints_property ;
multi_agent_system_declaration = "multi_agent_system" identifier "{" mas_properties "}" ;
mas_properties = { mas_property [ "," ] } ;
mas_property = agents_property
| orchestration_property
| communication_property
| shared_memory_property
| coordination_property
| load_balancing_property ;
settings_declaration = "settings" "{" settings_properties "}" ;
settings_properties = { settings_property [ "," ] } ;
settings_property = gpu_property
| execution_mode_property
| max_concurrent_property
| retry_policy_property
| ui_property
| port_property
| logging_property
| metrics_property
| security_property ;
tool_declaration = "tool" identifier "{" tool_properties "}" ;
tool_properties = { tool_property [ "," ] } ;
tool_property = description_property
| parameters_property
| implementation_property
| caching_property
| timeout_property
| rate_limit_property ;
workflow_declaration = "workflow" identifier "{" workflow_properties "}" ;
workflow_properties = { workflow_property [ "," ] } ;
workflow_property = nodes_property
| edges_property
| entry_point_property
| exit_conditions_property
| error_handling_property ;
knowledge_base_declaration = "knowledge_base" identifier "{" kb_properties "}" ;
kb_properties = { kb_property [ "," ] } ;
kb_property = type_property
| source_property
| embedding_model_property
| chunk_size_property
| chunk_overlap_property
| metadata_property ;
model_property = "model" ":" string_literal ;
system_prompt_property = "system_prompt" ":" string_literal ;
role_property = "role" ":" string_literal ;
temperature_property = "temperature" ":" number_literal ;
max_tokens_property = "max_tokens" ":" integer_literal ;
rag_property = "rag" ":" "{" rag_config "}" ;
rag_config = { rag_config_item [ "," ] } ;
rag_config_item = "vector_store" ":" string_literal
| "embedding_model" ":" string_literal
| "documents" ":" string_literal
| "chunk_size" ":" integer_literal
| "chunk_overlap" ":" integer_literal
| "top_k" ":" integer_literal
| "similarity_threshold" ":" number_literal
| "reranking" ":" boolean_literal ;
graphrag_property = "graphrag" ":" "{" graphrag_config "}" ;
graphrag_config = { graphrag_config_item [ "," ] } ;
graphrag_config_item = "graph_type" ":" string_literal
| "entity_extraction" ":" string_literal
| "relationship_extraction" ":" string_literal
| "traversal_depth" ":" integer_literal
| "max_entities" ":" integer_literal ;
tools_property = "tools" ":" "[" tool_list "]" ;
tool_list = [ identifier { "," identifier } ] ;
streaming_property = "streaming" ":" boolean_literal ;
caching_property = "caching" ":" ( boolean_literal | "{" cache_config "}" ) ;
cache_config = { cache_config_item [ "," ] } ;
cache_config_item = "capacity" ":" integer_literal
| "ttl" ":" integer_literal
| "strategy" ":" string_literal ;
monitoring_property = "monitoring" ":" boolean_literal ;
token_budget_property = "token_budget" ":" "{" budget_config "}" ;
budget_config = { budget_config_item [ "," ] } ;
budget_config_item = "max_per_request" ":" integer_literal
| "max_per_session" ":" integer_literal
| "warning_threshold" ":" number_literal ;
retry_policy_property = "retry_policy" ":" "{" retry_config "}" ;
retry_config = { retry_config_item [ "," ] } ;
retry_config_item = "max_retries" ":" integer_literal
| "backoff" ":" string_literal
| "initial_delay" ":" number_literal
| "max_delay" ":" number_literal ;
error_handling_property = "error_handling" ":" "{" error_config "}" ;
error_config = { error_config_item [ "," ] } ;
error_config_item = "strategy" ":" string_literal
| "fallback_response" ":" string_literal
| "notify_on_error" ":" boolean_literal ;
agents_property = "agents" ":" "[" agent_list "]" ;
agent_list = agent_item { "," agent_item } ;
agent_item = agent_declaration | identifier ;
orchestration_property = "orchestration" ":" "{" orchestration_config "}" ;
orchestration_config = { orchestration_config_item [ "," ] } ;
orchestration_config_item = "entry_point" ":" identifier
| "pattern" ":" string_literal
| "routing" ":" "{" routing_rules "}"
| "parallel_execution" ":" boolean_literal
| "timeout" ":" number_literal ;
routing_rules = routing_rule { "," routing_rule } ;
routing_rule = identifier "->" identifier [ ":" "when" condition_expression ] ;
condition_expression = lambda_expression | string_literal ;
lambda_expression = "lambda" parameter_list ":" expression ;
parameter_list = identifier { "," identifier } ;
expression = comparison_expression
| logical_expression
| arithmetic_expression
| function_call
| identifier
| literal ;
comparison_expression = expression comparison_operator expression ;
comparison_operator = "==" | "!=" | "<" | ">" | "<=" | ">=" | "in" | "contains" ;
logical_expression = expression logical_operator expression ;
logical_operator = "and" | "or" | "not" ;
arithmetic_expression = expression arithmetic_operator expression ;
arithmetic_operator = "+" | "-" | "*" | "/" | "%" | "**" ;
function_call = identifier "(" [ argument_list ] ")" ;
argument_list = expression { "," expression } ;
communication_property = "communication" ":" string_literal ;
shared_memory_property = "shared_memory" ":" "{" memory_config "}" ;
memory_config = { memory_config_item [ "," ] } ;
memory_config_item = "type" ":" string_literal
| "capacity" ":" integer_literal
| "persistence" ":" boolean_literal ;
coordination_property = "coordination" ":" "{" coordination_config "}" ;
coordination_config = { coordination_config_item [ "," ] } ;
coordination_config_item = "strategy" ":" string_literal
| "consensus_threshold" ":" number_literal
| "voting_mechanism" ":" string_literal ;
load_balancing_property = "load_balancing" ":" "{" lb_config "}" ;
lb_config = { lb_config_item [ "," ] } ;
lb_config_item = "strategy" ":" string_literal
| "max_queue_size" ":" integer_literal
| "timeout" ":" number_literal ;
gpu_property = "gpu" ":" string_literal ;
execution_mode_property = "execution_mode" ":" string_literal ;
max_concurrent_property = "max_concurrent" ":" integer_literal ;
ui_property = "ui" ":" string_literal ;
port_property = "port" ":" integer_literal ;
logging_property = "logging" ":" "{" logging_config "}" ;
logging_config = { logging_config_item [ "," ] } ;
logging_config_item = "level" ":" string_literal
| "output" ":" string_literal
| "format" ":" string_literal ;
metrics_property = "metrics" ":" "{" metrics_config "}" ;
metrics_config = { metrics_config_item [ "," ] } ;
metrics_config_item = "enabled" ":" boolean_literal
| "export_interval" ":" integer_literal
| "export_format" ":" string_literal ;
security_property = "security" ":" "{" security_config "}" ;
security_config = { security_config_item [ "," ] } ;
security_config_item = "authentication" ":" boolean_literal
| "rate_limiting" ":" boolean_literal
| "allowed_origins" ":" "[" string_list "]" ;
memory_property = "memory" ":" "{" memory_type_config "}" ;
memory_type_config = { memory_type_config_item [ "," ] } ;
memory_type_config_item = "type" ":" string_literal
| "window_size" ":" integer_literal
| "summarization" ":" boolean_literal ;
planning_property = "planning" ":" "{" planning_config "}" ;
planning_config = { planning_config_item [ "," ] } ;
planning_config_item = "enabled" ":" boolean_literal
| "max_steps" ":" integer_literal
| "strategy" ":" string_literal ;
reflection_property = "reflection" ":" "{" reflection_config "}" ;
reflection_config = { reflection_config_item [ "," ] } ;
reflection_config_item = "enabled" ":" boolean_literal
| "frequency" ":" string_literal
| "criteria" ":" string_literal ;
constraints_property = "constraints" ":" "{" constraints_config "}" ;
constraints_config = { constraints_config_item [ "," ] } ;
constraints_config_item = "max_execution_time" ":" number_literal
| "max_tool_calls" ":" integer_literal
| "allowed_tools" ":" "[" string_list "]" ;
description_property = "description" ":" string_literal ;
parameters_property = "parameters" ":" "[" parameter_definitions "]" ;
parameter_definitions = parameter_definition { "," parameter_definition } ;
parameter_definition = "{" param_def_items "}" ;
param_def_items = { param_def_item [ "," ] } ;
param_def_item = "name" ":" string_literal
| "type" ":" string_literal
| "required" ":" boolean_literal
| "default" ":" literal
| "description" ":" string_literal ;
implementation_property = "implementation" ":" string_literal ;
timeout_property = "timeout" ":" number_literal ;
rate_limit_property = "rate_limit" ":" "{" rate_limit_config "}" ;
rate_limit_config = { rate_limit_config_item [ "," ] } ;
rate_limit_config_item = "requests_per_minute" ":" integer_literal
| "burst_size" ":" integer_literal ;
nodes_property = "nodes" ":" "[" node_list "]" ;
node_list = node_definition { "," node_definition } ;
node_definition = "{" node_def_items "}" ;
node_def_items = { node_def_item [ "," ] } ;
node_def_item = "id" ":" string_literal
| "agent" ":" identifier
| "pattern" ":" string_literal
| "condition" ":" condition_expression ;
edges_property = "edges" ":" "[" edge_list "]" ;
edge_list = edge_definition { "," edge_definition } ;
edge_definition = "{" edge_def_items "}" ;
edge_def_items = { edge_def_item [ "," ] } ;
edge_def_item = "from" ":" string_literal
| "to" ":" string_literal
| "condition" ":" condition_expression ;
entry_point_property = "entry_point" ":" identifier ;
exit_conditions_property = "exit_conditions" ":" "[" condition_list "]" ;
condition_list = condition_expression { "," condition_expression } ;
type_property = "type" ":" string_literal ;
source_property = "source" ":" string_literal ;
embedding_model_property = "embedding_model" ":" string_literal ;
chunk_size_property = "chunk_size" ":" integer_literal ;
chunk_overlap_property = "chunk_overlap" ":" integer_literal ;
metadata_property = "metadata" ":" "{" metadata_items "}" ;
metadata_items = { metadata_item [ "," ] } ;
metadata_item = identifier ":" literal ;
string_list = [ string_literal { "," string_literal } ] ;
identifier = letter { letter | digit | "_" | "-" } ;
string_literal = '"' { character } '"' | "'" { character } "'" ;
integer_literal = digit { digit } ;
number_literal = integer_literal [ "." digit { digit } ] ;
boolean_literal = "true" | "false" ;
literal = string_literal | number_literal | boolean_literal ;
letter = "a" | "b" | "c" | "d" | "e" | "f" | "g" | "h" | "i" | "j" | "k" | "l" | "m"
| "n" | "o" | "p" | "q" | "r" | "s" | "t" | "u" | "v" | "w" | "x" | "y" | "z"
| "A" | "B" | "C" | "D" | "E" | "F" | "G" | "H" | "I" | "J" | "K" | "L" | "M"
| "N" | "O" | "P" | "Q" | "R" | "S" | "T" | "U" | "V" | "W" | "X" | "Y" | "Z" ;
digit = "0" | "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9" ;
character = letter | digit | symbol | escape_sequence ;
symbol = "!" | "@" | "#" | "$" | "%" | "^" | "&" | "*" | "(" | ")" | "-" | "+" | "="
| "{" | "}" | "[" | "]" | "|" | "\" | ":" | ";" | "'" | '"' | "<" | ">" | ","
| "." | "?" | "/" | " " | "\t" | "\n" ;
escape_sequence = "\" ( "n" | "t" | "\" | '"' | "'" ) ;
This EBNF specification provides a complete formal definition of the LLMDSL syntax. It defines all valid language constructs including declarations, properties, expressions, and literals. Parser implementations should strictly adhere to this grammar to ensure consistent behavior.
COMPLEX EXAMPLE ONE: ENTERPRISE CUSTOMER SUPPORT SYSTEM
This example demonstrates a sophisticated multi-agent customer support system with RAG, GraphRAG, multiple specialized agents, advanced orchestration, monitoring, and error handling.
# enterprise_support_system.llmdsl
# Advanced customer support system with multiple specialized agents
knowledge_base CustomerKnowledgeBase {
type: "hybrid",
source: "./customer_data",
embedding_model: "sentence-transformers/all-mpnet-base-v2",
chunk_size: 800,
chunk_overlap: 150,
metadata: {
domain: "customer_support",
language: "en",
version: "2.0"
}
}
knowledge_base ProductKnowledgeGraph {
type: "graph",
source: "./product_graph",
embedding_model: "sentence-transformers/all-mpnet-base-v2",
metadata: {
graph_type: "product_relationships",
entity_types: ["product", "feature", "issue", "solution"]
}
}
tool web_search {
description: "Search the web for current information",
parameters: [
{
name: "query",
type: "string",
required: true,
description: "Search query string"
},
{
name: "num_results",
type: "integer",
required: false,
default: 5,
description: "Number of results to return"
}
],
implementation: "tools.web_search_impl",
caching: {
capacity: 500,
ttl: 1800
},
timeout: 10.0,
rate_limit: {
requests_per_minute: 30,
burst_size: 10
}
}
tool ticket_system {
description: "Create or update support tickets",
parameters: [
{
name: "action",
type: "string",
required: true,
description: "Action to perform: create, update, close"
},
{
name: "ticket_id",
type: "string",
required: false,
description: "Ticket ID for update/close actions"
},
{
name: "details",
type: "string",
required: true,
description: "Ticket details or update information"
}
],
implementation: "tools.ticket_system_impl",
timeout: 5.0
}
tool knowledge_retrieval {
description: "Retrieve information from knowledge base",
parameters: [
{
name: "query",
type: "string",
required: true,
description: "Query for knowledge retrieval"
},
{
name: "top_k",
type: "integer",
required: false,
default: 5,
description: "Number of results to retrieve"
}
],
implementation: "tools.knowledge_retrieval_impl",
caching: true
}
agent IntakeAgent {
model: "gpt-4",
role: "Initial customer inquiry handler and classifier",
system_prompt: "You are an intake specialist for customer support. Your role is to understand customer inquiries, extract key information, classify the issue type, determine urgency, and route to the appropriate specialist. Be empathetic, professional, and thorough in gathering information. Always confirm your understanding before routing.",
temperature: 0.3,
max_tokens: 1024,
tools: [knowledge_retrieval],
memory: {
type: "conversation",
window_size: 10,
summarization: true
},
monitoring: true,
token_budget: {
max_per_request: 2048,
max_per_session: 50000,
warning_threshold: 0.8
}
}
agent TechnicalSupportAgent {
model: "gpt-4",
role: "Technical issue resolution specialist",
system_prompt: "You are a senior technical support engineer. You have deep expertise in troubleshooting complex technical issues. Use the knowledge base and product graph to find solutions. When needed, search for recent information. Provide step-by-step guidance and verify customer understanding. Create tickets for unresolved issues requiring engineering escalation.",
temperature: 0.2,
max_tokens: 2048,
tools: [web_search, ticket_system, knowledge_retrieval],
rag: {
vector_store: "chroma",
embedding_model: "sentence-transformers/all-mpnet-base-v2",
documents: "./technical_docs",
chunk_size: 800,
chunk_overlap: 150,
top_k: 5,
similarity_threshold: 0.7,
reranking: true
},
graphrag: {
graph_type: "technical_knowledge",
entity_extraction: "automatic",
relationship_extraction: "automatic",
traversal_depth: 3,
max_entities: 20
},
planning: {
enabled: true,
max_steps: 10,
strategy: "react"
},
reflection: {
enabled: true,
frequency: "after_each_step",
criteria: "solution_quality"
},
constraints: {
max_execution_time: 300.0,
max_tool_calls: 15,
allowed_tools: ["web_search", "ticket_system", "knowledge_retrieval"]
},
monitoring: true
}
agent BillingAgent {
model: "gpt-3.5-turbo",
role: "Billing and account specialist",
system_prompt: "You are a billing and account specialist. Handle all billing inquiries, payment issues, subscription changes, and account management. Always verify customer identity before discussing account details. Be clear about pricing, refund policies, and billing cycles. Escalate complex billing disputes appropriately.",
temperature: 0.1,
max_tokens: 1024,
tools: [ticket_system, knowledge_retrieval],
rag: {
vector_store: "chroma",
embedding_model: "sentence-transformers/all-mpnet-base-v2",
documents: "./billing_docs",
chunk_size: 600,
chunk_overlap: 100,
top_k: 3
},
memory: {
type: "conversation",
window_size: 8,
summarization: false
},
constraints: {
max_execution_time: 180.0,
max_tool_calls: 8
},
monitoring: true
}
agent ProductExpertAgent {
model: "gpt-4",
role: "Product features and usage specialist",
system_prompt: "You are a product expert who helps customers understand and maximize the value of our products. Explain features clearly, provide usage tips, suggest best practices, and help customers discover functionality they may not know about. Use the product knowledge graph to provide comprehensive, interconnected information about features and their relationships.",
temperature: 0.4,
max_tokens: 1536,
tools: [web_search, knowledge_retrieval],
graphrag: {
graph_type: "product_features",
entity_extraction: "automatic",
relationship_extraction: "automatic",
traversal_depth: 2,
max_entities: 15
},
rag: {
vector_store: "chroma",
embedding_model: "sentence-transformers/all-mpnet-base-v2",
documents: "./product_docs",
chunk_size: 1000,
chunk_overlap: 200,
top_k: 7
},
monitoring: true
}
agent EscalationAgent {
model: "gpt-4",
role: "Complex issue handler and escalation manager",
system_prompt: "You handle complex, sensitive, or escalated customer issues. Review the conversation history, understand the full context, identify what previous agents attempted, and determine the best resolution path. You have authority to make exceptions within guidelines. Always maintain professionalism and focus on customer satisfaction while protecting company interests.",
temperature: 0.3,
max_tokens: 2048,
tools: [web_search, ticket_system, knowledge_retrieval],
rag: {
vector_store: "chroma",
embedding_model: "sentence-transformers/all-mpnet-base-v2",
documents: "./escalation_docs",
chunk_size: 1000,
chunk_overlap: 200,
top_k: 5
},
memory: {
type: "conversation",
window_size: 20,
summarization: true
},
planning: {
enabled: true,
max_steps: 8,
strategy: "chain_of_thought"
},
monitoring: true
}
agent QualityAssuranceAgent {
model: "gpt-3.5-turbo",
role: "Conversation quality evaluator",
system_prompt: "You evaluate the quality of customer support interactions. Review conversation transcripts and assess: problem resolution, agent professionalism, adherence to policies, customer satisfaction indicators, and opportunities for improvement. Provide constructive feedback and quality scores.",
temperature: 0.2,
max_tokens: 1024,
monitoring: true
}
workflow SupportWorkflow {
nodes: [
{
id: "intake",
agent: IntakeAgent,
pattern: "sequential"
},
{
id: "technical",
agent: TechnicalSupportAgent,
pattern: "sequential"
},
{
id: "billing",
agent: BillingAgent,
pattern: "sequential"
},
{
id: "product",
agent: ProductExpertAgent,
pattern: "sequential"
},
{
id: "escalation",
agent: EscalationAgent,
pattern: "sequential"
},
{
id: "quality",
agent: QualityAssuranceAgent,
pattern: "sequential"
}
],
edges: [
{
from: "intake",
to: "technical",
condition: lambda output: "technical" in output.lower() or "bug" in output.lower() or "error" in output.lower()
},
{
from: "intake",
to: "billing",
condition: lambda output: "billing" in output.lower() or "payment" in output.lower() or "subscription" in output.lower()
},
{
from: "intake",
to: "product",
condition: lambda output: "feature" in output.lower() or "how to" in output.lower() or "usage" in output.lower()
},
{
from: "technical",
to: "escalation",
condition: lambda output: "unresolved" in output.lower() or "escalate" in output.lower()
},
{
from: "billing",
to: "escalation",
condition: lambda output: "dispute" in output.lower() or "escalate" in output.lower()
},
{
from: "product",
to: "escalation",
condition: lambda output: "escalate" in output.lower()
}
],
entry_point: intake,
exit_conditions: [
lambda output: "resolved" in output.lower(),
lambda output: "ticket created" in output.lower()
],
error_handling: {
strategy: "graceful_degradation",
fallback_response: "I apologize, but I'm experiencing technical difficulties. I've created a support ticket and our team will contact you shortly.",
notify_on_error: true
}
}
multi_agent_system EnterpriseSupport {
agents: [
IntakeAgent,
TechnicalSupportAgent,
BillingAgent,
ProductExpertAgent,
EscalationAgent,
QualityAssuranceAgent
],
orchestration: {
entry_point: IntakeAgent,
pattern: "dynamic_routing",
routing: {
IntakeAgent -> TechnicalSupportAgent: when lambda x: "category" in x and x["category"] == "technical",
IntakeAgent -> BillingAgent: when lambda x: "category" in x and x["category"] == "billing",
IntakeAgent -> ProductExpertAgent: when lambda x: "category" in x and x["category"] == "product",
TechnicalSupportAgent -> EscalationAgent: when lambda x: "escalate" in str(x).lower(),
BillingAgent -> EscalationAgent: when lambda x: "escalate" in str(x).lower(),
ProductExpertAgent -> EscalationAgent: when lambda x: "escalate" in str(x).lower(),
EscalationAgent -> QualityAssuranceAgent: when lambda x: "resolved" in str(x).lower()
},
parallel_execution: false,
timeout: 600.0
},
communication: "mcp",
shared_memory: {
type: "distributed",
capacity: 10000,
persistence: true
},
coordination: {
strategy: "hierarchical",
consensus_threshold: 0.7,
voting_mechanism: "weighted"
},
load_balancing: {
strategy: "least_loaded",
max_queue_size: 100,
timeout: 30.0
}
}
settings {
gpu: "auto",
execution_mode: "async",
max_concurrent: 10,
retry_policy: {
max_retries: 3,
backoff: "jittered_exponential",
initial_delay: 1.0,
max_delay: 30.0
},
ui: "web",
port: 8080,
logging: {
level: "info",
output: "./logs/support_system.log",
format: "json"
},
metrics: {
enabled: true,
export_interval: 60,
export_format: "prometheus"
},
security: {
authentication: true,
rate_limiting: true,
allowed_origins: ["https://support.company.com", "https://app.company.com"]
}
}
This enterprise customer support system demonstrates advanced LLMDSL features including multiple specialized agents with different models and configurations, hybrid RAG and GraphRAG knowledge retrieval, sophisticated workflow orchestration with conditional routing, comprehensive tool integration with rate limiting and caching, shared memory and coordination between agents, quality assurance evaluation, extensive monitoring and metrics, security controls and authentication, error handling with graceful degradation, and production-ready logging and configuration.
COMPLEX EXAMPLE TWO: RESEARCH ASSISTANT WITH MULTI-STAGE ANALYSIS
This example shows a research assistant system that performs multi-stage analysis with parallel processing, iterative refinement, and comprehensive knowledge integration.
# research_assistant_system.llmdsl
# Advanced research assistant with multi-stage analysis pipeline
knowledge_base AcademicPapers {
type: "vector",
source: "./academic_papers",
embedding_model: "sentence-transformers/allenai-specter",
chunk_size: 1200,
chunk_overlap: 300,
metadata: {
domain: "academic_research",
citation_tracking: true,
version_control: true
}
}
knowledge_base ResearchGraph {
type: "graph",
source: "./research_graph",
embedding_model: "sentence-transformers/allenai-specter",
metadata: {
graph_type: "citation_network",
entity_types: ["paper", "author", "concept", "methodology", "finding"],
relationship_types: ["cites", "authored_by", "uses_method", "contradicts", "supports"]
}
}
tool arxiv_search {
description: "Search arXiv for recent papers",
parameters: [
{
name: "query",
type: "string",
required: true,
description: "Search query for arXiv"
},
{
name: "max_results",
type: "integer",
required: false,
default: 10,
description: "Maximum number of papers to retrieve"
},
{
name: "sort_by",
type: "string",
required: false,
default: "relevance",
description: "Sort order: relevance, date"
}
],
implementation: "tools.arxiv_search_impl",
caching: {
capacity: 200,
ttl: 86400
},
timeout: 15.0
}
tool semantic_scholar {
description: "Search Semantic Scholar for papers and citations",
parameters: [
{
name: "query",
type: "string",
required: true,
description: "Search query"
},
{
name: "fields",
type: "string",
required: false,
default: "title,authors,abstract,citations",
description: "Fields to retrieve"
}
],
implementation: "tools.semantic_scholar_impl",
caching: true,
timeout: 20.0,
rate_limit: {
requests_per_minute: 100,
burst_size: 20
}
}
tool citation_analyzer {
description: "Analyze citation patterns and impact",
parameters: [
{
name: "paper_id",
type: "string",
required: true,
description: "Paper identifier"
},
{
name: "depth",
type: "integer",
required: false,
default: 2,
description: "Citation graph traversal depth"
}
],
implementation: "tools.citation_analyzer_impl",
caching: true,
timeout: 30.0
}
tool statistical_analysis {
description: "Perform statistical analysis on data",
parameters: [
{
name: "data",
type: "string",
required: true,
description: "Data to analyze in JSON format"
},
{
name: "analysis_type",
type: "string",
required: true,
description: "Type of analysis: descriptive, correlation, regression, hypothesis_test"
}
],
implementation: "tools.statistical_analysis_impl",
timeout: 60.0
}
tool code_executor {
description: "Execute Python code for data analysis",
parameters: [
{
name: "code",
type: "string",
required: true,
description: "Python code to execute"
},
{
name: "timeout",
type: "integer",
required: false,
default: 30,
description: "Execution timeout in seconds"
}
],
implementation: "tools.code_executor_impl",
timeout: 60.0
}
agent LiteratureReviewAgent {
model: "gpt-4",
role: "Comprehensive literature review specialist",
system_prompt: "You are an expert at conducting systematic literature reviews. Search academic databases, identify relevant papers, extract key findings, synthesize information across multiple sources, identify research gaps, and provide comprehensive summaries. Use citation analysis to understand the impact and relationships between papers. Always cite sources properly.",
temperature: 0.3,
max_tokens: 3072,
tools: [arxiv_search, semantic_scholar, citation_analyzer, knowledge_retrieval],
rag: {
vector_store: "chroma",
embedding_model: "sentence-transformers/allenai-specter",
documents: "./academic_papers",
chunk_size: 1200,
chunk_overlap: 300,
top_k: 10,
similarity_threshold: 0.65,
reranking: true
},
graphrag: {
graph_type: "citation_network",
entity_extraction: "automatic",
relationship_extraction: "automatic",
traversal_depth: 3,
max_entities: 50
},
planning: {
enabled: true,
max_steps: 15,
strategy: "tree_of_thought"
},
memory: {
type: "conversation",
window_size: 30,
summarization: true
},
streaming: true,
monitoring: true,
token_budget: {
max_per_request: 8192,
max_per_session: 200000
}
}
agent MethodologyAnalystAgent {
model: "gpt-4",
role: "Research methodology expert",
system_prompt: "You are an expert in research methodologies. Analyze research designs, evaluate methodological rigor, identify potential biases, assess validity and reliability, compare different approaches, and suggest improvements. Provide detailed critiques of experimental designs, statistical methods, and data collection procedures.",
temperature: 0.2,
max_tokens: 2048,
tools: [statistical_analysis, code_executor, knowledge_retrieval],
rag: {
vector_store: "chroma",
embedding_model: "sentence-transformers/allenai-specter",
documents: "./methodology_docs",
chunk_size: 1000,
chunk_overlap: 200,
top_k: 8
},
planning: {
enabled: true,
max_steps: 12,
strategy: "react"
},
reflection: {
enabled: true,
frequency: "after_each_step",
criteria: "methodological_soundness"
},
monitoring: true
}
agent DataAnalystAgent {
model: "gpt-4",
role: "Quantitative data analysis specialist",
system_prompt: "You are a data analysis expert. Perform statistical analyses, create visualizations, interpret results, identify patterns and trends, test hypotheses, and provide actionable insights. You can write and execute Python code for complex analyses. Always validate assumptions and report limitations.",
temperature: 0.1,
max_tokens: 2048,
tools: [statistical_analysis, code_executor],
planning: {
enabled: true,
max_steps: 10,
strategy: "react"
},
constraints: {
max_execution_time: 300.0,
max_tool_calls: 20
},
monitoring: true
}
agent SynthesisAgent {
model: "gpt-4",
role: "Research synthesis and integration specialist",
system_prompt: "You synthesize information from multiple sources to create coherent, comprehensive analyses. Identify common themes, reconcile contradictory findings, build theoretical frameworks, generate novel insights, and create structured summaries. Your syntheses should be well-organized, properly cited, and highlight both consensus and disagreement in the literature.",
temperature: 0.4,
max_tokens: 4096,
tools: [knowledge_retrieval],
graphrag: {
graph_type: "knowledge_synthesis",
entity_extraction: "automatic",
relationship_extraction: "automatic",
traversal_depth: 4,
max_entities: 100
},
memory: {
type: "conversation",
window_size: 50,
summarization: true
},
planning: {
enabled: true,
max_steps: 20,
strategy: "chain_of_thought"
},
streaming: true,
monitoring: true
}
agent CriticalEvaluatorAgent {
model: "gpt-4",
role: "Critical evaluation and quality assessment specialist",
system_prompt: "You critically evaluate research quality, identify strengths and weaknesses, assess evidence quality, detect logical fallacies, evaluate argument coherence, and provide constructive criticism. Be thorough, fair, and balanced in your assessments. Highlight both positive aspects and areas for improvement.",
temperature: 0.3,
max_tokens: 2048,
reflection: {
enabled: true,
frequency: "after_completion",
criteria: "evaluation_thoroughness"
},
monitoring: true
}
agent WritingAssistantAgent {
model: "gpt-4",
role: "Academic writing specialist",
system_prompt: "You help write clear, well-structured academic content. Draft sections, revise text for clarity and coherence, ensure proper citation format, maintain academic tone, organize arguments logically, and polish prose. Follow academic writing conventions and adapt style to different publication venues.",
temperature: 0.5,
max_tokens: 3072,
streaming: true,
monitoring: true
}
workflow ResearchPipeline {
nodes: [
{
id: "literature_review",
agent: LiteratureReviewAgent,
pattern: "sequential"
},
{
id: "parallel_analysis",
agent: MethodologyAnalystAgent,
pattern: "parallel"
},
{
id: "data_analysis",
agent: DataAnalystAgent,
pattern: "sequential"
},
{
id: "synthesis",
agent: SynthesisAgent,
pattern: "sequential"
},
{
id: "evaluation",
agent: CriticalEvaluatorAgent,
pattern: "sequential"
},
{
id: "writing",
agent: WritingAssistantAgent,
pattern: "sequential"
},
{
id: "refinement",
agent: SynthesisAgent,
pattern: "loop",
condition: lambda output: "needs_refinement" not in output.lower()
}
],
edges: [
{
from: "literature_review",
to: "parallel_analysis"
},
{
from: "parallel_analysis",
to: "data_analysis"
},
{
from: "data_analysis",
to: "synthesis"
},
{
from: "synthesis",
to: "evaluation"
},
{
from: "evaluation",
to: "writing",
condition: lambda output: "quality_acceptable" in output.lower()
},
{
from: "evaluation",
to: "refinement",
condition: lambda output: "needs_improvement" in output.lower()
},
{
from: "refinement",
to: "evaluation"
}
],
entry_point: literature_review,
exit_conditions: [
lambda output: "research_complete" in output.lower(),
lambda output: "final_draft_ready" in output.lower()
],
error_handling: {
strategy: "retry_with_fallback",
fallback_response: "Unable to complete this research task. Please review the partial results and provide additional guidance.",
notify_on_error: true
}
}
multi_agent_system ResearchAssistant {
agents: [
LiteratureReviewAgent,
MethodologyAnalystAgent,
DataAnalystAgent,
SynthesisAgent,
CriticalEvaluatorAgent,
WritingAssistantAgent
],
orchestration: {
entry_point: LiteratureReviewAgent,
pattern: "pipeline_with_feedback",
routing: {
LiteratureReviewAgent -> MethodologyAnalystAgent: when lambda x: "literature_complete" in str(x).lower(),
LiteratureReviewAgent -> DataAnalystAgent: when lambda x: "data_available" in str(x).lower(),
MethodologyAnalystAgent -> SynthesisAgent: when lambda x: "analysis_complete" in str(x).lower(),
DataAnalystAgent -> SynthesisAgent: when lambda x: "analysis_complete" in str(x).lower(),
SynthesisAgent -> CriticalEvaluatorAgent: when lambda x: "synthesis_complete" in str(x).lower(),
CriticalEvaluatorAgent -> WritingAssistantAgent: when lambda x: "approved" in str(x).lower(),
CriticalEvaluatorAgent -> SynthesisAgent: when lambda x: "revise" in str(x).lower()
},
parallel_execution: true,
timeout: 1800.0
},
communication: "mcp",
shared_memory: {
type: "hierarchical",
capacity: 50000,
persistence: true
},
coordination: {
strategy: "consensus_based",
consensus_threshold: 0.8,
voting_mechanism: "quality_weighted"
}
}
settings {
gpu: "auto",
execution_mode: "async",
max_concurrent: 8,
retry_policy: {
max_retries: 5,
backoff: "exponential",
initial_delay: 2.0,
max_delay: 60.0
},
ui: "web",
port: 8000,
logging: {
level: "debug",
output: "./logs/research_assistant.log",
format: "structured"
},
metrics: {
enabled: true,
export_interval: 30,
export_format: "json"
},
security: {
authentication: true,
rate_limiting: true,
allowed_origins: ["https://research.university.edu"]
}
}
This research assistant system demonstrates pipeline orchestration with feedback loops, parallel processing of multiple analysis stages, iterative refinement based on quality evaluation, integration of multiple knowledge sources including citation graphs, sophisticated planning strategies including tree of thought and chain of thought reasoning, streaming responses for long-form content generation, comprehensive tool integration for academic search and data analysis, quality assessment and critical evaluation, and production-ready configuration with extensive monitoring.
COMPLEX EXAMPLE THREE: AUTONOMOUS SOFTWARE DEVELOPMENT TEAM
This final example shows an autonomous software development system with multiple specialized agents collaborating on software projects.
# autonomous_dev_team.llmdsl
# Multi-agent software development system
knowledge_base CodebaseKnowledge {
type: "hybrid",
source: "./codebase",
embedding_model: "microsoft/codebert-base",
chunk_size: 500,
chunk_overlap: 50,
metadata: {
language: "python",
framework: "multiple",
indexing: "ast_aware"
}
}
knowledge_base ArchitectureGraph {
type: "graph",
source: "./architecture_graph",
embedding_model: "microsoft/codebert-base",
metadata: {
graph_type: "software_architecture",
entity_types: ["module", "class", "function", "dependency", "pattern"],
relationship_types: ["imports", "calls", "inherits", "implements", "depends_on"]
}
}
tool code_analyzer {
description: "Analyze code quality and complexity",
parameters: [
{
name: "code",
type: "string",
required: true,
description: "Code to analyze"
},
{
name: "language",
type: "string",
required: false,
default: "python",
description: "Programming language"
}
],
implementation: "tools.code_analyzer_impl",
timeout: 30.0
}
tool test_runner {
description: "Run tests and report results",
parameters: [
{
name: "test_path",
type: "string",
required: true,
description: "Path to tests"
},
{
name: "coverage",
type: "boolean",
required: false,
default: true,
description: "Generate coverage report"
}
],
implementation: "tools.test_runner_impl",
timeout: 120.0
}
tool git_operations {
description: "Perform git operations",
parameters: [
{
name: "operation",
type: "string",
required: true,
description: "Git operation: commit, branch, merge, diff"
},
{
name: "parameters",
type: "string",
required: true,
description: "Operation parameters as JSON"
}
],
implementation: "tools.git_operations_impl",
timeout: 60.0
}
tool documentation_generator {
description: "Generate code documentation",
parameters: [
{
name: "code_path",
type: "string",
required: true,
description: "Path to code for documentation"
},
{
name: "format",
type: "string",
required: false,
default: "markdown",
description: "Documentation format"
}
],
implementation: "tools.documentation_generator_impl",
caching: true,
timeout: 60.0
}
agent ProductManagerAgent {
model: "gpt-4",
role: "Product requirements and planning specialist",
system_prompt: "You are a product manager who translates user needs into clear technical requirements. Create user stories, define acceptance criteria, prioritize features, plan sprints, and ensure alignment with product vision. Break down complex features into manageable tasks.",
temperature: 0.4,
max_tokens: 2048,
tools: [knowledge_retrieval],
planning: {
enabled: true,
max_steps: 15,
strategy: "hierarchical"
},
monitoring: true
}
agent ArchitectAgent {
model: "gpt-4",
role: "Software architecture and design specialist",
system_prompt: "You are a software architect. Design system architecture, define component interactions, choose appropriate design patterns, ensure scalability and maintainability, create technical specifications, and review architectural decisions. Use the architecture graph to understand existing system structure.",
temperature: 0.3,
max_tokens: 3072,
tools: [code_analyzer, knowledge_retrieval],
graphrag: {
graph_type: "software_architecture",
entity_extraction: "automatic",
relationship_extraction: "automatic",
traversal_depth: 4,
max_entities: 50
},
rag: {
vector_store: "chroma",
embedding_model: "microsoft/codebert-base",
documents: "./architecture_docs",
chunk_size: 800,
chunk_overlap: 150,
top_k: 8
},
planning: {
enabled: true,
max_steps: 20,
strategy: "tree_of_thought"
},
reflection: {
enabled: true,
frequency: "after_each_step",
criteria: "architectural_soundness"
},
monitoring: true
}
agent DeveloperAgent {
model: "gpt-4",
role: "Software implementation specialist",
system_prompt: "You are an expert software developer. Write clean, efficient, well-documented code following best practices. Implement features according to specifications, handle edge cases, write meaningful tests, and ensure code quality. Use existing codebase patterns and maintain consistency.",
temperature: 0.2,
max_tokens: 4096,
tools: [code_analyzer, test_runner, git_operations, knowledge_retrieval],
rag: {
vector_store: "chroma",
embedding_model: "microsoft/codebert-base",
documents: "./codebase",
chunk_size: 500,
chunk_overlap: 50,
top_k: 10
},
planning: {
enabled: true,
max_steps: 25,
strategy: "react"
},
constraints: {
max_execution_time: 600.0,
max_tool_calls: 30
},
streaming: true,
monitoring: true,
token_budget: {
max_per_request: 8192,
max_per_session: 300000
}
}
agent QAEngineerAgent {
model: "gpt-4",
role: "Quality assurance and testing specialist",
system_prompt: "You are a QA engineer. Design comprehensive test strategies, write unit tests, integration tests, and end-to-end tests, identify edge cases, perform code reviews for testability, ensure adequate coverage, and validate functionality against requirements.",
temperature: 0.2,
max_tokens: 2048,
tools: [test_runner, code_analyzer, knowledge_retrieval],
planning: {
enabled: true,
max_steps: 15,
strategy: "react"
},
reflection: {
enabled: true,
frequency: "after_completion",
criteria: "test_coverage"
},
monitoring: true
}
agent CodeReviewerAgent {
model: "gpt-4",
role: "Code review and quality specialist",
system_prompt: "You perform thorough code reviews. Check for bugs, security vulnerabilities, performance issues, code style violations, maintainability concerns, and adherence to best practices. Provide constructive feedback with specific suggestions for improvement. Ensure code meets quality standards before approval.",
temperature: 0.1,
max_tokens: 2048,
tools: [code_analyzer],
reflection: {
enabled: true,
frequency: "after_each_step",
criteria: "review_thoroughness"
},
monitoring: true
}
agent DocumentationAgent {
model: "gpt-3.5-turbo",
role: "Technical documentation specialist",
system_prompt: "You create clear, comprehensive technical documentation. Write API documentation, user guides, architecture documents, code comments, and README files. Ensure documentation is accurate, up-to-date, and accessible to the target audience.",
temperature: 0.4,
max_tokens: 2048,
tools: [documentation_generator, knowledge_retrieval],
streaming: true,
monitoring: true
}
agent DevOpsAgent {
model: "gpt-4",
role: "DevOps and deployment specialist",
system_prompt: "You handle deployment, CI/CD, infrastructure, and operational concerns. Configure build pipelines, manage deployments, monitor system health, optimize performance, and ensure reliability. Implement infrastructure as code and automate operational tasks.",
temperature: 0.2,
max_tokens: 1536,
tools: [git_operations, code_analyzer],
planning: {
enabled: true,
max_steps: 12,
strategy: "react"
},
monitoring: true
}
workflow DevelopmentWorkflow {
nodes: [
{
id: "requirements",
agent: ProductManagerAgent,
pattern: "sequential"
},
{
id: "architecture",
agent: ArchitectAgent,
pattern: "sequential"
},
{
id: "implementation",
agent: DeveloperAgent,
pattern: "sequential"
},
{
id: "testing",
agent: QAEngineerAgent,
pattern: "parallel"
},
{
id: "review",
agent: CodeReviewerAgent,
pattern: "sequential"
},
{
id: "documentation",
agent: DocumentationAgent,
pattern: "parallel"
},
{
id: "deployment",
agent: DevOpsAgent,
pattern: "sequential"
},
{
id: "iteration",
agent: DeveloperAgent,
pattern: "loop",
condition: lambda output: "tests_passing" in output.lower() and "review_approved" in output.lower()
}
],
edges: [
{
from: "requirements",
to: "architecture"
},
{
from: "architecture",
to: "implementation"
},
{
from: "implementation",
to: "testing"
},
{
from: "implementation",
to: "documentation"
},
{
from: "testing",
to: "review",
condition: lambda output: "tests_passing" in output.lower()
},
{
from: "testing",
to: "iteration",
condition: lambda output: "tests_failing" in output.lower()
},
{
from: "review",
to: "deployment",
condition: lambda output: "approved" in output.lower()
},
{
from: "review",
to: "iteration",
condition: lambda output: "changes_requested" in output.lower()
},
{
from: "iteration",
to: "testing"
}
],
entry_point: requirements,
exit_conditions: [
lambda output: "deployed_successfully" in output.lower(),
lambda output: "feature_complete" in output.lower()
],
error_handling: {
strategy: "checkpoint_recovery",
fallback_response: "Development workflow encountered an error. Reverting to last stable checkpoint.",
notify_on_error: true
}
}
multi_agent_system AutonomousDevTeam {
agents: [
ProductManagerAgent,
ArchitectAgent,
DeveloperAgent,
QAEngineerAgent,
CodeReviewerAgent,
DocumentationAgent,
DevOpsAgent
],
orchestration: {
entry_point: ProductManagerAgent,
pattern: "agile_sprint",
routing: {
ProductManagerAgent -> ArchitectAgent: when lambda x: "requirements_defined" in str(x).lower(),
ArchitectAgent -> DeveloperAgent: when lambda x: "design_approved" in str(x).lower(),
DeveloperAgent -> QAEngineerAgent: when lambda x: "implementation_complete" in str(x).lower(),
DeveloperAgent -> DocumentationAgent: when lambda x: "implementation_complete" in str(x).lower(),
QAEngineerAgent -> CodeReviewerAgent: when lambda x: "tests_pass" in str(x).lower(),
QAEngineerAgent -> DeveloperAgent: when lambda x: "tests_fail" in str(x).lower(),
CodeReviewerAgent -> DevOpsAgent: when lambda x: "approved" in str(x).lower(),
CodeReviewerAgent -> DeveloperAgent: when lambda x: "revisions_needed" in str(x).lower()
},
parallel_execution: true,
timeout: 3600.0
},
communication: "mcp",
shared_memory: {
type: "versioned",
capacity: 100000,
persistence: true
},
coordination: {
strategy: "scrum_based",
consensus_threshold: 0.75,
voting_mechanism: "role_weighted"
},
load_balancing: {
strategy: "skill_based",
max_queue_size: 50,
timeout: 60.0
}
}
settings {
gpu: "auto",
execution_mode: "async",
max_concurrent: 12,
retry_policy: {
max_retries: 4,
backoff: "exponential",
initial_delay: 1.5,
max_delay: 45.0
},
ui: "web",
port: 9000,
logging: {
level: "info",
output: "./logs/dev_team.log",
format: "structured_json"
},
metrics: {
enabled: true,
export_interval: 45,
export_format: "prometheus"
},
security: {
authentication: true,
rate_limiting: true,
allowed_origins: ["https://dev.company.com", "https://ci.company.com"]
}
}
This autonomous development team demonstrates role-based agent specialization mimicking real software teams, iterative development workflow with feedback loops, parallel execution of independent tasks like testing and documentation, code quality enforcement through automated review, integration with development tools and version control, comprehensive planning and reflection capabilities, checkpoint-based error recovery, skill-based load balancing, and production-ready configuration for enterprise deployment.
These three complex examples showcase the full power and flexibility of LLMDSL for building sophisticated multi-agent LLM applications across different domains. The formal EBNF specification provides a complete reference for the language syntax, enabling consistent implementation and future extensions.
