开源AI工作流：非LLM AI代理框架

[=] Requirements

[~] To run and execute IntegratedPipeline, Requirement's include (Or see Requirements: - Machine (Choose one minimal, specified for your needs): - Windows Native OS - AbstractIntegratedModule.cp313-win_amd64 Supports Python 3.13 for Windows Only (Robust, Stable version) - Linux x86_64 - AbstractIntegratedModule.cpython-312-x86_64-linux-gnu.so supports Python 3.12 only. - Linux ARM64 - Raspberry Pi (Supports Raspberry pi 3 - 5) - AbstractIntegratedModule.cpython-310-aarch64-linux-gnu.so supports Python 3.10 only. - Libraries: - Pandas - aiohttp - scikit-learn - numpy - psutil - Download AbstractIntegratedModule from release section or in here, AbstractIntegratedModule Package - This file contains: - AbstractIntegratedModule.pyd (For windows machine). - AbstractIntegratedModule.cpython-39-x86_64-linux-gnu.so (For linux x86_64). - AbstractIntegratedModule.cpython-39-aarch64-linux-gnu.so (for Linux ARM64 - Raspberry Pi).

[=] for labels assignation: - CSV file that contains training labels and titles used for training and prediction label map, example will be provided below, Go to Step's for in-depth Usage

[=] Requirements for Docker container

• Dockerfile (For Container assembler)
• Python scripts (Such as main.py for Dockerfile usage).
• entrypoint.sh (for smart entry point for Dockerfile container usage).

[=] Docker Container Application

0. See Docker_installation_Section for an in-depth explanation, or Quick_Docker_start for a quick start. - Note Consider checking: - Dockerfile contains all the instructions need to assemble a Docker container. - start.sh for Quick single agent start in Docker container. - entrypoint.sh for a Smart entrypoint used in Dockerfile (Must be downloaded too along with Dockerfile and put in the same folder with Dockerfile after its downloaded) - start-multi-agent-cluster.sh for Multi-agent cluster start in Docker container, What it does: - Starts a multi-agent Docker cluster — runs docker-compose up -d in detached mode (background), scaling the agent-client service to 5 simultaneous instances, alongside whatever server is defined in the docker-compose.yml. - Tails the logs — runs docker-compose logs -f which streams live logs from all containers (server + all 5 clients) to your terminal until you hit Ctrl+C. - main.py for executing a python script in the Docker container that used main.py, like in this code: -

          docker run -it -v $(pwd):/app/data integrated-pipeline:latest python /app/data/main.py
          

1. Build Image: - Clone repository: -

     git clone https://github.com/Micro-Novelty/IntegratedPipeline-Continous-Learning-AI-Agent-library-framework.git
     cd IntegratedPipeline-Continous-Learning-AI-Agent-library-framework
     

       # remove .txt extension
       mv Dockerfile.txt Dockerfile
       mv env.txt .env
       mv dockerignore.txt .dockerignore
       

     cd /path/to/your/folder
     

     sudo docker build -t integrated-agent.
     

      docker run -it --name ai-agent integrated-agent:latest python
      

         from AbstractIntegratedModule import IntegratedPipeline, PipelinePredictionManager
         model = IntegratedPipeline('agent_memory')
         print("✓ IntegratedPipeline initialized successfully!")
         

   # Mount your local directory and run a script
   docker run -it -v $(pwd)/data:/app/data integrated-agent:latest python main.py # main.py could be replaced
   

     # For NVIDIA GPU support
     docker run -it --gpus all -v $(pwd)/data:/app/data integrated-agent:latest python main.py
     

   # Build image
   docker build -t integrated-agent:latest .

   # Run single agent
   docker run -it -v $(pwd)/data:/app/data integrated-agent:latest python
   

     cd /path/to/your/folder
     

     # Start multiple agents
     docker-compose up -d

     # View logs
     docker-compose logs -f

     # Stop all agents
     docker-compose down
     

CohesiveAgentDeployment is deeply tied and coupled with AgentDistributedInference,

if you already set an SSL cert and key, CohesiveAgentDeployment will use the SSL directly from AgentDistributedInference

allowing secure socket to be used directly by CohesiveAgentDeployment

Note: CohesiveAgentDeployment contains ConsecutivePeerAgent that can start a server once ensemble prediction from peer is started

be advised to stop the server too before shutdown-ing CohesiveAgentDeployment cluster

[=] Step's for in-depth Usage

