128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
|
"""Symbolic Pipeline Abstraction for XIC.
|
||
|
|
|
||
|
|
Provides a structured, glyph-aware pipeline for symbolic cognition execution.
|
||
|
|
Routes prompts through the LAIN 8-lane cognition kernel with explicit step tracking.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class SymbolicStep:
|
||
|
|
"""A single step in the symbolic pipeline execution."""
|
||
|
|
name: str
|
||
|
|
kind: str # "prompt", "glyph_call", "fused_symbol"
|
||
|
|
payload: Any
|
||
|
|
context: Dict[str, Any] = field(default_factory=dict)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class SymbolicPipelineResult:
|
||
|
|
"""Result of a symbolic pipeline execution."""
|
||
|
|
steps: List[SymbolicStep]
|
||
|
|
output_text: str
|
||
|
|
fused_symbol: Optional[Dict[str, Any]] = None
|
||
|
|
|
||
|
|
|
||
|
|
def run_symbolic_pipeline(
|
||
|
|
prompt: str,
|
||
|
|
context: Optional[Dict[str, Any]] = None,
|
||
|
|
glyph_id: Optional[str] = None,
|
||
|
|
) -> SymbolicPipelineResult:
|
||
|
|
"""
|
||
|
|
High-level symbolic pipeline entrypoint for XIC.
|
||
|
|
|
||
|
|
Accepts a prompt and optional symbolic/glyph context, routes through
|
||
|
|
the LAIN 8-lane cognition kernel via CognitiveKernel.execute_symbolic(),
|
||
|
|
and returns a structured SymbolicPipelineResult with execution steps,
|
||
|
|
final output text, and fused symbolic representation.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
prompt: User or system prompt text.
|
||
|
|
context: Optional dict of symbolic/cognitive context metadata.
|
||
|
|
glyph_id: Optional glyph identifier for glyph-aware cognition.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
SymbolicPipelineResult with:
|
||
|
|
- steps: List of SymbolicStep objects tracking execution flow.
|
||
|
|
- output_text: Final text result from cognition layer.
|
||
|
|
- fused_symbol: Fused symbolic representation (if produced by LAIN).
|
||
|
|
"""
|
||
|
|
from gx_compiler.compressor import GXCompressor
|
||
|
|
from .cognitive_kernel import get_kernel
|
||
|
|
|
||
|
|
steps: List[SymbolicStep] = []
|
||
|
|
kernel = get_kernel()
|
||
|
|
prompt_bytes = prompt.encode("utf-8")
|
||
|
|
|
||
|
|
# Step 1: Initial prompt
|
||
|
|
steps.append(SymbolicStep(
|
||
|
|
name="initial_prompt",
|
||
|
|
kind="prompt",
|
||
|
|
payload=prompt,
|
||
|
|
context=dict(context or {})
|
||
|
|
))
|
||
|
|
|
||
|
|
# Step 2: Prepare context for glyph-aware processing
|
||
|
|
exec_context = dict(context or {})
|
||
|
|
if glyph_id:
|
||
|
|
exec_context["glyph_id"] = glyph_id
|
||
|
|
steps.append(SymbolicStep(
|
||
|
|
name=f"glyph:{glyph_id}",
|
||
|
|
kind="glyph_call",
|
||
|
|
payload=prompt,
|
||
|
|
context=exec_context
|
||
|
|
))
|
||
|
|
|
||
|
|
# Step 3: Compress prompt and build manifest
|
||
|
|
try:
|
||
|
|
payload = GXCompressor.compress(prompt)
|
||
|
|
except Exception as e:
|
||
|
|
return SymbolicPipelineResult(
|
||
|
|
steps=steps,
|
||
|
|
output_text=f"[Pipeline Error] Compression failed: {e}",
|
||
|
|
fused_symbol=None
|
||
|
|
)
|
||
|
|
|
||
|
|
manifest = {
|
||
|
|
"source_file": "<symbolic>",
|
||
|
|
"source_type": "symbolic",
|
||
|
|
"version": "1.0.0",
|
||
|
|
"contributor": "XIC-symbolic",
|
||
|
|
"segments": [{"id": "seg_0", "start": 0, "end": 1,
|
||
|
|
"start_byte": 0, "end_byte": len(prompt_bytes)}],
|
||
|
|
}
|
||
|
|
segments = [{"id": "seg_0", "start": 0, "end": 1,
|
||
|
|
"start_byte": 0, "end_byte": len(prompt_bytes)}]
|
||
|
|
|
||
|
|
# Step 4: Execute through LAIN cognition pipeline
|
||
|
|
result = kernel.execute_symbolic(
|
||
|
|
manifest=manifest,
|
||
|
|
segments=segments,
|
||
|
|
payload=payload,
|
||
|
|
mode="symbolic",
|
||
|
|
context=exec_context,
|
||
|
|
)
|
||
|
|
|
||
|
|
# Step 5: Extract results
|
||
|
|
fused_symbol = result.get("fused_symbol")
|
||
|
|
output_text = result.get("output_text") or (
|
||
|
|
fused_symbol.get("summary") if fused_symbol else prompt
|
||
|
|
)
|
||
|
|
|
||
|
|
# Step 6: Record fusion step if fused_symbol present
|
||
|
|
if fused_symbol:
|
||
|
|
steps.append(SymbolicStep(
|
||
|
|
name="fusion",
|
||
|
|
kind="fused_symbol",
|
||
|
|
payload=fused_symbol,
|
||
|
|
context={}
|
||
|
|
))
|
||
|
|
|
||
|
|
return SymbolicPipelineResult(
|
||
|
|
steps=steps,
|
||
|
|
output_text=output_text,
|
||
|
|
fused_symbol=fused_symbol
|
||
|
|
)
|