KG prompt working

scripts/language-generic

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.language.prompt.generic import run
+
+run()
+

tests/test-lang-definition

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+
+import pulsar
+from trustgraph.prompt_client import PromptClient
+
+p = PromptClient(pulsar_host="pulsar://localhost:6650")
+
+chunk = """I noticed a cat in my garden.  It is a four-legged animal
+which is a mammal and can be tame or wild.  I wonder if it will be friends
+with me.  I think the cat's name is Fred and it has 4 legs"""
+
+resp = p.request_definitions(
+    chunk=chunk,
+)
+
+for d in resp:
+    print(d.name, ":", d.definition)
+

tests/test-lang-kg-prompt

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+
+import pulsar
+from trustgraph.prompt_client import PromptClient
+
+p = PromptClient(pulsar_host="pulsar://localhost:6650")
+
+facts = [
+      ("accident", "evoked", "a wide range of deeply felt public responses"),
+      ("Space Shuttle concept", "had", "genesis"),
+      ("Commission", "had", "a mandate to develop recommendations for corrective or other action based upon the Commission's findings and determinations"),
+      ("Commission", "established", "teams of persons"),
+      ("Space Shuttle Challenger", "http://www.w3.org/2004/02/skos/core#definition", "A space shuttle that was destroyed in an accident during mission 51-L."),
+      ("The mid fuselage", "contains", "the payload bay"),
+      ("Volume I", "contains", "Chapter IX"),
+      ("accident", "resulted in", "firm national resolve that those men and women be forever enshrined in the annals of American heroes"),
+      ("Volume I", "contains", "Chapter IV"),
+      ("Volume I", "contains", "Appendix A"),
+      ("Volume I", "contains", "Appendix B"),
+      ("Volume I", "contains", "The Staff"),
+      ("Commission", "required", "detailed investigation"),
+      ("Commission", "focused", "safety aspects of future flights"),
+      ("Commission", "http://www.w3.org/2004/02/skos/core#definition", "An independent group appointed to investigate the Space Shuttle Challenger accident."),
+      ("Commission", "moved forward with", "its investigation"),
+      ("President", "appointed", "an independent Commission"),
+      ("accident", "interrupted", "one of the most productive engineering, scientific and exploratory programs in history"),
+      ("Volume I", "contains", "Preface"),
+      ("Commission", "believes", "investigation"),
+      ("Volume I", "contains", "Chapter I"),
+      ("President", "was moved and troubled", "by this accident in a very personal way"),
+      ("PRESIDENTIAL COMMISSION", "Report to", "President"),
+      ("Volume I", "contains", "Chapter VI"),
+      ("Commission", "held", "public hearings dealing with the facts leading up to the accident"),
+      ("Volume I", "http://www.w3.org/2004/02/skos/core#definition", "The first volume of a multi-volume publication."),
+      ("Space Shuttle Challenger", "was involved in", "an accident"),
+      ("Volume I", "contains", "Chapter VII"),
+      ("Volume I", "contains", "Chapter II"),
+      ("Volume I", "contains", "Chapter V"),
+      ("Commission", "believes", "its investigation and report have been responsive to the request of the President and hopes that they will serve the best interests of the nation in restoring the United States space program to its preeminent position in the world"),
+      ("Commission", "supported", "panels"),
+      ("Volume I", "contains", "Chapter VIII"),
+      ("NASA", "cooperated", "Commission"),
+      ("liquid oxygen tank", "contains", "oxidizer"),
+      ("President", "http://www.w3.org/2004/02/skos/core#definition", "The head of state of the United States."),
+      ("Volume I", "contains", "Chapter III"),
+      ("Apollo lunar landing spacecraft", "had", "not yet flown"),
+      ("Commission", "construe", "mandate"),
+      ("accident", "became", "a milestone on the way to achieving the full potential that space offers to mankind"),
+      ("Volume I", "contains", "The Commission"),
+      ("Commission", "focused", "attention"),
+      ("Commission", "learned", "lessons"),
+      ("Commission", "required", "interfere with or supersede Congress"),
+      ("Commission", "was made up of", "persons not connected with the mission"),
+      ("Commission", "required", "review budgetary matters"),
+      ("Space Shuttle", "became", "focus of NASA's near-term future"),
+      ("Volume I", "contains", "Appendix C"),
+      ("accident", "caused", "grief and sadness for the loss of seven brave members of the crew"),
+      ("Commission", "http://www.w3.org/2004/02/skos/core#definition", "A group established to investigate the space shuttle accident"),
+      ("Volume I", "contains", "Appendix D"),
+      ("Commission", "had", "a mandate to review the circumstances surrounding the accident to establish the probable cause or causes of the accident"),
+      ("Volume I", "contains", "Recommendations")
+]
+
+query="Present 20 facts which are present in the knowledge graph."
+
+resp = p.request_kg_prompt(
+    query=query,
+    kg=facts,
+)
+
+print(resp)
+

