Data Flow and Processing Pipelines¶

# Document content
Document = {
    "id": str,
    "content": str,
    "title": str,
    "metadata": dict,
    "embedding": List[float],  # Vector representation
    "chunks": List[str]        # Processed chunks
}

# Memory object
Memory = {
    "id": str,
    "project_id": str,
    "memory_type": str,        # decision/preference/experience/etc.
    "title": str,
    "content": str,
    "importance": float,
    "tags": List[str]
}

# Code symbol
Symbol = {
    "id": str,
    "name": str,
    "type": str,               # function/class/variable
    "file_path": str,
    "line": int,
    "relationships": List[dict]
}

# Task result
Task = {
    "task_id": str,
    "status": str,             # pending/processing/success/failed
    "progress": float,
    "result": dict,
    "error": Optional[str]
}

# Request
{
    "content": "This is the document content...",
    "title": "My Document",
    "metadata": {"author": "John", "topic": "AI"}
}

# Size check
content_size = len(content.encode('utf-8'))
if content_size > settings.max_document_size:
    raise HTTPException(413, "Document too large")

# Content check
if not content.strip():
    raise HTTPException(400, "Empty content")

if content_size < 10 * 1024:  # < 10KB
    # Direct synchronous processing
    result = await knowledge_service.add_document(content, title, metadata)
    return {"success": True, "document_id": result["document_id"]}

elif content_size < 50 * 1024:  # 10-50KB
    # Background task processing
    task_id = await submit_document_processing_task(content, title, metadata)
    return {"task_id": task_id, "message": "Processing in background"}

else:  # > 50KB
    # Reject - too large for direct processing
    return {"error": "Document too large, use file upload or directory processing"}

from llama_index.core.node_parser import SimpleNodeParser

parser = SimpleNodeParser.from_defaults(
    chunk_size=512,        # Characters per chunk
    chunk_overlap=50       # Overlap between chunks
)

# Create chunks
chunks = parser.get_nodes_from_documents([document])

# For each chunk
embedding = await embed_model.get_text_embedding(chunk.text)
# embedding: List[float] with length = vector_dimension (e.g., 1536)

from llama_index.embeddings.ollama import OllamaEmbedding

embed_model = OllamaEmbedding(
    model_name="nomic-embed-text",
    base_url="http://localhost:11434"
)
# Dimension: 768

from llama_index.embeddings.openai import OpenAIEmbedding

embed_model = OpenAIEmbedding(
    model="text-embedding-ada-002"
)
# Dimension: 1536

# Configured in Knowledge Service
from llama_index.core import KnowledgeGraphIndex

# Extract entities and relationships
index = KnowledgeGraphIndex.from_documents(
    documents=[document],
    max_triplets_per_chunk=10
)

# Example triplets (subject, predicate, object)
[
    ("Python", "is_a", "Programming Language"),
    ("Python", "supports", "Object-Oriented Programming"),
    ("Django", "is_built_with", "Python")
]

{
    "success": true,
    "document_id": "doc_abc123",
    "chunks_created": 4,
    "entities_extracted": 12,
    "processing_time": 2.5  # seconds
}

try:
    result = await knowledge_service.add_document(content, title)
except EmbeddingError as e:
    # LLM/embedding provider unavailable
    return {"error": "Embedding service unavailable", "retry": True}

except Neo4jError as e:
    # Database connection issue
    return {"error": "Database error", "retry": True}

except ValidationError as e:
    # Invalid input
    return {"error": str(e), "retry": False}

except TimeoutError as e:
    # Operation timeout
    return {"error": "Processing timeout", "retry": True}

config = service._resolve_pipeline_config(
    mode=payload.get("mode", "hybrid"),
    use_graph=payload.get("use_graph"),
    use_vector=payload.get("use_vector"),
    use_tools=payload.get("use_tools", False),
    top_k=payload.get("top_k"),
    graph_depth=payload.get("graph_depth"),
)

graph_nodes = KnowledgeGraphRAGRetriever(
    storage_context=storage_context,
    llm=Settings.llm,
    graph_traversal_depth=config.graph_depth,
).retrieve(QueryBundle(question))

vector_nodes = VectorIndexRetriever(
    vector_index,
    similarity_top_k=config.top_k,
).retrieve(QueryBundle(question))

