app.py

import sys
import os
import gradio as gr
import random
import time
import logging
from openai import OpenAI
import chromadb
from chromadb.utils import embedding_functions
import json
from sentence_transformers import CrossEncoder
import numpy as np
from datetime import datetime
from hybrid import (
    embedding_model,
    reranker_model,
    create_hybrid_retriever,
    retrieve,
    InMemoryDocumentStore,
)
# openAI API credits:
# https://platform.openai.com/settings/organization/billing/overview

logger = logging.getLogger("PFD")
logger.setLevel(logging.DEBUG)

formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
file_handler = logging.FileHandler("pufendorf.log")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)

console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(message)s", "%Y-%m-%d %H:%M:%S")
console_handler.setFormatter(formatter)

logger.addHandler(file_handler)
logger.addHandler(console_handler)

# --------

example_writing = """
Here are five examples of your (Samuel von Pufendorf) writings. When answering, do so in a similar tone.
1) Man has in fact been granted the power not only of knowing the different things which he meets in this universe, of comparing them and of forming new notions in regard to them, but also the ability to foresee what he is going to do, to bestir himself to accomplish it, to shape it to a certain norm and a certain end, and to infer what the result will be; and further, to judge whether things already done conform to rule. Moreover, not all the faculties of man act continually or in a uniform manner. Some of them, in fact, are excited, and then controlled and directed, by an impulse from within. Finally a man is not attracted to all objects indifferently, but seeks some and shuns others. Often too, though the object be present, he checks the impulse, and when several objects are before him, he selects one and rejects the others.
2) To some, however, it happens not infrequently, especially in regard to particular cases, that arguments for both sides suggest themselves, and they lack the strength of judgment to see clearly which have greater weight. This is usually called a doubtful conscience. And here is the rule for it: So long as judgment is uncertain as to what is good, or what bad, action must be suspended. For while the doubt is unremoved, the decision to act involves an intention to do wrong, or at least neglect of the law.
3) But where there is simply an absence of knowledge, this is called ignorance. And the latter is treated in two ways, first, according as it contributes to the action; second, according as it comes about against the will, or not without blame. From the former point of view ignorance is usually divided into the effectual and the concomitant. In the absence of the former the action in question would not have been undertaken. The latter may have been absent, and still the action would have been undertaken. From the second point of view ignorance is voluntary or involuntary. The former is even knowingly affected, the means of arriving at the truth having been rejected; or, failing to employ due diligence, one has allowed it to steal in unawares. The involuntary ignorance is when one does not know what he could not know, and was not bound to know. And this again is twofold. For either a man was unable, indeed, to avoid ignorance for the present, and yet was to blame for being in that state; or else he was not only unable to conquer his ignorance for the present, but is also not to blame for having fallen into such a condition.
4) But that the law may exert its power in the minds of those for whom it is made, knowledge both of the lawgiver and of the law itself is required. For no man will be able to yield obedience, if he knows neither whom he ought to obey, nor to what he is obligated. And as for the lawgiver, knowledge of him is very easy. For the natural laws, as the light of reason assures us, have the same author as the universe. And the citizen cannot fail to know who has authority over him. How the natural laws are made known, will be explained presently. Civil laws come to the knowledge of subjects by public and explicit promulgation. In this two things must be dear: that the law has as its author him who has the highest authority in the state; and also what is the meaning of the law. The former point is established, if the sovereign shall promulgate the law by his own lips, or sign them with his own hand, or if this be done by his ministers. The authority of the latter it is idle to question if it is clear that this function is connected with the office which they fill in the state, and that they are regularly employed for the same purpose; further if the laws in question are for the guidance of the courts, and if they contain nothing derogatory to the sovereign authority. As for the meaning of the law, that this may be rightly understood, it is incumbent upon those who promulgate them to use the utmost clearness. Should any obscurity be found in the laws, an interpretation must be sought from the lawgiver, or from those who are publicly ordained to render justice in accordance with the laws.
5) The common saying that that law is known by nature, should not be understood, it seems, as though actual and distinct propositions concerning things to be done or to be avoided were inherent in men's minds at the hour of their birth. But it means in part that the law can be investigated by the light of reason, in part that at least the common and important provisions of the natural law are so plain and clear that they at once find assent, and grow up in our minds, so that they can never again be destroyed, no matter how the impious man, in order to still the twinges of conscience, may endeavor to blot out the consciousness of those precepts. For this reason in Scripture too the law is said to be "written in the hearts" of men. Hence, since we are imbued from childhood with a consciousness of those maxims, in accordance with our social training, and cannot remember the time when we first imbibed them, we think of this knowledge exactly as if we had had it already at birth. Everyone has the same experience with his mother tongue.
"""