1. Download: - AbstractIntegratedModule.pyd (For Windows) (Python 3.13), - AbstractIntegratedModule.cpython-39-x86_64-linux-gnu.so (For linux x86_64) (Python 3.12) - AbstractIntegratedModule.cpython-39-aarch64-linux-gnu.so (for Linux ARM64 - Raspberry Pi) (Python 3.10) -

      # Download from release
      # AbstractIntegratedModule.pyd (windows) /
      # Abstractcpython-39-x86_64-linux-gnu.so (x86_64) /
      # AbstractIntegratedModule.cpython-39-aarch64-linux-gnu.so   
      

           # prerequisites (for Raspberry pi OS Only)
           # Update system
           sudo apt-get update
           sudo apt-get upgrade -y

           # Install Python 3.13 and development tools
           sudo apt-get install python3.13 python3.13-dev python3.13-venv -y

           # Install additional build tools
           sudo apt-get install build-essential libatlas-base-dev libjasper-dev -y

           # Clone immediately for Windows and x86_64 only without prerequisites          
           git clone https://github.com/Micro-Novelty/IntegratedPipeline-Continous-Learning-AI-Agent-library-framework.git
           cd IntegratedPipeline-Continous-Learning-AI-Agent-library-framework     
           

      # Ubuntu/Debian
      sudo apt-get update
      sudo apt-get install python3.13 python3.13-dev python3.13-venv

      # CentOS/RHEL
      sudo yum install python313 python313-devel

      # Fedora
      sudo dnf install python3.13 python3.13-devel
      

         # Create virtual environment (windows)
         python -m venv venv
         # Activate virtual environment
         venv\Scripts\activate
         
         # Create virtual environment (x86_64) (ARM64 / raspberry pi)
         python3.13 -m venv venv
         # Activate virtual environment
         source venv/bin/activate
         

        # For windows:
        # Copy the .pyd file to your project root
        # AbstractIntegratedModule.pyd
        copy C:\path\to\AbstractIntegratedModule.pyd .\AbstractIntegratedModule.pyd
        
        # Copy the .so file to your project root (for x86_64)
        cp /path/to/AbstractIntegratedModule.cpython-39-x86_64-linux-gnu.so ./AbstractIntegratedModule.cpython-39-x86_64-linux-gnu.so
        
        # Copy ARM64 / Raspberry pi binary
        cp /path/to/AbstractIntegratedModule.cpython-39-aarch64-linux-gnu.so ./AbstractIntegratedModule.cpython-39-aarch64-linux-gnu.so

   5. Verify Installation:
      - 

        
   6. Run main.py for quick test of successful imports:
      - 

           
         
     
3. Create CSV file that contains training labels and titles:
   -  Example format:
      

      Note = window_title is target_title and label is target_label, check step below to use it.

4. Use IntegratedPipeline as in this example:
   

memory_name = 'agent_memory' main_model = IntegratedPipeline(memory_name, use_async=True, ssl_cert_file=cert_file, ssl_key_file=key_file) # provide cert_file path or key_file path (optional) main_prediction = PipelinePredictionManager(main_model, label_csv='example_manual_training.txt', target_title='window_title', label='label') # example_manual_training is a .txt file that contain csv format like above example. example_rules = [ # === WORK / PRODUCTIVITY === (r'code|programming|develop|debug|compile|script', 'focused_work'), (r'vscode|visual_studio|ide|terminal|shell', 'focused_work'), (r'notion|evernote|onenote|notes|todo|task', 'productive'), (r'slack|teams|discord|zoom|meeting|call', 'communication'), (r'email|gmail|outlook|inbox|mail', 'communication'), # === ENTERTAINMENT === (r'youtube|netflix|twitch|stream|video', 'entertainment'), (r'music|spotify|soundcloud|audio|player', 'entertainment'), (r'game|gaming|steam|epic|play', 'gaming'), (r'facebook|instagram|tiktok|social|post', 'social_media'), # === BROWSING === (r'chrome|firefox|edge|safari|browser', 'browsing'), (r'google|search|wiki|wiki|article', 'information'), (r'stackoverflow|github|docs|documentation', 'research'),

# more rules ] # activate explainability capability to explain uncertainty: main_model.show_explainability_details = True # test samples with more sophisticated rules and more complex titles for prediction # (title, intent) test_titles = [ ("Opening Thesis.docx", "slight_work"), ("Watching YouTube and Google Chrome", "distracted"), ("Watching Slack", "communication"), ("Programming in Visual Studio Code", "focused_work"), ("Watching netflix.com - Chrome", "break"), # more titles ] titles, y, label_map = main_prediction.load_labels_from_csv(<your_filename>, <target_title>, <target_label>) # small training with simple titles main_model.train(titles, y) results, chosen_label, confidence = main_prediction.advanced_prediction_method(test_titles, label_map, example_rules, show_proba=False, top_k=3, use_transformer=True, return_attention=False, save_results=True)

