Merge pull request #258 from nk-ag/niki/dev/statemanagerfrontend

State manager frontend with demo scenarios

Author: nk-ag

.coverage

@@ -0,0 +1,13 @@
+# Environment files (keep .env.example tracked)  
+.env  
+.env.*.local  
+.env.local  
+*.env  
+
+# Virtual environments  
+**/.venv/  
+**/venv/  
+
+# Python caches  
+**/__pycache__/  
+*.py[cod]

api-server/tests/integration/test_api_server_integration.py

@@ -0,0 +1 @@
+3.12

exosphere-runtimes/.gitignore

@@ -0,0 +1,6 @@
+def main():
+    print("Hello from exosphere-runtimes!")
+
+
+if __name__ == "__main__":
+    main()

exosphere-runtimes/.python-version

@@ -0,0 +1,103 @@
+# Parse List of Docs Runtime
+
+A simplified document processing workflow for parsing multiple documents from S3, generating Q&A pairs, and uploading reports back to S3.
+
+## Workflow Overview
+
+This runtime implements a streamlined document processing pipeline:
+
+1. **ListS3FilesNode** - Lists PDF files from an S3 bucket
+2. **ParseSinglePDFNode** - Parses a single PDF file from S3
+3. **ChunkDocumentNode** - Splits document content into manageable chunks
+4. **GenerateQuestionNode** - Generates questions from document chunks
+5. **GenerateAnswerNode** - Generates answers to the questions
+6. **VerifyAnswerNode** - Verifies the accuracy of generated answers
+7. **CreateReportNode** - Creates final reports from verification results
+8. **UploadToS3Node** - Uploads reports back to S3
+
+## Setup
+
+1. Install dependencies:
+```bash
+pip install -r requirements.txt
+```
+
+2. Set up environment variables in `.env`:
+```
+EXOSPHERE_STATE_MANAGER_URI=http://localhost:8000
+EXOSPHERE_API_KEY=your-api-key
+```
+
+3. Register the runtime:
+```bash
+python main.py
+```
+
+## Usage
+
+Run the graph template to register and trigger the workflow:
+
+```bash
+python graph_template.py
+```
+
+This will:
+- Register the graph template with the state manager
+- Trigger the workflow with sample S3 bucket and prefix
+- Process the first PDF file in the list
+- Generate Q&A for the first chunk
+- Create and upload a report
+
+## Configuration
+
+Update the following in `graph_template.py`:
+- API keys and secrets
+- S3 bucket names and prefixes
+- Chunk size and overlap parameters
+
+## Node Details
+
+### ListS3FilesNode
+- **Inputs**: `bucket_name`, `prefix`
+- **Outputs**: `file_list`, `bucket_name`
+- **Secrets**: AWS credentials
+
+### ParseSinglePDFNode
+- **Inputs**: `file_key`, `bucket_name`
+- **Outputs**: `extracted_content`, `file_key`, `document_title`
+- **Secrets**: AWS credentials
+
+### ChunkDocumentNode
+- **Inputs**: `extracted_content`, `file_key`, `chunk_size`, `overlap`
+- **Outputs**: `chunks`, `file_key`, `total_chunks`
+
+### GenerateQuestionNode
+- **Inputs**: `chunk`
+- **Outputs**: `source_chunk`, `generated_question`, `chunk_id`
+- **Secrets**: OpenAI API key
+
+### GenerateAnswerNode
+- **Inputs**: `source_chunk`, `generated_question`, `chunk_id`
+- **Outputs**: `generated_answer`, `question`, `source_chunk`, `chunk_id`
+- **Secrets**: OpenAI API key
+
+### VerifyAnswerNode
+- **Inputs**: `generated_answer`, `question`, `source_chunk`, `chunk_id`
+- **Outputs**: `verification_results`, `chunk_id`, `file_key`
+- **Secrets**: OpenAI API key
+
+### CreateReportNode
+- **Inputs**: `verification_results`, `chunk_id`, `file_key`
+- **Outputs**: `final_report`, `file_key`
+
+### UploadToS3Node
+- **Inputs**: `final_report`, `file_key`, `output_bucket`, `output_prefix`
+- **Outputs**: `upload_status`, `s3_key`, `file_key`
+- **Secrets**: AWS credentials
+
+## Notes
+
+- This is a simplified implementation with simulated S3 operations
+- In production, replace simulated functions with actual AWS SDK calls
+- The workflow processes one file and one chunk at a time
+- For processing multiple files/chunks, you would need to implement loops or parallel processing

