my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Mnemosyne Host LLM Backend Registry
#3	===================================
#4	Pluggable adapter for routing Mnemosyne's LLM-backed memory operations
#5	through a host-provided completion endpoint (e.g., Hermes' authenticated
#6	auxiliary client).
#7
#8	Standalone Mnemosyne ignores this registry. When a host registers a backend
#9	and the user opts in via MNEMOSYNE_HOST_LLM_ENABLED=true, both consolidation
#10	(summarize_memories) and structured fact extraction (extract_facts) consult
#11	the backend before falling through to the existing remote/local chain.
#12
#13	The interface is intentionally tiny: one method, prompt-shaped, returning
#14	text-or-None. The caller owns the system prompt and content per task; the
#15	backend just routes to whichever provider/model the host has authenticated.
#16	"""
#17
#18	from __future__ import annotations
#19
#20	from dataclasses import dataclass
#21	from typing import Callable, Optional, Protocol
#22
#23
#24	class LLMBackend(Protocol):
#25	"""A host-provided LLM completion endpoint.
#26
#27	Implementations route a single prompt string through the host's
#28	authenticated provider and return the cleaned text, or None on failure.
#29
#30	The method is named ``complete`` (not ``summarize``) because the same
#31	backend serves both memory consolidation and structured fact extraction;
#32	the caller, not the backend, owns the system prompt.
#33	"""
#34
#35	name: str
#36
#37	def complete(
#38	self,
#39	prompt: str,
#40	*,
#41	max_tokens: int,
#42	temperature: float,
#43	timeout: float,
#44	provider: Optional[str] = None,
#45	model: Optional[str] = None,
#46	) -> Optional[str]:
#47	...
#48
#49
#50	@dataclass
#51	class CallableLLMBackend:
#52	"""Wrap a callable as an :class:`LLMBackend`. Useful for tests and one-off callers."""
#53
#54	name: str
#55	func: Callable[..., Optional[str]]
#56
#57	def complete(
#58	self,
#59	prompt: str,
#60	*,
#61	max_tokens: int,
#62	temperature: float,
#63	timeout: float,
#64	provider: Optional[str] = None,
#65	model: Optional[str] = None,
#66	) -> Optional[str]:
#67	return self.func(
#68	prompt,
#69	max_tokens=max_tokens,
#70	temperature=temperature,
#71	timeout=timeout,
#72	provider=provider,
#73	model=model,
#74	)
#75
#76
#77	_backend: Optional[LLMBackend] = None
#78
#79
#80	def set_host_llm_backend(backend: Optional[LLMBackend]) -> None:
#81	"""Register (or clear) the process-global host LLM backend.
#82
#83	Hosts call this from their initialize/shutdown hooks. Pass ``None``
#84	to unregister.
#85	"""
#86	global _backend
#87	_backend = backend
#88
#89
#90	def get_host_llm_backend() -> Optional[LLMBackend]:
#91	"""Return the registered host LLM backend, or None."""
#92	return _backend
#93
#94
#95	def call_host_llm(
#96	prompt: str,
#97	*,
#98	max_tokens: int,
#99	temperature: float = 0.3,
#100	timeout: float = 15.0,
#101	provider: Optional[str] = None,
#102	model: Optional[str] = None,
#103	) -> Optional[str]:
#104	"""Convenience wrapper: call the registered backend if any, swallow failures.
#105
#106	Returns ``None`` when no backend is registered or the backend raises.
#107	Logging is intentionally minimal here; callers that need provenance
#108	should log around the call site (and never log the prompt itself).
#109	"""
#110	backend = get_host_llm_backend()
#111	if backend is None:
#112	return None
#113	try:
#114	return backend.complete(
#115	prompt,
#116	max_tokens=max_tokens,
#117	temperature=temperature,
#118	timeout=timeout,
#119	provider=provider,
#120	model=model,
#121	)
#122	except Exception:
#123	return None
#124

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public