wuenlp_tools.models.sentiment.SentimentPromptingLLM

  1from dataclasses import dataclass, field
  2import json
  3from typing import Type
  4
  5from loguru import logger
  6from pydantic import BaseModel
  7from tqdm.auto import tqdm
  8
  9from wuenlp_tools.keys import OPENAI_API_KEY
 10from wuenlp_tools.pipeline import PipelineStep, PipelineProcessor
 11from wuenlp_tools.utils.prompting import LLMArchitecture, default_llm, LLM
 12from wuenlp.impl.UIMANLPStructs import UIMADocument, UIMAAnnotation, UIMASpan, UIMASystemScene
 13
 14
 15class FloatSentimentOutput(BaseModel):
 16    sentiment: float
 17
 18
 19DEFAULT_BASE_LLM_SYSTEM_PROMPT = (
 20    "You are given a piece of text from a novel. You should determine the sentiment of the text on a scale "
 21    "from -1 to 1, depending on how well the overall situation described in the text is. Return a valid json "
 22    "object of the required format."
 23)
 24
 25
 26@dataclass(frozen=True)
 27class BaseLLMSentimentConfig:
 28    model: LLMArchitecture = field(default_factory=lambda: default_llm)
 29    system_prompt: str = DEFAULT_BASE_LLM_SYSTEM_PROMPT
 30    base_url: str = "https://ollama.professor-x.de/v1/"
 31    feature_key: str = "llama_sentiment"
 32
 33
 34class BaseLLMSentimentModel(object):
 35    model: LLM
 36
 37    def __init__(self, config: BaseLLMSentimentConfig | None = None, output_format: type[BaseModel] = FloatSentimentOutput):
 38        cfg = config or BaseLLMSentimentConfig()
 39        self.model = LLM(
 40            model=cfg.model,
 41            system_prompt=cfg.system_prompt,
 42            output_format=output_format,
 43            base_url=cfg.base_url,
 44        )
 45
 46    @staticmethod
 47    def normalize_sentiment(sentiment):
 48        return min(1, max(-1, sentiment))
 49
 50    def __call__(self, text, output_type: Type = float):
 51        assert output_type in (float,), "Only float output_type is supported for GPTSentimentModel"
 52        response = self.model(text)
 53        if hasattr(response, "sentiment"):
 54            return self.normalize_sentiment(response.sentiment)
 55
 56        if isinstance(response, str):
 57            try:
 58                parsed = json.loads(response)
 59                if isinstance(parsed, dict) and "sentiment" in parsed:
 60                    return self.normalize_sentiment(float(parsed["sentiment"]))
 61            except Exception:
 62                try:
 63                    return self.normalize_sentiment(float(response))
 64                except Exception:
 65                    pass
 66
 67        raise ValueError(f"Unexpected sentiment response type: {type(response).__name__}")
 68
 69
 70def base_llm_sentiment(
 71        doc: UIMADocument,
 72        unit_type: Type[UIMASpan],
 73        overwrite: bool = False,
 74        config: BaseLLMSentimentConfig | None = None,
 75) -> UIMADocument:
 76    cfg = config or BaseLLMSentimentConfig()
 77    if not overwrite and any(
 78            cfg.feature_key in segment.additional_features for segment in doc._get_annos_of_type(unit_type)):
 79        logger.warning(
 80            f"Skipping {unit_type} sentiment prediction because some segments already have a sentiment prediction")
 81        return doc
 82    model = BaseLLMSentimentModel(config=cfg)
 83    segments = doc._get_annos_of_type(unit_type)
 84    progress_desc = f"LLM sentiment ({unit_type.__name__})"
 85    for segment in tqdm(segments, desc=progress_desc, unit="segment"):
 86        try:
 87            segment.additional_features[cfg.feature_key] = model(segment.text)
 88        except Exception as exc:
 89            logger.warning(f"Falling back to neutral sentiment due to model error: {exc}")
 90            segment.additional_features[cfg.feature_key] = 0.0
 91    return doc
 92
 93
 94class BaseLLMSentimentProcessor(PipelineProcessor):
 95    def __init__(self, config: BaseLLMSentimentConfig | None = None):
 96        self.config = config or BaseLLMSentimentConfig()
 97
 98    def __call__(self, doc: UIMADocument, unit_type: Type[UIMASpan], overwrite: bool = False, **kwargs) -> UIMADocument:
 99        return base_llm_sentiment(doc, unit_type, overwrite=overwrite, config=self.config)
100
101
102def BaseLLMSentimentStep(
103        config: BaseLLMSentimentConfig | None = None,
104        unit_type: Type[UIMAAnnotation] = UIMASystemScene,
105        name: str = "LLM Sentiment",
106) -> PipelineStep[BaseLLMSentimentProcessor]:
107    cfg = config or BaseLLMSentimentConfig()
108    return PipelineStep[BaseLLMSentimentProcessor](
109        name,
110        BaseLLMSentimentProcessor(config=cfg),
111        unit_type=unit_type,
112        added_additional_features=[cfg.feature_key],
113        requires_api_key=OPENAI_API_KEY,
114        requires_paid_api_requests=True,
115    )
116
117
118BaseLLMSentiment = BaseLLMSentimentStep()