nostrdvm/tests/dvm_llm_agents.py

import random
import time
from pathlib import Path
import dotenv
from typing import Dict, List
import requests, json

from nostr_sdk import Kind

from nostr_dvm.framework import DVMFramework
from nostr_dvm.tasks.generic_dvm import GenericDVM
from nostr_dvm.utils.admin_utils import AdminConfig
from nostr_dvm.utils.dvmconfig import build_default_config
from nostr_dvm.utils.nip89_utils import NIP89Config, check_and_set_d_tag
from nostr_dvm.utils.outbox_utils import AVOID_OUTBOX_RELAY_LIST

RELAY_LIST = ["wss://nostr.mom",
              "wss://relay.nostrdvm.com",
              "wss://nostr.oxtr.dev",
              #"wss://relay.nostr.net"
              ]

SYNC_DB_RELAY_LIST = ["wss://relay.damus.io",
                      #"wss://relay.primal.net",
                      "wss://nostr.oxtr.dev"]




def playground(announce):
    framework = DVMFramework()

    # An Agent that pretends to be a purple Ostrich
    name = "PurpleOstrichIntelligence"
    identifier = "duckduckchat"  # Chose a unique identifier in order to get a lnaddress
    description = "I'm a funny purple Ostrich"
    picture = "https://image.nostr.build/afd3aaf3fba8e71e1a45611e833967c71ae69f1e1c7a271df4370b5d03feee6d.jpg"
    system_prompt = "You are a funny purple ostrich. Reply in a funny way."
    #purple_ostrich = build_dvm(identifier, name, description, picture, system_prompt, announce)
    #framework.add(purple_ostrich)

    # An Agent that pretends to be a Satoshi Nakamoto
    name = "Nakamoto"
    identifier = "satoshichat"  # Chose a unique identifier in order to get a lnaddress
    description = "If you don't believe me or don't get it, I don't have time to try to convince you, sorry."
    picture = "https://image.nostr.build/34583da23265d955bc09550890029d815c44a1351613b7a76e324c8ad6e1e5a7.jpg"
    system_prompt = "Answer as if you were Satoshi Nakamoto, the inventor of Bitcoin."
    satoshi = build_dvm(identifier, name, description, picture, system_prompt, announce)
    framework.add(satoshi)

    # An Agent that pretends to be a Satoshi Nakamoto

    name = "SamurAI"
    identifier = "samuraichat"  # Chose a unique identifier in order to get a lnaddress
    description = "I follow the way of the Samurai"
    picture = "https://image.nostr.build/c370203fe9fefe0f80b8f72e2fdeee926a22fbce7af13d0013b0071756f1f5e7.jpg"
    system_prompt = "Answer as if you were a wise Samrurai."
    #satoshi = build_dvm(identifier, name, description, picture, system_prompt, announce)
    #framework.add(satoshi)






    framework.run()



def build_dvm(identifier, name, description, picture, system_prompt, announce):
        options = {
            "system_prompt": system_prompt,
            "input": "",
        }

        kind = 5050
        admin_config = AdminConfig()
        admin_config.REBROADCAST_NIP89 = announce
        admin_config.REBROADCAST_NIP65_RELAY_LIST = announce
        admin_config.UPDATE_PROFILE = announce

        dvm_config = build_default_config(identifier)
        dvm_config.KIND = Kind(kind)  # Manually set the Kind Number (see data-vending-machines.org)
        dvm_config.SEND_FEEDBACK_EVENTS = False
        dvm_config.AVOID_OUTBOX_RELAY_LIST = AVOID_OUTBOX_RELAY_LIST
        dvm_config.RELAY_LIST = RELAY_LIST
        dvm_config.SYNC_DB_RELAY_LIST = SYNC_DB_RELAY_LIST

        # Add NIP89
        nip89info = {
            "name": name,
            "picture": picture,
            "about": description,
            "supportsEncryption": True,
            "acceptsNutZaps": dvm_config.ENABLE_NUTZAP,
            "nip90Params": {
            }
        }

        nip89config = NIP89Config()
        nip89config.KIND = Kind(kind)
        nip89config.DTAG = check_and_set_d_tag(identifier, name, dvm_config.PRIVATE_KEY, nip89info["picture"])
        nip89config.CONTENT = json.dumps(nip89info)

        dvm =  GenericDVM(name=name, dvm_config=dvm_config, nip89config=nip89config,
                         admin_config=admin_config, options=options)

        async def process(request_form):
            options = dvm.set_options(request_form)
            result = "Kinda rate limited."
            try:
                class ConversationOver(Exception):
                    """Raised when the conversation limit is reached."""
                    pass

                class ChatModel:
                    """Available models for chat."""
                    claude = "claude-3-haiku-20240307"
                    gpt = "gpt-4o-mini"
                    llama = "meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo"
                    mistral = "mistralai/Mixtral-8x7B-Instruct-v0.1"

                class ChatInstance:
                    def __init__(self, model: str):
                        self.base = "https://duckduckgo.com/duckchat/v1%s"
                        self.vqd: str = requests.get(
                            self.base % "/status",
                            headers={"x-vqd-accept": "1"},
                            timeout=5
                        ).headers["x-vqd-4"]
                        self.model: str = model
                        self.transcript: List[Dict[str, str]] = []

                    def chat(self, message: str) -> str:
                        """
                        Chat with the chosen model. Takes a message and returns the model's response.
                        """
                        self.transcript.append({"role": "user", "content": message})
                        res = requests.post(
                            self.base % "/chat",
                            headers={"x-vqd-4": self.vqd},
                            timeout=10,
                            json={"model": self.model, "messages": self.transcript},
                        )
                        self.vqd = res.headers["x-vqd-4"]

                        out: str = ""
                        for item in (i.removeprefix("data: ") for i in res.text.split("\n\n")):
                            if item.startswith("[DONE]"):
                                if item.endswith("[LIMIT_CONVERSATION]"):
                                    raise ConversationOver
                                break
                            out += json.loads(item).get("message", "").encode("latin-1").decode()

                        self.transcript.append({"role": "assistant", "content": out})
                        return out

                chat = ChatInstance(ChatModel.llama)
                query = ("{role: system, content: " + options["system_prompt"] + "}" + " {role: user, content:" +
                         options["input"] + "}")


                result = chat.chat(query)
                print(result)

            except Exception as e:
                print(e)
            return result

        dvm.process = process  # overwrite the process function with the above one
        return dvm


if __name__ == '__main__':
    env_path = Path('.env')
    if not env_path.is_file():
        with open('.env', 'w') as f:
            print("Writing new .env file")
            f.write('')
    if env_path.is_file():
        print(f'loading environment from {env_path.resolve()}')
        dotenv.load_dotenv(env_path, verbose=True, override=True)
    else:
        raise FileNotFoundError(f'.env file not found at {env_path} ')

    playground(announce=True)