Simple chat.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install 'cosmonapse[receptor]' python-dotenv
# a Hugging Face token with read scope, in cosmonapse-examples/.env
$ echo "HF_TOKEN=hf_xxxxxxxxxxxxxxxx" >> cosmonapse-examples/.env

$ cd cosmonapse-examples/18-simple-chat-hf
$ python brain.py
$ open http://127.0.0.1:8000

18-simple-chat-hf/
  neurons/assistant.py   Axon.huggingface(...) + a hook that folds chat history in
  receptors/chat.py      the ChatReceptor
  brain.py               worker + edge, run_brain

import os
from pathlib import Path

from dotenv import load_dotenv

from cosmonapse import Axon

load_dotenv()                                                  # ./.env
load_dotenv(Path(__file__).resolve().parents[2] / ".env")      # cosmonapse-examples/.env

AXON = Axon.huggingface(
    neuron_id="assistant",
    endpoint="https://router.huggingface.co",
    model=os.environ.get("MODEL", "meta-llama/Llama-3.1-8B-Instruct"),
    api_key="hf_dummy_token",
    use_chat_api=True,
    max_new_tokens=400,
    capabilities=["chat", "text-generation"],
)


@AXON.before_task
def fold_history(input_data: dict) -> dict:
    """ChatReceptor rides prior turns in as history=[{role, content}, ...];
    the HF Neuron only ever reads prompt/messages, so fold them in here
    before the Neuron runs."""
    turns = input_data.get("history") or []
    if not turns:
        return input_data
    transcript = "\n".join(f"{t['role']}: {t['content']}" for t in turns)
    prompt = input_data.get("prompt", "")
    return {**input_data, "prompt": f"{transcript}\nuser: {prompt}"}

from cosmonapse import ChatReceptor

RECEPTOR = ChatReceptor(
    neuron="assistant",          # addresses the Neuron directly, no routing needed
    input_key="prompt",          # matches what Axon.huggingface's Neuron reads
    title="Cosmonapse HF Chat",
    greeting="Ask me anything.",
    history_turns=8,
    timeout_s=120.0,
    # host="127.0.0.1", port=8000 are the defaults - open http://127.0.0.1:8000
)

from __future__ import annotations

import asyncio
import contextlib
import os
import signal

from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain

from neurons import assistant
from receptors import chat

NAMESPACE = "simple-chat-hf"
SYNAPSE_URL = os.environ.get("SYNAPSE_URL", "")   # unset -> in-process bus


def build_worker(synapse) -> Dendrite:
    """Hosts the assistant Axon - services the TASK a turn dispatches."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="worker", role="worker")
    node.attach_axon(assistant.AXON)
    return node


def build_edge(synapse) -> Dendrite:
    """Hosts the chat Receptor - the only node allowed to dispatch."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="edge", role="orchestrator", heartbeat_s=0)
    node.attach_receptor(chat.RECEPTOR)
    return node

# terminal 1 - a dev synapse (TCP, single host)
$ cosmo synapse start memory --namespace=simple-chat-hf

# terminal 2 - the same brain, over that synapse
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python brain.py

# terminal 3 - Prism, the live view (http://127.0.0.1:7071)
$ cosmo prism --url=cosmo://127.0.0.1:7070 -n simple-chat-hf

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python brain.py
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py