tests/test-lang-relationships

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+
+import pulsar
+from trustgraph.prompt_client import PromptClient
+
+p = PromptClient(pulsar_host="pulsar://localhost:6650")
+
+chunk = """I noticed a cat in my garden.  It is a four-legged animal
+which is a mammal and can be tame or wild.  I wonder if it will be friends
+with me.  I think the cat's name is Fred and it has 4 legs"""
+
+resp = p.request_relationships(
+    chunk=chunk,
+)
+
+for d in resp:
+    print(d.s)
+    print("  ", d.p)
+    print("  ", d.o)
+    print("  ", d.o_entity)
+

trustgraph/model/language/prompt/generic/__init__.py

@@ -0,0 +1,3 @@
+
+from . service import *
+

trustgraph/model/language/prompt/generic/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . service import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph/model/language/prompt/generic/prompts.py

@@ -0,0 +1,81 @@
+
+def to_relationships(text):
+
+    prompt = f"""<instructions>
+Study the following text and derive entity relationships.  For each
+relationship, derive the subject, predicate and object of the relationship.
+Output relationships in JSON format as an arary of objects with fields:
+- subject: the subject of the relationship
+- predicate: the predicate
+- object: the object of the relationship
+- object-entity: false if the object is a simple data type: name, value or date.  true if it is an entity.
+</instructions>
+
+<text>
+{text}
+</text>
+
+<requirements>
+You will respond only with raw JSON format data. Do not provide
+explanations. Do not use special characters in the abstract text. The
+abstract must be written as plain text.  Do not add markdown formatting
+or headers or prefixes.
+</requirements>"""
+    
+    return prompt
+    
+def to_definitions(text):
+
+    prompt = f"""<instructions>
+Study the following text and derive definitions for any discovered entities.
+Do not provide definitions for entities whose definitions are incomplete
+or unknown.
+Output relationships in JSON format as an arary of objects with fields:
+- entity: the name of the entity
+- definition: English text which defines the entity
+</instructions>
+
+<text>
+{text}
+</text>
+
+<requirements>
+You will respond only with raw JSON format data. Do not provide
+explanations. Do not use special characters in the abstract text. The
+abstract will be written as plain text.  Do not add markdown formatting
+or headers or prefixes.  Do not include null or unknown definitions.
+</requirements>"""
+    
+    return prompt
+
+def get_cypher(kg):
+
+    sg2 = []
+
+    for f in kg:
+
+        print(f)
+
+        sg2.append(f"({f.s})-[{f.p}]->({f.o})")
+
+    print(sg2)
+
+    kg = "\n".join(sg2)
+    kg = kg.replace("\\", "-")
+
+    return kg
+
+def to_kg_query(query, kg):
+
+    cypher =  get_cypher(kg)
+
+    prompt=f"""Study the following set of knowledge statements. The statements are written in Cypher format that has been extracted from a knowledge graph. Use only the provided set of knowledge statements in your response. Do not speculate if the answer is not found in the provided set of knowledge statements.
+
+Here's the knowledge statements:
+{cypher}
+
+Use only the provided knowledge statements to respond to the following:
+{query}
+"""
+
+    return prompt

trustgraph/model/language/prompt/generic/service.py