response = response_synthesizer.synthesize(QueryBundle(question), merged_nodes)

{
    "project_id": "myapp",
    "memory_type": "decision",
    "title": "Use PostgreSQL for main database",
    "content": "Decided to use PostgreSQL instead of MySQL",
    "reason": "Need advanced JSON support and better performance",
    "importance": 0.9,
    "tags": ["database", "architecture", "backend"]
}

# 1. Validate
if not 0 <= importance <= 1:
    raise ValueError("Importance must be 0-1")

# 2. Generate ID
memory_id = f"mem_{uuid.uuid4().hex[:12]}"

# 3. Store in Neo4j
await memory_store.add_memory(
    project_id=project_id,
    memory_type=memory_type,
    title=title,
    content=content,
    reason=reason,
    importance=importance,
    tags=tags
)

prompt = f"""Analyze this conversation and extract important project knowledge.

Conversation:
{format_conversation(conversation)}

Extract:
1. Technical decisions and rationale
2. Preferences and conventions
3. Problems encountered and solutions
4. Important insights or learnings

For each item, provide:
- Type (decision/preference/experience/note)
- Title (short summary)
- Content (detailed description)
- Confidence score (0-1)

Return as JSON list.
"""

[
    {
        "type": "decision",
        "title": "Use Redis for caching",
        "content": "Decided to add Redis caching layer for API responses",
        "reason": "Reduce database load and improve response times",
        "confidence": 0.85,
        "importance": 0.8,
        "tags": ["caching", "performance", "redis"]
    }
]

if auto_save:
    for memory in extracted_memories:
        if memory['confidence'] > 0.7:  # High confidence threshold
            await memory_store.add_memory(**memory)

{
    "project_id": "myapp",
    "query": "database decisions",
    "memory_type": "decision",
    "tags": ["database"],
    "min_importance": 0.7,
    "limit": 10
}

def rank_memories(results: List[dict]) -> List[dict]:
    """Rank by search score and importance"""
    for result in results:
        result['rank_score'] = (
            result['search_score'] * 0.6 +
            result['importance'] * 0.4
        )
    return sorted(results, key=lambda x: x['rank_score'], reverse=True)

{
    "repo_path": "/path/to/repository",
    "repo_name": "myapp",
    "file_patterns": ["*.py", "*.js"]  # Optional filters
}

def scan_repository(repo_path: str) -> List[str]:
    """Scan directory and collect files"""
    files = []

    for root, dirs, filenames in os.walk(repo_path):
        # Skip common ignored directories
        dirs[:] = [d for d in dirs if d not in {
            '.git', 'node_modules', '__pycache__', '.venv'
        }]

        for filename in filenames:
            file_path = os.path.join(root, filename)
            if should_process_file(file_path):
                files.append(file_path)

    return files

LANGUAGE_EXTENSIONS = {
    '.py': 'python',
    '.js': 'javascript',
    '.ts': 'typescript',
    '.java': 'java',
    '.go': 'go',
    '.cpp': 'cpp',
    '.sql': 'sql'
}

language = LANGUAGE_EXTENSIONS.get(file_extension)

import ast

def parse_python_file(file_path: str) -> ParseResult:
    """Parse Python file using AST"""
    with open(file_path, 'r') as f:
        source = f.read()

    tree = ast.parse(source, filename=file_path)

    symbols = []
    relationships = []

    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef):
            # Extract function
            symbols.append({
                'type': 'function',
                'name': node.name,
                'line': node.lineno,
                'params': [arg.arg for arg in node.args.args],
                'decorators': [d.id for d in node.decorator_list if isinstance(d, ast.Name)]
            })

            # Extract function calls
            for child in ast.walk(node):
                if isinstance(child, ast.Call):
                    if isinstance(child.func, ast.Name):
                        relationships.append({
                            'from': node.name,
                            'to': child.func.id,
                            'type': 'CALLS'
                        })

        elif isinstance(node, ast.ClassDef):
            # Extract class
            symbols.append({
                'type': 'class',
                'name': node.name,
                'line': node.lineno,
                'bases': [base.id for base in node.bases if isinstance(base, ast.Name)],
                'methods': [m.name for m in node.body if isinstance(m, ast.FunctionDef)]
            })

            # Extract inheritance
            for base in node.bases:
                if isinstance(base, ast.Name):
                    relationships.append({
                        'from': node.name,
                        'to': base.id,
                        'type': 'INHERITS'
                    })

    return ParseResult(symbols=symbols, relationships=relationships)