# ... more features you can add

   
5. To use IntegratedPipeline prediction without Transformer, Only Specialized MLP:
      Note: IntegratedPipeline without Transformer is'nt recommended due to it being weak at certain contextual prediction's, excel's at classification task's.
      - Example without transformer:
   

6. Asynchronous prediction:
  - Asynchronous prediction request is important and is critical because it keeps prediction interfaces responsive, maximizes local hardware efficiency, and enables apps to handle background tasks seamlessly without waiting on remote server responses,
  - for asynchronous prediction handling, consider using this setup

print(" = TESTING ASYNCHRONOUS PREDICTION MANAGER = ")

for better and faster advanced prediction, consider using advanced batch prediction like in the above example

[=] Note:
 - Asynchronous prediction used Event loop that handles incoming request, There are conditions where event loop will not start and can't accet requests:
   - CPU Above > 95%    - Disk space is < 100 MB
   - RAM above > 95%
 - When event loop is not triggered, Asynchronous prediction can't be initiated and must be restarted/retried.

7. Peer-to-Peer Probability coordination:
   - Each peer is both server and client simultaneously for robustness and resilience during during P2P.
   - To Make the Agent cooperate with other peers, consider using this setup:
   - [=] for ensemble prediction from multiple peers, exchanging predicted label with each other, consider using this setup:

prediction_manager = PipelinePredictionManager(main_model, label_csv=<your_training_labels.txt>, target_title=<target_title>, label=<target_label>)

print("=== SECURE PEER-TO-PEER CLUSTER ===")

example peer_Ip_lists_config.json (de-comment to use)

Set discovery secret (in production, use environment variable)

secret_key = 'my-ultra-safe-secret-key-for-authentication' # you can customize this key

