62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
import numpy as np
|
|
import requests
|
|
from .emma_mdp import EmmaAction, EmmaContext, EmmaState
|
|
from .emma_identity import get_system_prompt
|
|
|
|
|
|
class OllamaBackend:
|
|
"""
|
|
LLM-backend mot Ollama (llama3.2 eller annen lokal modell).
|
|
Injiserer Emmas identitet i alle system-prompts.
|
|
"""
|
|
|
|
def __init__(self, model: str = "llama3.2", embed_model: str = "nomic-embed-text",
|
|
base_url: str = "http://localhost:11434"):
|
|
self.model = model
|
|
self.embed_model = embed_model
|
|
self.base_url = base_url
|
|
self.system_prompt = get_system_prompt()
|
|
|
|
def embed(self, text: str) -> np.ndarray:
|
|
r = requests.post(f"{self.base_url}/api/embeddings",
|
|
json={"model": self.embed_model, "prompt": text}, timeout=30)
|
|
r.raise_for_status()
|
|
return np.array(r.json()["embedding"], dtype=np.float32)
|
|
|
|
def think(self, ctx: EmmaContext, patterns: list, state: EmmaState) -> str:
|
|
context_hint = "\n".join(str(p.get("action", "")) for p in patterns[:3] if isinstance(p, dict))
|
|
prompt = f"[State: {state.name}]\nPatterns: {context_hint}\nUser: {ctx.userinput}"
|
|
return self._chat(prompt)
|
|
|
|
def compress(self, thoughts: list) -> str:
|
|
joined = "\n".join(thoughts)
|
|
return self._chat(f"Komprimér til ett kort sammendragsspørsmål:\n{joined}")
|
|
|
|
def choose_action(self, ctx: EmmaContext, thought: str, patterns: list, actions: list) -> EmmaAction:
|
|
names = [a.name for a in actions]
|
|
resp = self._chat(f"Velg én handling fra {names} basert på: {thought[:200]}. Svar kun med handlingens navn.")
|
|
for a in actions:
|
|
if a.name.lower() in resp.lower():
|
|
return a
|
|
return EmmaAction.RESPOND
|
|
|
|
def execute_action(self, action: EmmaAction, ctx: EmmaContext) -> dict:
|
|
resp = self._chat(f"[Action: {action.name}] {ctx.userinput}")
|
|
return {"response": resp, "task_completed": True}
|
|
|
|
def _chat(self, prompt: str) -> str:
|
|
r = requests.post(
|
|
f"{self.base_url}/api/chat",
|
|
json={
|
|
"model": self.model,
|
|
"messages": [
|
|
{"role": "system", "content": self.system_prompt},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
"stream": False,
|
|
},
|
|
timeout=60,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()["message"]["content"]
|