logos-router-reasoning-mesh

# Logos Router Reasoning Mesh > Skill by [ara.so](https://ara.so) — MCP Skills collection. Logos Router is a distributed semantic reasoning gateway that fragments AI reasoning across local nodes with zero-drift consensus guarantees. Unlike traditional single-model inference pipelines, it distributes subproblems across a mesh of reasoning units that coordinate through Strict Write Discipline (SWD) protocol—ensuring every output is cross-verified before committing. ## Installation ### Prerequisites - Python 3.11 or later - At least one local inference engine (VLLM, Ollama, llama.cpp) - Network access between mesh nodes (localhost sufficient for single-machine) ### Basic Setup ```bash # Clone the repository git clone https://github.com/rak7777/mythic-mcp-proxy.git cd mythic-mcp-proxy # Install dependencies pip install -r requirements.txt # Verify installation python -m logos --version ``` ## Core Architecture Logos Router operates on three layers: - **Grail Layer**: Content-addressed persistence store for immutable reasoning DAG - **Chiron Layer**: Semantic routing and node assignment - **Orpheus Layer**: Multi-fragment synthesis and conflict resolution ## Configuration ### Basic Mesh Configuration Create a YAML configuration file (e.g., `mesh_config.yaml`): ```yaml mesh: name: local-thought-lattice consensus_threshold: 0.95 # Confidence required before write commit nodes: - type: local engine: vllm model: opus-mini-4.8 max_thinking_depth: 7 host: localhost port: 8001 - type: local engine: ollama model: llama3.1-70b max_thinking_depth: 5 host: localhost port: 11434 languages: - en - es - zh - ar - fr - de strict_write_discipline: enabled: true min_peer_verifications: 2 escalation_depth_threshold: 4 semantic_caching: enabled: true cache_ttl_hours: 24 max_cache_size_mb: 2048 reasoning: default_depth: adaptive max_parallel_threads: 4 graceful_degradation: true ``` ### Initialize Router with Config ```python from logos import Router, RouterConfig # Load configuration config = RouterConfig.from_yaml("mesh_config.yaml") # Initialize router router = Router(config=config) # Verify mesh connectivity status = router.health_check() print(f"Active nodes: {status.active_nodes}") print(f"Consensus available: {status.consensus_ready}") ``` ## Single-Query Reasoning ### Basic Query Execution ```python from logos import Router router = Router(config="mesh_config.yaml") # Execute single query with adaptive depth response = router.query( prompt="Explain the halting problem and its implications for AI safety", depth="adaptive", language="en" ) print(response.text) print(f"Reasoning depth used: {response.depth}") print(f"Nodes consulted: {response.nodes_involved}") print(f"Consensus score: {response.consensus_score}") print(f"Verification steps: {len(response.audit_trail)}") ``` ### Explicit Depth Control ```python # Force shallow reasoning (faster, less thorough) quick_response = router.query( prompt="What is photosynthesis?", depth=2, language="en" ) # Force deep reasoning (slower, more thorough) deep_response = router.query( prompt="Design a distributed consensus algorithm for Byzantine fault tolerance", depth=7, language="en", require_consensus=True ) ``` ### Multilingual Reasoning ```python # Query in Spanish, reason in universal IR, respond in Spanish response = router.query( prompt="¿Cuál es la diferencia entre aprendizaje supervisado y no supervisado?", depth="adaptive", language="es" ) # Query in one language, respond in another response = router.query( prompt="Explain quantum entanglement", depth=5, language="en", response_language="zh" # Respond in Mandarin ) ``` ## Multi-Step Workflows ### Creating Reasoning Workflows ```python from logos import Router, Workflow router = Router(config="mesh_config.yaml") # Create multi-step workflow workflow = router.create_workflow( name="research-paper-analysis", input_file="paper.pdf" ) # Define workflow steps workflow.add_step( name="extract_claims", operation="extract", params={"claim_types": ["theorem", "hypothesis", "conclusion"]} ) workflow.add_step( name="verify_claims", operation="verify", params={"consensus": True, "min_confidence": 0.95} ) workflow.add_step( name="generate_summary", operation="synthesize", params={"style": "academic", "max_length": 500} ) # Execute workflow with automatic state management result = workflow.execute() print(f"Claims extracted: {len(result.steps['extract_claims'].output)}") print(f"Claims verified: {result.steps['verify_claims'].verified_count}") print(f"Summary: {result.steps['generate_summary'].text}") ``` ### Workflow with Error Handling ```python workflow = router.create_workflow("code-review") try: workflow.add_step("analyze_complexity", operation="analyze") workflow.add_step("detect_vulnerabilities", operation="security_scan") workflow.add_step("suggest_improvements", operation="generate") result = workflow.execute( timeout_seconds=300, graceful_degradation=True ) # Access individual step results for step_name, step_result in result.steps.items(): print(f"{step_name}: {step_result.status}") if step_result.degraded: print(f" Warning: Reduced depth to {step_result.actual_depth}") except workflow.WorkflowError as e: print(f"Workflow failed at step: {e.failed_step}") print(f"Partial results available: {e.partial_results}") ``` ## Strict Write Discipline Protocol ### Understanding SWD Flow The Strict Write Discipline protocol ensures zero-drift reasoning: 1. **Proposal**: Node generates candidate reasoning step 2. **Broadcast**: Candidate sent to peer nodes 3. **Verification**: Peers evaluate consistency and accuracy 4. **Consensus**: If confidence > threshold, commit 5. **Commit**: Write to Grail layer 6. **Escalation**: If consensus fails, escalate to higher-depth node ### Monitoring SWD ```python from logos import Router router = Router(config="mesh_config.yaml") # Enable detailed SWD logging router.configure_logging( level="DEBUG", swd_trace=True ) response = router.query( prompt="Prove that the set of real numbers is uncountable", depth="adaptive" ) # Inspect SWD trace for step in response.swd_trace: print(f"Step {step.index}:") print(f" Proposal: {step.proposal_summary}") print(f" Verifiers: {step.verifier_nodes}") print(f" Confidence scores: {step.confidence_scores}") print(f" Consensus: {step.consensus_reached}") if step.escalated: print(f" Escalated to depth: {step.escalation_depth}") ``` ### Custom Consensus Thresholds ```python # Override consensus threshold for specific query response = router.query( prompt="Design a safety-critical flight control algorithm", depth=7, consensus_threshold=0.99, # Higher threshold for safety-critical min_peer_verifications=3 ) ``` ## Semantic Caching ### Enable and Configure Caching ```python from logos import Router, CacheConfig cache_config = CacheConfig( enabled=True, ttl_hours=24, max_size_mb=2048, similarity_threshold=0.85 # Semantic similarity for cache hits ) router = Router( config="mesh_config.yaml", cache_config=cache_config ) # First query (cache miss) response1 = router.query( prompt="What are the benefits of functional programming?", depth="adaptive" ) print(f"Cache hit: {response1.cache_hit}") # False # Similar query (cache hit) response2 = router.query( prompt="Explain advantages of functional programming paradigm", depth="adaptive" ) print(f"Cache hit: {response2.cache_hit}") # True print(f"Latency reduction: {response2.cache_stats.time_saved_ms}ms") ``` ### Cache Management ```python # Clear cache for specific semantic domain router.cache.clear_domain("programming_concepts") # Inspect cache statistics stats = router.cache.statistics() print(f"Hit rate: {stats.hit_rate:.2%}") print(f"Average similarity score: {stats.avg_similarity}") print(f"Memory usage: {stats.memory_mb}MB") # Manually warm cache with common queries common_queries = [ "What is machine learning?", "Explain neural networks", "Define supervised learning" ] for query in common_queries: router.query(query, depth=3) ``` ## Audit Trails and Reasoning Inspection ### Accessing Causal Audit Trails ```python response = router.query( prompt="Should we adopt microservices architecture?", depth="adaptive" ) # Inspect complete reasoning chain audit_trail = response.audit_trail for i, step in enumerate(audit_trail.steps): print(f"\n--- Step {i+1} ---") print(f"Node: {step.node_id}") print(f"Reasoning: {step.reasoning_fragment}") print(f"Confidence: {step.confidence}") print(f"Dependencies: {step.dependencies}") print(f"Timestamp: {step.timestamp}") # Export audit trail for external verification audit_trail.export_to_file("reasoning_trace.json") ``` ### Visualizing Reasoning DAG ```python from logos.visualization import ReasoningGraph # Generate visual representation of reasoning flow graph = ReasoningGraph(response.audit_trail) graph.render( output_file="reasoning_dag.svg", format="svg", show_confidence_scores=True, show_node_assignments=True ) ``` ## Advanced Routing Strategies ### Custom Node Selection ```python from logos import Router, NodeSelector # Define custom node selection strategy class CapabilityBasedSelector(NodeSelector): def select_nodes(self, query, available_nodes): """Select nodes based on specific capabilities""" if "mathematical" in query.domain: return [n for n in available_nodes if n.has_capability("symbolic_math")] elif "code" in query.domain: return [n for n in available_nodes if n.has_capability("code_analysis")] return available_nodes[:2] # Default: first two nodes router = Router( config="mesh_config.yaml", node_selector=CapabilityBasedSelector() ) ``` ### Load Balancing ```python # Configure load-aware routing router.configure_routing( strategy="load_balanced", max_node_queue_depth=10, rebalance_interval_seconds=30 ) # Monitor node load for node in router.active_nodes(): load = router.get_node_load(node.id) print(f"Node {node.id}: {load.queue_depth} queries, {load.avg_latency_ms}ms avg") ``` ## REST API Integration ### Starting REST Server ```python from logos import Router from logos.server import RESTServer router = Router(config="mesh_config.yaml") # Start REST API server server = RESTServer( router=router, host="0.0.0.0", port=8080, enable_cors=True ) server.start() ``` ### Using REST API ```bash # Query via REST API curl -X POST http://localhost:8080/query \ -H "Content-Type: application/json" \ -d '{ "prompt": "Explain the CAP theorem", "depth": "adaptive", "language": "en" }' # Get mesh status curl http://localhost:8080/status # Retrieve audit trail curl http://localhost:8080/audit/query_id_12345 ``` ### Python REST Client ```python import requests response = requests.post( "http://localhost:8080/query", json={ "prompt": "Compare ACID vs BASE database properties", "depth": 5, "require_consensus": True } ) result = response.json() print(result["text"]) print(f"Consensus score: {result['consensus_score']}") ``` ## WebSocket Streaming ### Real-Time Reasoning Updates ```python from logos import Router from logos.streaming import WebSocketStreamer router = Router(config="mesh_config.yaml") # Start WebSocket server for streaming streamer = WebSocketStreamer(router, port=8081) streamer.start() # Client-side streaming consumer import asyncio import websockets