exosphere-runtimes/README.md

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""
+Parse List of Docs Graph Template
+
+This file defines the graph template for the simplified document processing workflow.
+It processes a list of documents from S3: list files → parse each PDF → upload each to S3.
+"""
+
+import asyncio
+import aiohttp
+import os
+
+# Graph Template Definition
+PARSE_LIST_OF_DOCS_GRAPH_TEMPLATE = {
+    "secrets": {
+        "openai_api_key": "testkey",
+        "aws_access_key_id": "testkey",
+        "aws_secret_access_key": "testkey",
+        "aws_region": "us-east-1"
+    },
+    "nodes": [
+        {
+            "node_name": "ListS3FilesNode",
+            "namespace": "parse-list-of-docs",
+            "identifier": "list_files",
+            "inputs": {
+                "bucket_name": "initial",
+                "prefix": "initial",
+                "files_only": "true",
+                "recursive": "false"
+            },
+            "next_nodes": ["parse_pdf"]
+        },
+        {
+            "node_name": "ParseSinglePDFNode",
+            "namespace": "parse-list-of-docs",
+            "identifier": "parse_pdf",
+            "inputs": {
+                "bucket_name": "initial",
+                "key": "${{ list_files.outputs.key }}"
+            },
+            "next_nodes": ["upload_to_s3"]
+        },
+
+        {
+            "node_name": "UploadToS3Node",
+            "namespace": "parse-list-of-docs",
+            "identifier": "upload_to_s3",
+            "inputs": {
+                "extracted_content": "${{ parse_pdf.outputs.extracted_content }}",
+                "key": "${{ parse_pdf.outputs.key }}",
+                "output_bucket": "processed-docs-bucket",
+                "output_prefix": "processed-documents/"
+            },
+            "next_nodes": []
+        }
+    ]
+}
+
+async def create_graph_template(graph_name: str):
+    """Create a graph template with simplified document processing nodes"""
+    namespace = "parse-list-of-docs"
+    api_key = os.getenv("STATE_MANAGER_API_KEY")  # TODO: Replace with your actual API key
+    
+    graph_request = {
+        "secrets": {
+            "openai_api_key": "testkey",
+            "aws_access_key_id": "testkey",
+            "aws_secret_access_key": "testkey",
+            "aws_region": "us-east-1"
+        },
+        "nodes": PARSE_LIST_OF_DOCS_GRAPH_TEMPLATE["nodes"]
+    }
+    
+    async with aiohttp.ClientSession() as session:
+        # Create the graph template
+        url = f"http://localhost:8000/v0/namespace/{namespace}/graph/{graph_name}"
+        headers = {"X-API-Key": api_key, "Content-Type": "application/json"}
+        
+        try:
+            async with session.put(url, json=graph_request, headers=headers) as response:
+                if response.status == 201:
+                    print(f"Parse list of docs graph template created successfully: {graph_name}")
+                    return graph_name
+                else:
+                    print(f"Failed to create graph template: {response.status}")
+                    print(await response.text())
+                    return None
+        except Exception as e:
+            print(f"Error creating graph template: {e}")
+            return None
+
+async def trigger_graph_execution(graph_name: str):
+    """Trigger the first node using the new trigger graph API endpoint"""
+    namespace = "parse-list-of-docs"
+    api_key = "niki"  # TODO: Replace with your actual API key
+    
+    # Trigger graph with the first node
+    trigger_request = {
+        "states": [
+            {
+                "identifier": "list_files",
+                "inputs": {
+                    "bucket_name": "my-documents-bucket",
+                    "prefix": "pdfs/"
+                }
+            }
+        ]
+    }
+    
+    async with aiohttp.ClientSession() as session:
+        url = f"http://localhost:8000/v0/namespace/{namespace}/graph/{graph_name}/trigger"
+        headers = {"X-API-Key": api_key, "Content-Type": "application/json"}
+        
+        try:
+            async with session.post(url, json=trigger_request, headers=headers) as response:
+                if response.status == 200:
+                    response_data = await response.json()
+                    print("✅ Parse list of docs graph triggered successfully!")
+                    print(f"   - Graph: {graph_name}")
+                    print(f"   - Run ID: {response_data['run_id']}")
+                    print(f"   - State ID: {response_data['states'][0]['state_id']}")
+                    print(f"   - Node: {response_data['states'][0]['node_name']}")
+                    print(f"   - Status: {response_data['status']}")
+                    return response_data['states'][0]['state_id'], response_data['run_id']
+                else:
+                    print(f"❌ Failed to trigger graph: {response.status}")
+                    print(await response.text())
+                    return None, None
+        except Exception as e:
+            print(f"❌ Error triggering graph: {e}")
+            return None, None
+
+
+async def main():
+    """Main function to create graph and trigger workflow"""
+    print("🚀 Starting Parse List of Docs workflow setup...")
+    
+    # Step 1: Create the graph template
+    print("\n📋 Step 1: Creating parse list of docs graph template...")
+    graph_name = "parse-list-of-docs-workflow"
+    # await create_graph_template(graph_name)
+    
+    # Step 2: Trigger the graph execution
+    print("\n🎯 Step 2: Triggering parse list of docs workflow...")
+    state_id, run_id = await trigger_graph_execution(graph_name)
+    
+    if state_id and run_id:
+        print("\n✅ Parse list of docs workflow initiated successfully!")
+        print("   The workflow will now execute:")
+        print("   1. ListS3FilesNode (list_files) - List PDF files from S3")
+        print("   2. ParseSinglePDFNode (parse_pdf) - Parse each PDF file")
+        print("   3. UploadToS3Node (upload_to_s3) - Upload processed document to S3")
+        print(f"\n   State ID: {state_id}")
+        print(f"   Run ID: {run_id}")
+        print(f"   Graph: {graph_name}")
+    else:
+        print("❌ Failed to trigger parse list of docs workflow.")
+
+if __name__ == "__main__":
+    # Run the async main function
+    asyncio.run(main())

exosphere-runtimes/main.py

@@ -0,0 +1,34 @@
+from dotenv import load_dotenv
+from exospherehost import Runtime
+from nodes.list_s3_files import ListS3FilesNode
+from nodes.parse_single_pdf import ParseSinglePDFNode
+from nodes.chunk_document import ChunkDocumentNode
+from nodes.generate_question import GenerateQuestionNode
+from nodes.generate_answer import GenerateAnswerNode
+from nodes.verify_answer import VerifyAnswerNode
+from nodes.create_report import CreateReportNode
+from nodes.upload_to_s3 import UploadToS3Node
+
+# Load environment variables from .env file
+# EXOSPHERE_STATE_MANAGER_URI is the URI of the state manager
+# EXOSPHERE_API_KEY is the key of the runtime
+load_dotenv()
+
+# Note on node ordering:
+# The order of node classes in the `nodes` list does not define execution sequence.
+# Nodes are registered with the state manager; orchestration and dependencies are handled externally.
+# Nodes are listed in logical processing order for readability only.
+Runtime(
+    name="parse-list-of-docs-runtime",
+    namespace="parse-list-of-docs",
+    nodes=[
+        ListS3FilesNode,
+        ParseSinglePDFNode,
+        ChunkDocumentNode,
+        GenerateQuestionNode,
+        GenerateAnswerNode,
+        VerifyAnswerNode,
+        CreateReportNode,
+        UploadToS3Node
+    ]
+).start()

exosphere-runtimes/parse-list-of-docs/README.md

@@ -0,0 +1,55 @@
+from exospherehost import BaseNode
+from pydantic import BaseModel
+
+
+class ChunkDocumentNode(BaseNode):
+
+    class Inputs(BaseModel):
+        extracted_content: str
+        key: str
+        chunk_size: str = "1000"
+        overlap: str = "200"
+
+    class Outputs(BaseModel):
+        chunk: str
+        
+    class Secrets(BaseModel):
+        pass
+
+    async def execute(self) :
+        """
+        Split document content into chunks for processing.
+        """
+        
+        # Convert string inputs to appropriate types
+        chunk_size = int(self.inputs.chunk_size)
+        overlap = int(self.inputs.overlap)
+        
+        content = self.inputs.extracted_content
+        chunks = []
+        
+        # Simple chunking by character count
+        # In production, you might want more sophisticated chunking
+        start = 0
+        chunk_id = 0
+        
+        while start < len(content):
+            end = min(start + chunk_size, len(content))
+            chunk_text = content[start:end]
+            
+            chunk = {
+                "chunk_id": chunk_id,
+                "content": chunk_text,
+                "start_pos": start,
+                "end_pos": end,
+                "key": self.inputs.key
+            }
+            
+            chunks.append(chunk)
+            chunk_id += 1
+            start = end - overlap if end < len(content) else end
+        
+        return [
+            self.Outputs(chunk = chunk)
+            for chunk in chunks
+        ]