File: //opt/textanalyse/tools/morphology_tool.py
# tools/morphology_tool.py
"""
Agentic Morphology‑Tool für deutsche Wörter.
Dieses Modul implementiert ein komplettes Agno‑Toolkit, das spaCy nutzt,
um Lemma, POS und detaillierte morphologische Merkmale von deutschen
Tokens oder Texten zu extrahieren. Es kann als Tool in Agno‑Agents
eingebunden oder als eigenständige CLI‑Anwendung genutzt werden.
Abhängigkeiten
--------------
pip install agno spacy==3.7.4 de_core_news_sm
"""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import asdict, dataclass
from functools import lru_cache
from typing import Any, Dict, List, Union
from agno.tools import Toolkit
from agno.utils.log import log_debug
try:
import spacy
except Exception as e:
print(f"Warning: spaCy import disabled due to error: {e}")
spacy = None
# --------------------------------------------------------------------------- #
# spaCy‑Pipeline mit Caching laden
# --------------------------------------------------------------------------- #
# @lru_cache(maxsize=1) # auskommentiert zur Fehlerisolation
# def _load_spacy(model: str = "de_core_news_sm") -> "spacy.language.Language":
# """Lädt das gewünschte spaCy‑Modell mit Cacheunterstützung."""
# try:
# return spacy.load(model)
# except OSError:
# from spacy.cli import download
#
# download(model)
# return spacy.load(model)
# Globalinstanz – wird durch LRU‑Cache nur einmal erstellt
# _nlp = _load_spacy()
_nlp = None # spaCy-Pipeline deaktiviert zur Fehlerisolation
# --------------------------------------------------------------------------- #
# Datenstruktur
# --------------------------------------------------------------------------- #
@dataclass(slots=True)
class MorphInfo:
text: str
lemma: str
pos: str
morph: Dict[str, str]
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_json(self) -> str:
return json.dumps(self.to_dict(), ensure_ascii=False)
# --------------------------------------------------------------------------- #
# Hilfsfunktionen
# --------------------------------------------------------------------------- #
def _analyze_token(token: "spacy.tokens.Token") -> Dict[str, Any]:
"""Wandelt ein spaCy‑Token in ein serialisierbares Dict um."""
return {
"text": token.text,
"lemma": token.lemma_,
"pos": token.pos_,
"morph": token.morph.to_dict(),
}
# --------------------------------------------------------------------------- #
# Toolkit‑Implementierung
# --------------------------------------------------------------------------- #
class GermanMorphologyTools(Toolkit):
"""
Toolkit, das deutschsprachige Morphologie‑Analyse via spaCy bereitstellt.
Registrierte Tools (abhängig von __init__‑Parametern):
* analyze_word(word:str) -> str (JSON)
* analyze_text(text:str) -> str (JSON‑Liste)
* analyze_words(words:list)-> str (JSON‑Dict)
"""
def __init__(
self,
analyze_word: bool = True,
analyze_text: bool = True,
analyze_words: bool = True,
model: str = "de_core_news_sm",
**kwargs,
) -> None:
super().__init__(name="german_morphology_tools", **kwargs)
self._model_name = model
if analyze_word:
self.register(self.analyze_word)
if analyze_text:
self.register(self.analyze_text)
if analyze_words:
self.register(self.analyze_words)
# ---------- interne Helfer --------------------------------------------- #
@property
def nlp(self) -> "spacy.language.Language":
"""Stellt sicher, dass das Modell lazy & gecacht geladen wird."""
return _load_spacy(self._model_name)
# ---------- Tool‑Funktionen -------------------------------------------- #
def analyze_word(self, word: str) -> str:
"""Analysiert ein einzelnes Wort und liefert JSON."""
word = word.strip()
if not word:
return json.dumps({"error": "Leerer Eingabestring"}, ensure_ascii=False)
doc = self.nlp(word)
token = doc[0] if doc else None
if token is None:
return json.dumps(
{"error": f"Konnte '{word}' nicht analysieren"}, ensure_ascii=False
)
result = _analyze_token(token)
log_debug(f"Morphologie für '{word}': {result}")
return json.dumps(result, ensure_ascii=False)
def analyze_text(self, text: str) -> str:
"""Tokenisiert Text und gibt eine Liste von Analysen zurück (JSON)."""
text = text.strip()
if not text:
return json.dumps({"error": "Leerer Eingabestring"}, ensure_ascii=False)
doc = self.nlp(text)
results = [_analyze_token(t) for t in doc]
log_debug(f"{len(results)} Tokens in analyse_text verarbeitet.")
return json.dumps(results, ensure_ascii=False)
def analyze_words(self, words: Union[List[str], str]) -> str:
"""
Bulk‑Analyse mehrerer Wörter.
Akzeptiert entweder eine Python‑Liste oder einen JSON‑String, der eine Liste enthält.
"""
if isinstance(words, str):
try:
words = json.loads(words)
except json.JSONDecodeError:
# Whitespace‑getrennte Wortliste
words = [w for w in words.split() if w]
if not isinstance(words, list):
return json.dumps({"error": "Erwarte Liste von Strings"}, ensure_ascii=False)
results = {w: json.loads(self.analyze_word(w)) for w in words}
return json.dumps(results, ensure_ascii=False)
# --------------------------------------------------------------------------- #
# Stub-Funktion für Morphologie-Analyse, damit Agenten nicht abstürzen
def analyze_morphology(text: str) -> str:
"""Dummy: Morphologie-Analyse deaktiviert aufgrund spaCy-Inkompatibilität."""
return json.dumps({"error": "Morphologie-Tool deaktiviert."})
# --------------------------------------------------------------------------- #
# CLI‑Entry‑Point
# --------------------------------------------------------------------------- #
def _main() -> None:
parser = argparse.ArgumentParser(
description="Deutscher Morphologie‑Analyser (spaCy + Agno Toolkit)"
)
grp = parser.add_mutually_exclusive_group(required=True)
grp.add_argument("-w", "--word", help="Ein einzelnes Wort analysieren")
grp.add_argument("-t", "--text", help="Einen Satz/Absatz analysieren")
grp.add_argument(
"-l",
"--list",
help="Kommagetrennte Wortliste analysieren (z. B. \"Haus,Baum,Frau\")",
)
args = parser.parse_args()
tool = GermanMorphologyTools()
if args.word:
print(tool.analyze_word(args.word))
elif args.text:
print(tool.analyze_text(args.text))
elif args.list:
words = [w.strip() for w in args.list.split(",") if w.strip()]
print(tool.analyze_words(words))
else:
parser.print_help(sys.stderr)
if __name__ == "__main__": # pragma: no cover
_main()