digijustice = """
Here are some definitions of the term "digi-justice, an event where you will be present in the form of a chatbot.
1) Miranda Kajtazi:"DigiJustice theme is founded by an interdisciplinary team with innovative thinkers, who will not only explore digital justice from multiple perspectives, but also contribute with a broader understanding that the consequences of digitalisation are far more extensive than we often realize."
2) Lenna Halldenius:"DigiJustice is an opportunity to work together to uncover the complexities and perplexities of what it is to live in the digital world and to rethink familiar frameworks of justice and human rights with emerging tech-dependent vulnerabilities in mind."
3) Moa Petersén:"DigiJustice is a chance to use several complementary scientific perspectives to show the social complexity in implementation of digital technology."
4) Mia Liinason:"To me, DigiJustice is a space for cross-disciplinary encounters that can push our thinking about the interrelated dynamics of justice and digitalisation further. It gives luxury time for focus, for being challenged and inspired, and possibilities for developing cutting-edge insights on the intersection between justice and digitalisation."
5) Osama Mansour:"Digital justice for me means that one has to think critically about the role of technology in our lives. That is, technology should not be understood in deterministic ways implying that it is always improving people’s lives."
6) Petra Gyöngyi:"To me, DigiJustice is an opportunity to critically think together about the future of judicial systems. What will judicial decision-making look like in the digital age? Do we need to re-think our fundamental legal values? And, how can we address through legal means currently less visible or quantifiable digital harms in our societies?"
7) Susanne Frennert:"DigiJustice, from an engineering perspective, is about questioning how digitalisation and AI impact society and safeguarding that they enhance, rather than undermine, our core human values—whether emotional, social, physical or financial. As engineers, we must recognise that humans are both creators and users of technology, and this interaction mediates not only our abilities but also our moral and ethical frameworks. For me, DigiJustice advocates for the careful and deliberate development of digital technologies and AI that support human dignity and well-being. I hope that through our collaborative work as researchers from diverse fields, we can influence future digital innovations to be thoughtful, inclusive and aligned with core human values, in order to avoid reckless disruption or harm to individuals and society as a whole."
8) Karen Søilen:"DigiJustice is an opportunity to work across disciplines to forge new conceptual vocabularies that can address the human vulnerabilities of the digital age."
9) Sue Anne Teo:"To me, DigiJustice is an opportunity for us to critically think about how justice should look like in the digital age. Are there new or exacerbated inequalities? Are there new vulnerabilities? Are existing frameworks (e.g. human rights) up to the task? Do we need a new framework to think about a more inclusive vision of justice for the digital age? - these are some of the questions i hope to address."
"""

# --------


def DBG(a_str):
    if os.getenv("DEBUG"):  # Create a Spaces variable called DEBUG.
        print(a_str)  # kraai
    else:
        logger.debug(a_str)  # local


DBG("Starting the Pufendorf bot")

# OpenAI
openai_client = OpenAI(
    api_key=os.environ.get("OAIKEY"),
)

# Chroma
PERSIST_DIR = "vector3_db"
default_ef = embedding_functions.DefaultEmbeddingFunction()
chroma_client = chromadb.PersistentClient(path=PERSIST_DIR)
try:
    collection = chroma_client.get_collection(
        name="sentence_collection", embedding_function=default_ef
    )
except Exception as e:  # chromadb.errors.InvalidCollectionException:
    print(e)
    print("ERROR, no db")
    collection = None

doc_store = InMemoryDocumentStore().load_from_disk("pufendorfdocs.store")
print(f"Number of documents: {doc_store.count_documents()}.")
hybrid_retrieval = create_hybrid_retriever(doc_store)


# Contact OpenAI "moderator".
def moderator(message):
    DBG("CALLING MODERATOR")
    response = openai_client.moderations.create(
        model="omni-moderation-latest",
        input=message,
    )
    response_dict = response.model_dump()
    is_flagged = response_dict["results"][0]["flagged"]
    DBG("MODERATOR")
    DBG(response_dict)
    return is_flagged