security_config = SecurityConfig( max_text_length=10000, # can be extended max_queue_size=100, # can be extended rate_limit_requests=60, # 60 per minute require_api_key=True, # max_pending_tasks=50, request_timeout=60.0,

# Start with no IP restrictions, you can add allowed IPs for asynchronous prediction externally, boothstrap_auth for better security allowed_ips=[], blocklisted_ips=[], require_bootstrap_auth = False # true for better security (Not recommended, cause less flexibility) )

async_manager = PipelineAsyncManager(main_model, main_prediction, # your previous initialized PipelinePredictionManager config=security_config, state_file=None, # state file is used to load known security logs ex: ip used, ip blacklisted, etc. security_level=SecurityLevel.PRODUCTION, # production level security initiated max_workers=4, # workers to initiate prediction, more workers, more capabilities to process prediction requests. task_timeout=30, max_retries=3 ) # retries after failure during prediction

async_manager.start(method='Transformer_included', bootstrap_token=None) # boothstrap token is optional for better security

texts = {'test_titles': test_titles, 'label_map': label_map, 'rules': example_rules, 'use_transformer': True} regular_predict = async_manager.predict( texts=texts, timeout=60, retries=None, api_key=secret_key) # advanced prediction method for asynchronous prediction.

3 options for ensemble weighting method:

set connection timeout (Optional)

main_model.distribution.connection_timeout = 30 # 30 seconds before timeout calibrated_probability = main_model._handle_distributed_connections(probs, attn_weights, sequence_input, agreement)

Purpose: Analyzes the geometric structure of data (anisotropy, spectral properties, complexity) and generates optimal weight matrices based on that geometry. Essentially teaches the model how to "understand" the shape of data before processing it, Highly robust to noise, making it an excellent fit for messy environment.

1. Activation

Purpose: A complete transformer implementation with multi-head attention, positional embeddings, feed-forward networks, layer normalization, and custom backpropagation. Includes both fixed (stable) and dynamic (adaptive) training modes necessary for Scarce data environment using algorithm such as Alpha based computing directly during forward pass.

1. Dense

with retries: async_manager.predict(texts, timeout=60, retries=5, api_key=secret_key) # 5 times retry if failed

print('[==] Initiating advanced batch prediction') predicted_output = async_manager.advanced_batch_prediction(test_titles, label_map, example_rules, secret_key, client_ip=None) # you can add client_ip to provide a robust authentication paired with secret_key

Purpose: High-level prediction interface that loads labels from CSV, performs regular/advanced/hybrid predictions, displays results, and calculates entropy for uncertainty estimation.

1. ThreadedMessageQueue

[=] IntegratedPipeline-Specialized-AI-Agent-library

[~] IntegratedPipeline is a standalone Specialized AI Agent Library for Non-LLL memory Augmented Agentic Framework orchestrator, Specifically designed to provide Agentic capability for any Autonomous Agentic Framework locally and Coordinatively that runs efficiently from consumer based machine to High-end embedded systems, where the AI Can directly and continously learn, with minimal and efficient compute, built-in augmented memory, Secure Peer-To-Peer (Multi-Agent) Coordination with security layers as an option, And Explainability capability based on proof from in it's internal metrics, reducing Black-Box condition necessary for reliability. Containing specialized MLP using Its Own specialized geometric Weight shaping (AWE) and Specialized efficient Transformer for Scarce Data with Alpha-based computation, specifically designed for low-amount samples environment, or Messy environments.

<img width="1600" height="600" alt="WhatsApp Image 2026-05-10 at 18 17 37" src="https://github.com/user-attachments/assets/aaf10427-7ff6-4f54-837d-58d46049de78" />

[+] Why IntegratedPipeline?

[~] IntegratedPipeline is a great choice for a sophisticated Non-LLM AI Program for The Main Orchestrator of a Distributed MANN-Type AI Agent Working in Edge-device/Consumer-Based machine Where LLM is'nt a great fit for Messy, Noisy environments. while still run efficiently on High-end Embedded systems in single-instance or as a distributed network during multi Agent cooperation.

[=] IntegratedPipeline offers: 1. Local-Based AI Orchestrator: - IntegratedPipeline Creates its own SQLite Database inside Your Computer once the library is executed, This database is used directly to store the AI Memory, Attention weights, predicted Output, and identified peer, all without leaving the machine, The Database will be created Automatically once you run the library, database name saved as activity_log.db.

2. Continously Learning behavior for an Agent: - different from LLM that is static and cannot improve beyond its given training condition, AI Agent using IntegratedPipeline has a dynamic, flexible continously learning behavior with conditional training algorithms included in the library that has both supervised and unsupervised learning present, The learned input and predicted Output will be stored in the database, allowing it to recall its memory during processing and find matching known prediction given if input matched with the stored input inside the database. this Continous learning behavior is efficient because its not relying on weights for memory, allowing flexible and predictable behavior inside a given environment. 3. Robust Specialized MLP and Transformer Architecture: - IntegratedPipeline has 2 Different type's of AI Architecture stacked together, Specialized MLP for Noise robustness And Specialized Transformer that used Alpha-based Computing algorithm for contextual reasoning, The reason why those Models complement and used together : - Specialized MLP Provides synchronous robust classification Against noise with its specialized Weight Encoder (AWE) to handle noise using eigenvalue based computing that is lightweight and efficient. This Method can't be replicated Inside Transformer FFN (Feed-forward-network) because of Transformer dynamic brute force computing where AWE-Based generated weight's get diluted over time by Transformer dynamic projection embedding, making AWE Generated weight causes inefficient inside Transformer dynamic FFN/QKV projection. - Specialized Transformer provides robust synchronous advanced contextual relationships, efficient data processing using Alpha based computing, The Transformer is tuned towards to be as flexible as possible to provide dynamic projection or fixed FFN projection training with minimal head's and dimension's to reduce computational power. 4. flexible and secure Peer-to-Peer Coordination (Multi-Agent): - IntegratedPipeline offers Peer to Peer communication capabilities asynchronously, Where IntegratedPipeline directly checks for other Peer presence directly to the local database present in the local computer or system (Synchronous prediction from peer previous data in the database), or externally, by using asynchronous request for initiating prediction, P2P is secured Using: - secure socket using user provided SSL CERT. on both client and server, - API key for requesting, - Alpha rate limiting, - HMAC secret key for authentication, - and IP validation. [~] Each agent has double roles during P2P: - Server provider: the peer Agent can start a server to listen for peer client's - Connecting Agent: the peer Agent which happens to connect to other peer that has or have opened and provided a server listener to act as a receiver.

5. Cross-Session memory availability: - IntegratedPipeline offers share-able Memory capability, included capability below: - Exportable memory: this allows a flexible memory saving for later use, such as cross transfer memory between model, the memory is saved as .json file after exporting. - Importable Memory: allowing to import memory from the exported .json file directly for the model to use. - syncing with other model: socket-based communication to export memory to other external machine. - [=] Note: socket Syncing is unsecure witout additional security layer wrapped, For a safer syncing, directly transfer the .json file memory to the target machine via other ways such as manual send. - list sessions: listing available sessions using model's memory name.

[=] With its Specialized Multi-Layer-Perceptron (using AWE Encoder) and Optimized Transformer module with optimized Embedding, IntegratedPipeline can directly tolerate low samples-amount of Data, including noisy ambiguous data, using Weighted Confidence assembling from both specialized MLP and Transformer for better reliability during training and prediction over Messy, noisy environments, such as:

[1.] User data's : User data is often messy and ambiguous, Specialized MLP with AWE will do the job for shaping the necessary Weight to complement for the Ambiguous noisy pattern, AWE MLP is highly robust to noise, proven in synthetic Environment such as scikit-learn Make-Classification scarce and Noisy Input robustness during generalization test. making it a great fit for messy, Scarce data.

[2.] Small Dataset's : We often don't have enough Dataset to train a Transformer Model, Thats Why IntegratedPipeline Offers a Highly optimized Transformer that supports scarce dataset processing, Using Alpha-Based computing as a Warm-up for training, it provides a direct Boost for the transformer to be efficient during training in scarce-data Environment.

[3.] Non-Representative data (Undersampled) : IntegratedPipeline Support's ambiguous data that come's from file with format such as CSV Format to extract title's and label's necessary to create automatic Dataset for Later use in Training, making it optimized for specific task's and easier dataset creation with lower overfitting rate for reliability.

[=] Architectural-Overview <img width="1600" height="860" alt="WhatsApp Image 2026-05-09 at 16 13 41" src="https://github.com/user-attachments/assets/580722aa-bbd8-4148-a425-4bff01c06c47" /> [=] Contextual meaning: 1. - Sequence encoding is a machine learning technique that transforms a sequential input (like text, time-series data, or audio) into a compact, fixed-length numerical vector, often called a context vector 2. - TF-IDF (Term Frequency-Inverse Document Frequency) is a numerical statistic used in machine learning and NLP to evaluate how important a word is to a document within a collection (corpus). It boosts rare words and penalizes common words (like "the", "and") by multiplying two metrics: how often a word appears in a document (TF) and the inverse frequency of the word across all documents (IDF). 3. - Explainability provides deeper transparency of why a model thought about a detail by showing its internal metrics like attention quality, from distributed peer memory or Ensemble prediction result's.

Main Components

[=] 1. - Consider checking and run: IntegratedPipeline_Flow.html regarding each function of the whole components and deep-dive mechanism. 2. - consider checking ARCHITECTURE.md for more explanation about the main components. [=] With 20 total architectures working together as a standalone library that is efficient and robust, Main components include: 1. GeometricWeightShaping

[+] Troubleshooting

1. Issue 1: "ModuleNotFoundError: No module named 'AbstractIntegratedModule'" Solution:

- [=] Verify the binary file is in the correct location:

   ls -la AbstractIntegratedModule*.so  # Linux
   dir AbstractIntegratedModule.pyd     # Windows
   

- [=] Check Python path:

   python -c "import sys; print('\n'.join(sys.path))"
   

- 1. Ensure AbstractIntegratedModule.pyd is in your project root - 2. Install Visual C++ redistributables: Download from: https://support.microsoft.com/en-us/help/2977003 - 3. Verify Python architecture (32-bit vs 64-bit) matches the .pyd file

3. Issue 3: "Permission denied" (Linux/Raspberry Pi) [=] Solution: -

     # Make sure you have read permissions
     chmod 644 AbstractIntegratedModule.cpython-39-*.so
     # If in virtual environment, ensure it's activated
     source venv/bin/activate
    

4. Issue 4: Missing Dependencies [=] Solution: -

      # Reinstall all dependencies
      pip install --upgrade pip setuptools wheel
      pip install numpy pandas scikit-learn matplotlib scipy requests
     
      # Verify installation
      pip list
      

5. Issue 5: Virtual Environment Issues [=] Solution: -

      # Deactivate current environment
      deactivate

      # Create a fresh virtual environment
      python -m venv fresh_venv

      # Activate it
      source fresh_venv/bin/activate  # Linux/Raspberry Pi
      # or
      fresh_venv\Scripts\activate     # Windows

      # Reinstall
      pip install --upgrade pip
      pip install numpy pandas scikit-learn matplotlib scipy
      

6. Issue 6: Raspberry Pi - "Bus error" or Performance Issues Solution: - [=] Ensure adequate swap space:

      free -h  # Check current swap
      sudo nano /etc/dphys-swapfile  # Increase if needed
       

     [❌] Failed to connect to <host>:<port>: Nonetype object has no attribute .accept()
     

       # initiate socket first
       main_model.distribution.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
       main_model.distribution.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
       main_model.distribution.socket.bind(('0.0.0.0', self.port)) # self.port could be changed with other ports
       main_model.distribution.socket.listen(5) # listens for 5 seconds