@@ -0,0 +1,195 @@
+
+"""
+Language service abstracts prompt engineering from LLM.
+"""
+
+import json
+
+from ..... schema import Definition, Relationship, Triple
+from ..... schema import PromptRequest, PromptResponse
+from ..... schema import TextCompletionRequest, TextCompletionResponse
+from ..... schema import text_completion_request_queue
+from ..... schema import text_completion_response_queue
+from ..... schema import prompt_request_queue, prompt_response_queue
+from ..... base import ConsumerProducer
+from ..... llm_client import LlmClient
+
+from . prompts import to_definitions, to_relationships, to_kg_query
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = prompt_request_queue
+default_output_queue = prompt_response_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        tc_request_queue = params.get(
+            "text_completion_request_queue", text_completion_request_queue
+        )
+        tc_response_queue = params.get(
+            "text_completion_response_queue", text_completion_response_queue
+        )
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": PromptRequest,
+                "output_schema": PromptResponse,
+                "text_completion_request_queue": tc_request_queue,
+                "text_completion_response_queue": tc_response_queue,
+            }
+        )
+
+        self.llm = LlmClient(
+            subscriber=subscriber,
+            input_queue=tc_request_queue,
+            output_queue=tc_response_queue,
+            pulsar_host = self.pulsar_host
+        )
+
+    def handle(self, msg):
+
+        v = msg.value()
+
+        # Sender-produced ID
+
+        id = msg.properties()["id"]
+
+        kind = v.kind
+
+        print(f"Handling kind {kind}...", flush=True)
+
+        if kind == "extract-definitions":
+
+            self.handle_extract_definitions(id, v)
+            return
+
+        elif kind == "extract-relationships":
+
+            self.handle_extract_relationships(id, v)
+            return
+
+        elif kind == "kg-prompt":
+
+            self.handle_kg_prompt(id, v)
+            return
+
+        else:
+
+            print("Invalid kind.", flush=True)
+            return
+
+    def handle_extract_definitions(self, id, v):
+
+        prompt = to_definitions(v.chunk)
+
+        print(prompt)
+
+        ans = self.llm.request(prompt)
+
+        print(ans)
+
+        defs = json.loads(ans)
+
+        output = []
+
+        for defn in defs:
+
+            try:
+                e = defn["entity"]
+                d = defn["definition"]
+
+                output.append(
+                    Definition(
+                        name=e, definition=d
+                    )
+                )
+
+            except:
+                pass
+
+        print("Send response...", flush=True)
+        r = PromptResponse(definitions=output)
+        self.producer.send(r, properties={"id": id})
+
+        print("Done.", flush=True)
+        
+    def handle_extract_relationships(self, id, v):
+
+        prompt = to_relationships(v.chunk)
+
+        ans = self.llm.request(prompt)
+
+        defs = json.loads(ans)
+
+        output = []
+
+        for defn in defs:
+
+            try:
+                output.append(
+                    Relationship(
+                        s = defn["subject"],
+                        p = defn["predicate"],
+                        o = defn["object"],
+                        o_entity = defn["object-entity"],
+                    )
+                )
+
+            except Exception as e:
+                print(e)
+
+        print("Send response...", flush=True)
+        r = PromptResponse(relationships=output)
+        self.producer.send(r, properties={"id": id})
+
+        print("Done.", flush=True)
+        
+    def handle_kg_prompt(self, id, v):
+
+        prompt = to_kg_query(v.query, v.kg)
+
+        print(prompt)
+
+        ans = self.llm.request(prompt)
+
+        print(ans)
+
+        print("Send response...", flush=True)
+        r = PromptResponse(answer=ans)
+        self.producer.send(r, properties={"id": id})
+
+        print("Done.", flush=True)
+        
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '--text-completion-request-queue',
+            default=text_completion_request_queue,
+            help=f'Text completion request queue (default: {text_completion_request_queue})',
+        )
+
+        parser.add_argument(
+            '--text-completion-response-queue',
+            default=text_completion_response_queue,
+            help=f'Text completion response queue (default: {text_completion_response_queue})',
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph/prompt_client.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+
+import pulsar
+import _pulsar
+from pulsar.schema import JsonSchema
+import hashlib
+import uuid
+
+from . schema import PromptRequest, PromptResponse, Fact
+from . schema import prompt_request_queue
+from . schema import prompt_response_queue
+
+# Ugly
+ERROR=_pulsar.LoggerLevel.Error
+WARN=_pulsar.LoggerLevel.Warn
+INFO=_pulsar.LoggerLevel.Info
+DEBUG=_pulsar.LoggerLevel.Debug
+
+class PromptClient:
+
+    def __init__(
+            self, log_level=ERROR,
+            subscriber=None,
+            input_queue=None,
+            output_queue=None,
+            pulsar_host="pulsar://pulsar:6650",
+    ):
+
+        if input_queue == None:
+            input_queue = prompt_request_queue
+
+        if output_queue == None:
+            output_queue = prompt_response_queue
+
+        if subscriber == None:
+            subscriber = str(uuid.uuid4())
+
+        self.client = pulsar.Client(
+            pulsar_host,
+            logger=pulsar.ConsoleLogger(log_level),
+        )
+
+        self.producer = self.client.create_producer(
+            topic=input_queue,
+            schema=JsonSchema(PromptRequest),
+            chunking_enabled=True,
+        )
+
+        self.consumer = self.client.subscribe(
+            output_queue, subscriber,
+            schema=JsonSchema(PromptResponse),
+        )
+
+    def request_definitions(self, chunk, timeout=500):
+
+        id = str(uuid.uuid4())
+
+        r = PromptRequest(
+            kind="extract-definitions",
+            chunk=chunk,
+        )
+
+        self.producer.send(r, properties={ "id": id })
+
+        while True:
+
+            msg = self.consumer.receive(timeout_millis=timeout * 1000)
+
+            mid = msg.properties()["id"]
+
+            if mid == id:
+                resp = msg.value().definitions
+                self.consumer.acknowledge(msg)
+                return resp
+
+            # Ignore messages with wrong ID
+            self.consumer.acknowledge(msg)
+
+    def request_relationships(self, chunk, timeout=500):
+
+        id = str(uuid.uuid4())
+
+        r = PromptRequest(
+            kind="extract-relationships",
+            chunk=chunk,
+        )
+
+        self.producer.send(r, properties={ "id": id })
+
+        while True:
+
+            msg = self.consumer.receive(timeout_millis=timeout * 1000)
+
+            mid = msg.properties()["id"]
+
+            if mid == id:
+                resp = msg.value().relationships
+                self.consumer.acknowledge(msg)
+                return resp
+
+            # Ignore messages with wrong ID
+            self.consumer.acknowledge(msg)
+
+    def request_kg_prompt(self, query, kg, timeout=500):
+
+        id = str(uuid.uuid4())
+
+        r = PromptRequest(
+            kind="kg-prompt",
+            query=query,
+            kg=[
+                Fact(s=v[0], p=v[1], o=v[2])
+                for v in kg
+            ],
+        )
+
+        self.producer.send(r, properties={ "id": id })
+
+        while True:
+
+            msg = self.consumer.receive(timeout_millis=timeout * 1000)
+
+            mid = msg.properties()["id"]
+
+            if mid == id:
+                resp = msg.value().answer
+                self.consumer.acknowledge(msg)
+                return resp
+
+            # Ignore messages with wrong ID
+            self.consumer.acknowledge(msg)
+
+    def __del__(self):
+
+        if hasattr(self, "consumer"):
+            self.consumer.close()
+            
+        if hasattr(self, "producer"):
+            self.producer.flush()
+            self.producer.close()
+            
+        self.client.close()
+

trustgraph/schema.py

@@ -176,3 +176,47 @@ graph_rag_response_queue = topic(
 
 ############################################################################
 
+# Prompt services, abstract the prompt generation
+
+class Definition(Record):
+    name = String()
+    definition = String()
+
+class Relationship(Record):
+    s = String()
+    p = String()
+    o = String()
+    o_entity = Boolean()
+
+class Fact(Record):
+    s = String()
+    p = String()
+    o = String()
+
+# extract-definitions:
+#   chunk -> definitions
+# extract-relationships:
+#   chunk -> relationships
+# prompt-rag:
+#   query, triples -> answer
+
+class PromptRequest(Record):
+    kind = String()
+    chunk = String()
+    query = String()
+    kg = Array(Fact())
+
+class PromptResponse(Record):
+    answer = String()
+    definitions = Array(Definition())
+    relationships = Array(Relationship())
+
+prompt_request_queue = topic(
+    'prompt', kind='non-persistent', namespace='request'
+)
+prompt_response_queue = topic(
+    'prompt-response', kind='non-persistent', namespace='response'
+)
+
+############################################################################
+