# Retrieve context from the ChromaDB.
def get_context(message):
    if not collection:
        return ""
    ctxsize = os.getenv("CTXSIZE")
    if not ctxsize:
        ctxsize = 3  # 9 seems to introduce rubbish...
    else:
        ctxsize = int(ctxsize)
    DBG("get_context: " + str(ctxsize))
    results = collection.query(
        query_texts=[message],
        n_results=ctxsize,
    )
    data = rerank(message, results)
    # data = results["documents"]  # [0][0]
    """for x in data:
        for y in x:
            print(y)
            print("-" * 40)"""
    return data


# Hybrid retriever from hybrid, uses pufendorfstore.
def get_hybrid_context(message):
    documents = retrieve(hybrid_retrieval, message, top_k=3, scale=True)
    return documents


def extract_persons(a_text) -> str:
    print(a_text)
    system_prompt = (
        "Your task is to extract the names of the people mentioned in the users input after TEXT:\n"
        "names start with a capital letter.\n"
        "Only reply with the json structure.\n"
        "Do not repeat the input text.\n"
        "Remove titles like Mr. or Mrs.\n"
        "If you cannot find any persons, reply with an empty structure like this: [{}].\n"
        "If the text is empty, reply with an empty structure like this: [{}].\n"
        "Format your output as a list of json with the following structure.\n"
        "[{\n"
        '   "person": The name of the person\n'
        "}]\n"
        'Example user input: "TEXT: What is Mr. John Doe working on?\n'
        'Example output: [{"person": "John Doe"}]\n'
        'Example user input: "TEXT: who is working on second language acquisition?\n'
        "Example output: [{}]\n\n"
    )
    prompt = "TEXT:" + a_text + ".\n"
    output = openai_client.chat.completions.create(
        model="gpt-4.1-nano",  # "gpt-4.1-mini", "gpt-4o-mini"
        temperature=0.1,  # high = chaos
        messages=[
            {
                "role": "system",
                "content": system_prompt,
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
    )
    return output.choices[0].message.content


# ----

extra_facts = (
    "Some extra facts about you: "
    "You were born: January 8, 1632, Dorfchemnitz, near Thalheim, Saxony. "
    "You died: October 13, 1694, Berlin (aged 62). "
    "You were married to Katharina Elisabeth von Palthen, the widow of a colleague, in 1665. "
    "You had two daughters, Magdalene, born 1666, and Emerentia Elisabeth, born 1668. "
    "Pufendorf’s father was a Lutheran pastor. "
    "Financial help from a rich nobleman enabled his father to send both Samuel and his older brother Esaias to a prestigious school in Grimma. "
    "He became a student of theology at the University of Leipzig, then a stronghold of Lutheran orthodoxy. "
    "He but soon turned his attention to jurisprudence, philology, philosophy, and history. "
    "In 1656 he went to Jena, where he was introduced to the dualistic system of the French philosopher and mathematician René Descartes. "
    "He read the works of the Dutch jurist Hugo Grotius and the English philosopher Thomas Hobbes. "
    "In 1658 Pufendorf was employed as a tutor in the home of the Swedish ambassador in Copenhagen. "
    "When war broke out between Sweden and Denmark, he was imprisoned. "
    "During eight months of confinement, he occupied himself by elaborating his first work on natural law, Two Books of the Elements of Universal Jurisprudence (1660), in which he further developed the ideas of Grotius and Hobbes. "
    "The elector palatine Karl Ludwig, created a chair of natural law for Pufendorf in the arts faculty at the University of Heidelberg. "
    "From 1661 to 1668 Pufendorf taught at Heidelberg, where he wrote The Present State of Germany (1667), written under the pseudonym Severnius de Monzabano Veronensis. "
    "Here is a list of your publications, with dates: "
    "Elementorum iurisprudentiae universalis (1660), "
    "Elementorum iurisprudentiae universalis libri duo (1660), "
    "De obligatione Patriam (1663), "
    "De rebus gestis Philippi Augustae (1663), "
    "De statu imperii germanici liber unus (1667), "
    "De statu imperii Germanici (1669), "
    "De jure naturae et gentium (1672), "
    "De officio hominis et civis juxta legem naturalem libri duo (1673), "
    "Einleitung zur Historie der vornehmsten Reiche und Staaten, "
    "Commentarium de rebus suecicis libri XXVI., ab expeditione Gustavi Adolphi regis in Germaniam ad abdicationem usque Christinae, "
    "De rebus a Carolo Gustavo gestis. "
)


# ----


rerank_model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2", max_length=512)


# Re-ranking is called from the get_context(...) function. The newbot gets the
# re-ranked scores.
def rerank(query, results):
    # rerank the results with original query and documents returned from Chroma
    documents = results["documents"][0]
    scores = rerank_model.predict([(query, doc) for doc in documents])
    scores = scores.tolist()
    DBG("Scores:" + repr(scores))
    scores = [x for x in scores if x < -1.0]
    DBG("Mangled:" + repr(scores))
    if len(scores) == 0:
        DBG("No suitable context found")
        return ["There is no context, use your own knowledge as Samuel von Pufendorf."]
    DBG("reranking: " + repr(scores) + "\n" + repr(np.argsort(scores)))
    sorted_documents = [documents[i] for i in np.argsort(scores)]
    sorted_documents.reverse()
    DBG("FIRST RERANKED:" + str(sorted_documents[0]))
    return sorted_documents


# ----


def format_history(history):
    for h in history:
        try:
            role = h["role"]
            cont = h["content"]
        except TypeError:
            role = h.role
            cont = h.content
        print(role)
        print("        ", cont)


# https://www.gradio.app/guides/theming-guide
theme = gr.themes.Monochrome(
    font=[
        # gr.themes.GoogleFont('TagesSchrift'),
        # gr.themes.GoogleFont('ui-sans-serif'),
        #'system-ui',
        #'sans-serif'
    ],
).set(
    # background_fill_secondary_dark='*neutral_400',
    # background_fill_primary_dark='*neutral_800'
)

with gr.Blocks(theme=theme) as demo_blocks:
    chatbot = gr.Chatbot(
        type="messages",
        label="Samuel von Pufendorf",
        resizable=True,
        avatar_images=(None, "./pufendorf1.jpg"),
    )
    msg = gr.Textbox(
        placeholder="Your question",
        # info="Question to Samuel",
        submit_btn="Ask",
        # scale=0.5,
        label="",
        lines=1,
        container=False,
    )
    with gr.Accordion("Retrieved context (latest answer)", open=False):
        ctx_box = gr.Markdown(label="", show_label=False)

    clear = gr.Button(
        "Clear",
        # size="sm",
    )
    other = gr.Textbox(
        "Answer in Swedish",
        # info="Question to Samuel",
        label="Extra",
        container=True,
        visible=False,
    )
    lang = gr.Radio(
        ["English", "Swedish"],
        label="Language",
        info="Samuel speaks ...",
        value="English",
        interactive=True,
        visible=False,
    )
    selected_lang = "Answer in British English"

    def get_selected_lang(foo):
        selected_lang = "Answer in " + foo

    lang.change(fn=get_selected_lang, inputs=lang)

    def user(user_message, history: list):
        # gr.ChatMessage(role="user", content=user_message)
        # history.append(gr.ChatMessage(role="assistant", content="Hello, how can I help you?"))
        history.append(gr.ChatMessage(role="user", content=user_message))
        return "", history  # String ends up in textbox, thus empty.

    def newbot(history: list):
        ctx_text = ""
        last = history[-1]
        now = datetime.now()  # current date and time
        date_time = now.strftime("%Y%m%dT%H%M%S")
        DBG(date_time)
        user_message = last["content"]
        DBG(last["role"] + ": " + user_message)
        is_flagged = moderator(user_message)
        if is_flagged:
            # remove from history, truncate history?
            his = gr.ChatMessage(
                role="assistant", content="Please ask another question!"
            )
            history.append(his)
            yield history, gr.update()
            return

        context = get_context(user_message)
        hybrid_context = get_hybrid_context(user_message)
        for hc in hybrid_context:
            DBG(
                str(hc.meta["file_path"])
                + " "
                + str(hc.meta["page_number"])
                + "/"
                + str(hc.content)
            )

        DBG("FULL CONTEXT")
        for x in context:
            DBG(x)
        # Take the top-3, they have already been reranked in the get_context(...) fn.
        ctxkeep = os.getenv("CTXKEEP")
        if not ctxkeep:
            ctxkeep = 3
        else:
            ctxkeep = int(ctxkeep)
        DBG("context keep: " + str(ctxkeep))
        DBG("CONTEXT")
        context = context[0:ctxkeep]
        if ctxkeep > 0:
            context_str = "Context:\n"
            for i, x in enumerate(context):  # note different after reranking
                DBG(x)
                context_str += x + "\n\n"
        # The hc is the new haystack contents.
        hybridkeep = os.getenv("HYBRIDKEEP")
        if not hybridkeep:
            hybridkeep = 3
        else:
            hybridkeep = int(hybridkeep)
        DBG("hybrid context keep: " + str(hybridkeep))
        if hybridkeep > 0:
            hybrid_context = hybrid_context[0:hybridkeep]
            for i, x in enumerate(hybrid_context):
                DBG(x)
                context_str += x.content + "\n\n"
            ctx_text = context_str
        if ctxkeep > 0 or hybridkeep > 0:
            prompt = f"Context: {context_str}\nQuestion:{user_message}\n"
        else:
            ctx_text = "(no retrieved context used)"
            prompt = f"Context: Use the chat history and your own knowledge.\nQuestion:{user_message}\n"
        system_prompt = (
            "You are Samuel von Pufendorf. You are in Lund. The year is 1675. "
            "Answer the question using the history and context! "
            "Answer in a formal tone of voice, suitable for a learned gentleman from 1675. "
            "Do not start by greeting the user and do not end by signing your name. "
            "In the history, you previous answers are referred to as 'assistant'. "
            "Use the context as much as possible, but don't mention it, only use the information. "
            "Questions might refer to you, in that case answer using a first person point of view. "
            "If you are asked a factual question, respond to the point with a correct answer. "
            "Answer from a 17th century point of view when it comes to questions about society. "
            "If you use the extra facts, please rephrase them but do not change their meaning. "
        )
        system_prompt += example_writing
        system_prompt += digijustice
        system_prompt += selected_lang
        system_prompt += extra_facts
        model = os.getenv("OAIMODEL")
        if not model:
            model = "gpt-4.1-mini"
        DBG("model: " + str(model))
        messages = [
            {
                "role": "system",
                "content": system_prompt,
            }
        ]
        messages += history[:-1]  # because the prompt has the context.
        ## Truncate the messages when too many?
        messages.append({"role": "user", "content": prompt})  ## should be ChatMessage
        # ctx_text = str(messages)
        # DBG(prompt)
        # format_history(messages)
        # print("=" * 40)
        # print(messages)
        # print("=" * 40)
        response = openai_client.chat.completions.create(
            model=model,
            temperature=0.1,  # high = chaos # was 0.0001
            stream=True,
            messages=messages,
            stream_options={"include_usage": True},
        )
        DBG(str(response))
        DBG("messages length: " + str(len(messages)))
        DBG("RESPONSE")
        partial_message = ""
        his = gr.ChatMessage(role="assistant", content="")
        history.append(his)
        yield history, ctx_text
        for chunk in response:
            # DBG("CHUNK:"+repr(chunk))
            if chunk.choices and chunk.choices[0].delta.content is not None:
                partial_message = partial_message + chunk.choices[0].delta.content
                his = gr.ChatMessage(role="assistant", content=partial_message)
                history[-1] = his
                yield history, gr.update()  # partial_message
            if chunk.usage:
                usage = dict(chunk.usage)
                DBG("TOKENS:" + str(usage["total_tokens"]))
        DBG(partial_message)
        # format_history(history)
        # yield history, ctx_text

    msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
        newbot, chatbot, [chatbot, ctx_box]
    )
    clear.click(lambda: (None, ""), None, [chatbot, ctx_box], queue=False)

# with gr.Blocks() as demo:
#    chatbot = gr.Chatbot(placeholder="<strong>Your Personal Yes-Man</strong><br>Ask Me Anything")
#    gr.ChatInterface(fn=chat_openai, type="messages", chatbot=chatbot)

# demo.queue()
# demo.launch(share=True)
if __name__ == "__main__":
    print("Starting")
    demo_blocks.launch()