upload codes

Author: sherlockchou86

.gitignore

@@ -172,3 +172,6 @@ cython_debug/
 
 # PyPI configuration file
 .pypirc
+
+# test files
+tests/files/

README.md

@@ -0,0 +1,12 @@
+from .lpnode import LPNode, LPNodeType, LPNodeState
+from .lpbegin import LPBegin
+from .lpend import LPEnd
+from .lpclassifier import LPClassifier
+from .lpextractor import LPExtractor
+from .lpaggregator import LPAggregator
+from .lpgenerator import LPGenerator
+from .lpchatter import LPChatter
+from .lpbaseinvoker import LPBaseInvoker
+from .lpbaserouter import LPBaseRouter
+from .lpsqlcreator import LPSQLCreator
+from .lpboardrender import LPBoardRender

docs/pipeline.png

@@ -0,0 +1,66 @@
+import json
+from ollama import generate
+from .lpnode import LPNode, LPNodeType
+
+class LPAggregator(LPNode):
+    """
+    data aggregation using LLM base on Ollama.
+    access `lpdata['global_vars']['aggregated_data']` for the aggregated data.
+    """
+    def __init__(self, name, aggregate_desc=None, model='minicpm-v:8b') -> None:
+        super().__init__(name, LPNodeType.LLM, model)
+        self.__aggregate_desc = aggregate_desc
+        self.__aggregated_data = None
+        self.__aggregate_prompt_template = """
+        你是一个智能信息聚合器（Aggregator），负责整合多个来源的信息，并基于所有可用数据生成高质量、清晰且有逻辑性的最终回答。
+        以下是所有可用的信息：
+        ---
+        {0}
+        ---
+
+        以下是待回答的问题：
+        ---
+        {1}
+        ---
+
+        任务要求：
+        1. **信息整合**：充分利用所有来源的数据，确保信息完整，不遗漏任何重要内容。
+        2. **语义流畅**：避免直接罗列数据，而是用自然语言组织，使回答清晰易懂。
+
+        请给出最终整合后的完整回答：
+        """
+
+    def _handle(self, lpdata) -> None:
+        query = lpdata['query'] if self.__aggregate_desc is None else self.__aggregate_desc
+        prompt = self.__aggregate_prompt_template.format(json.dumps(lpdata['global_vars'], indent=4, ensure_ascii=False), query)
+
+        response = generate(model=self.model, 
+                            prompt=prompt,
+                            options={
+                                'top_k': 1,
+                                'temperature': 0.5
+                            })
+        self.__aggregated_data = response['response']
+
+        # update records
+        messages = lpdata['records'][-1]['messages']
+        message = {}
+        message['role'] = 'user'
+        message['content'] = prompt
+        messages.append(message)
+
+        message = {}
+        message['role'] = 'assistant'
+        message['content'] = response['response']
+        messages.append(message)
+    
+    def _after_handle(self, lpdata) -> None:
+        super()._after_handle(lpdata)
+
+        # update local vars
+        record = lpdata['records'][-1]
+        record['local_vars']['__aggregate_desc'] = self.__aggregate_desc
+
+        # update global variables
+        lpdata['final_out'] = self.__aggregated_data
+        lpdata['global_vars']['aggregated_data'] = self.__aggregated_data

langpipe/__init__.py

@@ -0,0 +1,20 @@
+
+from abc import ABC, abstractmethod
+from typing import final
+from .lpnode import LPNode, LPNodeType
+
+class LPBaseInvoker(ABC, LPNode):
+    """
+    base class for all invoker nodes, which invokes external services, access database, or call 3rd tools.
+    """
+    def __init__(self, name) -> None:
+        super().__init__(name, LPNodeType.Invoke, None)
+    
+    @final
+    def _handle(self, lpdata) -> None:
+        # hidden _handle(...) in derived classes, using _invoke(...) instead
+        self._invoke(lpdata)
+    
+    @abstractmethod
+    def _invoke(self, lpdata) -> None:
+        pass

langpipe/lpaggregator.py