# Using esprima or similar parser
import esprima

def parse_javascript_file(file_path: str) -> ParseResult:
    """Parse JavaScript file"""
    with open(file_path, 'r') as f:
        source = f.read()

    tree = esprima.parseScript(source, {'loc': True})

    # Extract functions, classes, imports
    # Similar to Python parsing

async def ingest_repository_with_progress(repo_path: str, task_id: str):
    """Ingest with progress updates"""
    files = scan_repository(repo_path)
    total_files = len(files)

    for i, file_path in enumerate(files):
        # Parse file
        result = parse_file(file_path)

        # Store in graph
        await store_file_in_graph(result)

        # Update progress
        progress = (i + 1) / total_files
        await task_queue.update_progress(
            task_id,
            progress,
            f"Processed {i+1}/{total_files} files"
        )

async def submit_document_processing_task(
    content: str,
    title: str,
    metadata: dict
) -> str:
    """Submit document processing as background task"""
    # 1. Create task data
    task_data = {
        'type': 'document_processing',
        'content': content,
        'title': title,
        'metadata': metadata
    }

    # 2. Submit to queue
    task_id = await task_queue.submit_task(
        task_func=process_document_task,
        task_type='document_processing',
        task_data=task_data
    )

    return task_id

# Store in SQLite
await task_storage.store_task(
    task_id=task_id,
    status=TaskStatus.PENDING,
    task_type='document_processing',
    metadata=task_data
)

async def _process_pending_tasks(self):
    """Background worker continuously processes tasks"""
    while True:
        try:
            # 1. Get pending tasks (with lock)
            pending = await self._storage.get_pending_tasks(limit=10)

            # 2. Process each task (respecting concurrency limit)
            for task in pending:
                if len(self.running_tasks) < self.max_concurrent_tasks:
                    # Start task processing
                    asyncio.create_task(self._execute_task(task))

            # 3. Wait before next poll
            await asyncio.sleep(1)

        except asyncio.CancelledError:
            logger.info("Worker shutting down")
            break
        except Exception as e:
            logger.error(f"Worker error: {e}")
            await asyncio.sleep(5)  # Back off on error

async def _execute_task(self, task: TaskResult):
    """Execute single task"""
    async with self.task_semaphore:  # Limit concurrency
        try:
            # 1. Update status
            task.status = TaskStatus.PROCESSING
            task.started_at = datetime.now()
            await self._storage.update_task_status(task.task_id, task.status)

            # 2. Get appropriate processor
            processor = processor_registry.get_processor(task.metadata['type'])

            # 3. Execute with progress updates
            result = await processor.process(task)

            # 4. Mark success
            task.status = TaskStatus.SUCCESS
            task.result = result
            task.completed_at = datetime.now()
            task.progress = 1.0

        except Exception as e:
            # Mark failed
            task.status = TaskStatus.FAILED
            task.error = str(e)
            task.completed_at = datetime.now()
            logger.error(f"Task {task.task_id} failed: {e}")

        finally:
            # Save final state
            await self._storage.update_task(task)
            self.running_tasks.pop(task.task_id, None)

@app.get("/api/v1/sse/task/{task_id}")
async def stream_task_progress(task_id: str):
    """Stream task progress via Server-Sent Events"""
    async def event_generator():
        while True:
            # Get current task status
            task = task_queue.get_task(task_id)

            if task:
                # Send update
                yield {
                    "event": "progress",
                    "data": json.dumps({
                        "task_id": task.task_id,
                        "status": task.status.value,
                        "progress": task.progress,
                        "message": task.message
                    })
                }

                # Stop if terminal state
                if task.status in {TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.CANCELLED}:
                    break

            await asyncio.sleep(0.5)  # Poll interval

    return EventSourceResponse(event_generator())

@app.websocket("/api/v1/ws/task/{task_id}")
async def task_websocket(websocket: WebSocket, task_id: str):
    """WebSocket for real-time task updates"""
    await websocket.accept()

    try:
        while True:
            task = task_queue.get_task(task_id)

            if task:
                # Send update
                await websocket.send_json({
                    "task_id": task.task_id,
                    "status": task.status.value,
                    "progress": task.progress,
                    "message": task.message
                })

                # Stop if done
                if task.status in {TaskStatus.SUCCESS, TaskStatus.FAILED}:
                    break

            await asyncio.sleep(0.5)

    except WebSocketDisconnect:
        logger.info(f"Client disconnected from task {task_id}")