@@ -0,0 +1,31 @@
+import threading
+from abc import ABC, abstractmethod
+from typing import final
+from .lpnode import LPNode, LPNodeType
+
+class LPBaseRouter(ABC, LPNode):
+    """
+    base class for router node, which is used to route the data to different branches in pipeline.
+    """
+    def __init__(self, name) -> None:
+        super().__init__(name, LPNodeType.Invoke, None)
+    
+    @abstractmethod
+    def _condition_check(self, lpdata) -> int:
+        """
+        if/elif/.../else condition check, return int value to identity which branch to run.
+        - 0 means first branch to run
+        - 1 means second branch to run
+        - ...
+        """
+        pass
+
+    def _dispatch(self, lpdata) -> None:
+        route_id = self._condition_check(lpdata)
+
+        if route_id >= 0 and len(self.next_nodes) > route_id:
+            node = self.next_nodes[route_id]
+            if lpdata['sync']:
+                node.run(lpdata)
+            else:
+                threading.Thread(target=lambda d: node.run(d), args=(lpdata,)).start()

langpipe/lpbaseinvoker.py

@@ -0,0 +1,44 @@
+from datetime import datetime
+from .lpnode import LPNode, LPNodeType
+
+
+class LPBegin(LPNode):
+    """
+    begin node in pipeline.
+    """
+
+    def __init__(self, name) -> None:
+        super().__init__(name, LPNodeType.Begin, None)
+    
+    def input(self, query, query_images=None, sync=True):
+        """
+        construct lpdata and call super().run(), start to run the pipeline.
+
+        **parameters**
+        - query(str): input text or prompt.
+        - query_images([]): input images with cv2.mat format, None by default.
+        - sync(bool): start pipeline with sync or aysnc mode.
+        """ 
+        
+        # construct lpdata which will flowing the piepline
+        lpdata = {}
+        # running mode
+        lpdata['sync'] = sync
+        # begin time
+        lpdata['begin_t'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        # end time
+        lpdata['end_t'] = None
+        # input text or prompt
+        lpdata['query'] = query
+        # input images
+        lpdata['query_imgs'] = query_images
+        # final output generated by pipeline
+        lpdata['final_out'] = None
+        # global variables when lpdata flowing the pipeline
+        lpdata['global_vars'] = {}
+        # handle records when lpdata flowing the pipeline
+        lpdata['records'] = []
+
+        super().run(lpdata)
+
+

langpipe/lpbaserouter.py

@@ -0,0 +1,112 @@
+import cv2
+import threading
+import numpy as np
+from datetime import datetime
+from .lpnode import LPNodeState
+
+class LPBoardRender:
+    """
+    pipeline visualization tools based on OpenCV library.
+    """
+    def __init__(self, node_size=50, h_spacing=100, v_spacing=80):
+        self.__node_size = node_size
+        self.__h_spacing = h_spacing
+        self.__v_spacing = v_spacing
+        self.__y_positions = {}
+        self.__img = None
+        self.__colors = {
+            LPNodeState.Pending: (0, 0, 0),    # bgr is black
+            LPNodeState.Runing: (0, 0, 255),   # bgr is red
+            LPNodeState.Completed: (255, 0, 0) # bgr is blue
+        }
+        self.__runing = False
+    
+    def __del__(self):
+        self.__runing = False
+
+    def __get_tree_depth(self, root):
+        if not root.next_nodes:
+            return 1
+        return 1 + max(self.__get_tree_depth(child) for child in root.next_nodes)
+
+    def __get_layer_nodes(self, root, depth=0, layers=None):
+        if layers is None:
+            layers = {}
+        if depth not in layers:
+            layers[depth] = []
+        layers[depth].append(root)
+        for child in root.next_nodes:
+            self.__get_layer_nodes(child, depth + 1, layers)
+        return layers
+
+    def __draw_node(self, node, x, y):
+        cv2.rectangle(self.__img, (x, y), 
+                      (x + self.__node_size, y + self.__node_size), 
+                      self.__colors[node.state], 2)
+        
+        # name
+        text_size = cv2.getTextSize(node.name, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)[0]
+        text_x = x + (self.__node_size - text_size[0]) // 2
+        text_y = y + (self.__node_size + text_size[1]) // 2
+        cv2.putText(self.__img, node.name, (text_x, text_y), 
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, self.__colors[node.state], 1)
+        
+        # type
+        text_size = cv2.getTextSize('[' + node.type.name + ']', cv2.FONT_HERSHEY_SIMPLEX, 0.4, 1)[0]
+        text_x = x + (self.__node_size - text_size[0]) // 2
+        text_y = y + text_size[1] + 5
+        cv2.putText(self.__img, '[' + node.type.name + ']', (text_x, text_y), 
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.4, self.__colors[node.state], 1)
+
+        # cost time
+        if node.state == LPNodeState.Completed:
+            text_size = cv2.getTextSize(str(node.cost_time) + 'sec', cv2.FONT_HERSHEY_SIMPLEX, 0.4, 1)[0]
+            text_x = x + (self.__node_size - text_size[0]) // 2
+            text_y = y + self.__node_size - 5
+            cv2.putText(self.__img, str(node.cost_time) + 'sec', (text_x, text_y), 
+                        cv2.FONT_HERSHEY_SIMPLEX, 0.4, self.__colors[node.state], 1)
+
+        # child nodes
+        child_x = x + self.__node_size + self.__h_spacing
+        for child in node.next_nodes:
+            child_y = self.__y_positions[child]
+            cv2.line(self.__img, (x + self.__node_size, y + self.__node_size // 2), 
+                     (child_x, child_y + self.__node_size // 2), self.__colors[child.state], 1, cv2.LINE_AA)
+            cv2.circle(self.__img, (child_x, child_y + self.__node_size // 2), 
+                       5, self.__colors[child.state], -1)
+            self.__draw_node(child, child_x, child_y)
+    
+    def __draw_info(self):
+        cv2.putText(self.__img, datetime.now().strftime("%Y-%m-%d %H:%M:%S"), (20, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 1)
+        
+    def __draw_board(self, root):
+        layers = self.__get_layer_nodes(root)
+        max_depth = max(layers.keys())
+        max_width = max(len(layers[d]) for d in layers)
+        
+        img_height = max(400, max_width * (self.__node_size + self.__v_spacing))
+        img_width = (max_depth + 1) * (self.__node_size + self.__h_spacing)
+        
+        for depth, nodes in layers.items():
+            layer_height = len(nodes) * (self.__node_size + self.__v_spacing)
+            start_y = (img_height - layer_height + self.__v_spacing) // 2
+            self.__y_positions.update({node: start_y + i * (self.__node_size + self.__v_spacing) for i, node in enumerate(nodes)})
+                
+        root_x = self.__h_spacing // 2
+        root_y = self.__y_positions[root]
+
+        while self.__runing:
+            self.__img = np.ones((img_height, img_width, 3), dtype=np.uint8) * 255
+            self.__draw_node(root, root_x, root_y)
+            self.__draw_info()
+            cv2.imshow("Tree", self.__img)
+            if cv2.waitKey(100) & 0xFF == 27:
+                break
+
+    def render(self, root, block=True):
+        self.__runing = True
+        render_th = threading.Thread(target=self.__draw_board, args=(root,), daemon=True)
+        render_th.start()
+
+        if block:
+            render_th.join()

langpipe/lpbegin.py

@@ -0,0 +1,40 @@
+import json
+from ollama import chat
+from .lpnode import LPNode, LPNodeType
+
+class LPChatter(LPNode):
+    """
+    chat with LLM based on Ollama, supporting chat histories which is different from text generation.
+    access `lpdata['global_vars']['chatted_text']` to get the chat text.
+    """
+    def __init__(self, name, model='minicpm-v:8b') -> None:
+        super().__init__(name, LPNodeType.LLM, model)
+        self.__chatted_text = None
+
+    def _handle(self, lpdata) -> None:
+        # convert query to json object something like: [{'role': 'user', 'content': 'Hello, how are you?'}, {...}, {...}]
+        messages_from_query = json.loads(lpdata['query'])
+        response = chat(model=self.model, 
+                        messages=messages_from_query,
+                        options={
+                            'top_k': 100,
+                            'temperature': 0.8
+                        })
+        # take care <think>...</think> in some reason models
+        self.__chatted_text = response['message']['content']
+
+        # update records
+        messages = lpdata['records'][-1]['messages']
+        messages.extend(messages_from_query)
+
+        message = {}
+        message['role'] = 'assistant'
+        message['content'] = response['message']['content']
+        messages.append(message)
+    
+    def _after_handle(self, lpdata) -> None:
+        super()._after_handle(lpdata)
+
+        # update global variables
+        lpdata['final_out'] = self.__chatted_text
+        lpdata['global_vars']['chatted_text'] = self.__chatted_text