# Inside task processor
await task_queue.update_progress(
    task_id=task_id,
    progress=0.5,
    message="Processed 50% of documents"
)

class TaskQueue:
    async def update_progress(self, task_id: str, progress: float, message: str):
        """Update progress and notify listeners"""
        # Update task
        task = self.tasks[task_id]
        task.progress = progress
        task.message = message

        # Store in database
        await self._storage.update_task(task)

        # Notify SSE listeners
        await self._notify_sse_listeners(task_id, task)

        # Notify WebSocket listeners
        await self._notify_ws_listeners(task_id, task)

async def _notify_sse_listeners(self, task_id: str, task: TaskResult):
    """Send update to SSE clients"""
    if task_id in self.sse_listeners:
        event_data = {
            "task_id": task_id,
            "status": task.status.value,
            "progress": task.progress,
            "message": task.message
        }

        for queue in self.sse_listeners[task_id]:
            await queue.put(event_data)

# Client (Claude Desktop) calls tool
{
    "method": "tools/call",
    "params": {
        "name": "query_knowledge",
        "arguments": {
            "question": "How does authentication work?",
            "top_k": 5
        }
    }
}

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> Sequence[TextContent]:
    """Route tool call to appropriate handler"""
    # Get handler for tool
    handler = TOOL_HANDLERS.get(name)

    if not handler:
        raise ValueError(f"Unknown tool: {name}")

    # Execute handler
    result = await handler(arguments)

    # Format response
    return [TextContent(
        type="text",
        text=format_result(result)
    )]

async def handle_query_knowledge(arguments: dict) -> dict:
    """Handle knowledge query"""
    # Validate arguments
    question = arguments.get("question")
    if not question:
        return {"success": False, "error": "Question required"}

    top_k = arguments.get("top_k", 5)

    # Call service
    result = await neo4j_knowledge_service.query(
        question=question,
        top_k=top_k
    )

    return result

def format_result(result: dict) -> str:
    """Format result for MCP client"""
    if not result.get("success"):
        return f"Error: {result.get('error', 'Unknown error')}"

    # Format answer with sources
    answer = result.get("answer", "")
    sources = result.get("sources", [])

    formatted = f"Answer:\n{answer}\n\n"

    if sources:
        formatted += "Sources:\n"
        for i, source in enumerate(sources, 1):
            formatted += f"{i}. {source.get('title')} (relevance: {source.get('relevance', 0):.2f})\n"

    return formatted

async def batch_process_files(file_paths: List[str]) -> dict:
    """Process multiple files in parallel"""
    # Create tasks
    tasks = [
        asyncio.create_task(process_file(path))
        for path in file_paths
    ]

    # Wait for all tasks
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Aggregate results
    successful = [r for r in results if not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]

    return {
        "total": len(file_paths),
        "successful": len(successful),
        "failed": len(failed),
        "results": successful
    }

# Instead of one at a time
for text in texts:
    embedding = await embed_model.get_text_embedding(text)

# Batch process
embeddings = await embed_model.get_text_embeddings(texts)

# Cache by content hash
content_hash = hashlib.sha256(text.encode()).hexdigest()
if content_hash in embedding_cache:
    return embedding_cache[content_hash]

# Adjust based on resources
task_queue = TaskQueue(max_concurrent_tasks=5)  # CPU-bound
task_queue = TaskQueue(max_concurrent_tasks=20)  # I/O-bound

# High priority tasks first
async def get_pending_tasks(self, priority_order: bool = True):
    if priority_order:
        query = "ORDER BY priority DESC, created_at ASC"
    else:
        query = "ORDER BY created_at ASC"

async def retry_on_error(func, max_retries=3, backoff=2):
    """Retry with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return await func()
        except TransientError as e:
            if attempt == max_retries - 1:
                raise
            wait_time = backoff ** attempt
            await asyncio.sleep(wait_time)

class CircuitBreaker:
    """Prevent cascading failures"""
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    async def call(self, func):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise CircuitBreakerOpen()

        try:
            result = await func()
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
            raise