wuenlp_tools.models.suspense.prompt_llm
1from typing import Literal, Type 2 3from loguru import logger 4from openai import NOT_GIVEN 5from pydantic import BaseModel 6from wuenlp import UIMADocument 7from wuenlp.impl.UIMANLPStructs import UIMASpan, UIMASystemScene 8from wuenlp.impl.uima.extensions.danger import DangerDocument, DangerMixin 9 10from wuenlp_tools.keys import OPENAI_API_KEY 11from wuenlp_tools.pipeline import PipelineProcessor 12from wuenlp_tools.models.suspense import AnnoType 13from wuenlp_tools.pipeline import PipelineStep 14from wuenlp_tools.utils.prompting import LLMArchitecture, LLM, YesNoLLM, default_llm 15 16 17class Response(BaseModel): 18 answer: str 19 20 21class FineGrained(Response): 22 answer: list[Literal[ 23 "DangerousSituationDuel", "DangerousSituationAbduction", "DangerousSituationNatural", "DangerousSituationSupernatural", "DangerousSituationAmbush", "DangerousSituationHitchcock", "DangerousSituationOther", "NoDanger"]] 24 25 26class FineGrainedReason(Response): 27 answer: list[Literal[ 28 "DangerousSituationDuel", "DangerousSituationAbduction", "DangerousSituationNatural", "DangerousSituationSupernatural", "DangerousSituationAmbush", "DangerousSituationHitchcock", "DangerousSituationOther", "NoDanger"]] 29 reason: str 30 31 32class YesNoReason(Response): 33 answer: bool 34 reason: str 35 36 37class YesNo(Response): 38 answer: bool 39 40 41danger_classes = sorted(( 42 "DangerousSituationDuel", "DangerousSituationAbduction", "DangerousSituationNatural", 43 "DangerousSituationSupernatural", 44 "DangerousSituationAmbush", "DangerousSituationHitchcock", "DangerousSituationOther")) 45 46 47def get_labels(response_lower): 48 labels = danger_classes 49 answers = [] 50 for label in labels: 51 if label.lower() in response_lower: 52 answers.append(label) 53 return answers 54 55 56class FineGrainedLLM(LLM): 57 def __init__(self, model: LLMArchitecture, system_prompt: str, post_prompt: str = None, cache_maxsize: int = 0, 58 seed=NOT_GIVEN, max_tries=3, reason: bool = False): 59 if model.openai: 60 super().__init__(model, system_prompt, post_prompt, cache_maxsize, seed=seed, 61 output_format=FineGrainedReason if reason else FineGrained) 62 else: 63 if not post_prompt or "yes" not in post_prompt: 64 logger.warning("Post prompt does not contain 'yes'. Adding instruction.") 65 post_prompt = post_prompt + ("Answer with the classes that are present.") \ 66 if not reason else "Answer with a list of the classes that are present followed by a reason" 67 super(FineGrainedLLM, self).__init__(model, system_prompt, 68 post_prompt, 69 cache_maxsize, 70 seed=seed, json=False) 71 self.max_tries = max_tries 72 self.reason = reason 73 74 def __call__(self, prompt) -> (bool, str): 75 if self.model.openai: 76 response = super().__call__(prompt) 77 return response 78 tries = 0 79 while tries < self.max_tries: 80 response = super().__call__(prompt) 81 response_lower = response.lower().strip("*").strip() 82 83 label_list = get_labels(response_lower) 84 if self.reason: 85 return FineGrainedReason(answer=label_list, reason=response) 86 else: 87 return FineGrained(answer=label_list) 88 tries += 1 89 raise ValueError(f"Could not get a yes/no answer after {self.max_tries} tries.") 90 91 92def annotate_doc(doc: UIMADocument, annotation_type: str, 93 annotation_unit: Type[UIMASpan] = UIMASystemScene, 94 model: LLMArchitecture = default_llm) -> DangerDocument: 95 if annotation_type == "Fear": 96 types = ("FearDescription",) 97 system_prompt = """The task is to detect whether there is fear in the current situation in the given text 98 unit. A dangerous situation is not enough. The situation must describe concrete fear.""" 99 100 elif annotation_type == "Danger": 101 types = ("DangerousSituation",) 102 system_prompt = """The task is to detect whether there is a concrete danger at the current time and location 103 in the given text unit. A suggestion, threat or potentially dangerous situation is not enough. Fear of a 104 situation is not enough. The situation must be dangerous and concrete.""" 105 else: 106 raise ValueError(f"Unknown annotation type: {annotation_type}") 107 108 llm = YesNoLLM(model, system_prompt, 109 post_prompt="Is there a dangerous situation as described above in this text?", seed=42, 110 reason=True, max_tries=2, 111 retry_prompt_add=" Note that the text may be empty. In this case, answer 'No, because the text is empty.' Do not include anything else in your response.") 112 113 doc: DangerDocument = DangerDocument.from_other_doc(doc) 114 115 units = doc._get_annos_of_type(annotation_unit) 116 117 for unit in units: 118 response = llm(unit.text) 119 120 has_dangerous_situation = response.answer if annotation_type == 'Danger' else None 121 has_fear_description = response.answer if annotation_type == 'Fear' else None 122 has_other_suspense = None 123 124 unit.additional_features[ 125 f"{annotation_type.lower()}_score"] = response.answer 126 unit.additional_features[f"{annotation_type.lower()}_reason"] = response.reason 127 128 doc.create_system_danger_paragraph(begin=unit.begin, end=unit.end, 129 has_dangerous_situation=has_dangerous_situation, 130 has_fear_description=has_fear_description, 131 has_other_suspense=has_other_suspense, add_to_document=True) 132 133 return doc 134 135 136class PromptingDangerProcessor(PipelineProcessor): 137 def __init__(self, anno_type: AnnoType, model: LLMArchitecture = default_llm): 138 self.anno_type = anno_type 139 self.model = model 140 141 def __call__(self, doc: UIMADocument, unit_type: Type[UIMASpan] = UIMASystemScene, 142 overwrite: bool = False, **kwargs): 143 return annotate_doc(doc, annotation_type=self.anno_type, annotation_unit=unit_type, model=self.model) 144 145 146PromptingDangerAnnotator = PipelineStep("PromptingDangerAnnotator", 147 PromptingDangerProcessor(anno_type=AnnoType.DANGER, model=default_llm), 148 unit_type=UIMASystemScene, 149 added_mixins=DangerMixin, 150 added_additional_features=["danger_score", "danger_reason"], 151 requires_api_key=OPENAI_API_KEY, requires_paid_api_requests=True 152 ) 153PromptingFearAnnotator = PipelineStep("PromptingFearAnnotator", 154 PromptingDangerProcessor(anno_type=AnnoType.FEAR, model=default_llm), 155 unit_type=UIMASystemScene, 156 added_mixins=DangerMixin, 157 added_additional_features=["fear_score", "fear_reason"], 158 requires_api_key=OPENAI_API_KEY, requires_paid_api_requests=True 159 )
!!! abstract "Usage Documentation" Models
A base class for creating Pydantic models.
Attributes:
__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The `origin` and `args` items map to the [`__origin__`][genericalias.__origin__]
and [`__args__`][genericalias.__args__] attributes of [generic aliases][types-genericalias],
and the `parameter` item maps to the `__parameter__` attribute of generic classes.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.
22class FineGrained(Response): 23 answer: list[Literal[ 24 "DangerousSituationDuel", "DangerousSituationAbduction", "DangerousSituationNatural", "DangerousSituationSupernatural", "DangerousSituationAmbush", "DangerousSituationHitchcock", "DangerousSituationOther", "NoDanger"]]
!!! abstract "Usage Documentation" Models
A base class for creating Pydantic models.
Attributes:
__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The `origin` and `args` items map to the [`__origin__`][genericalias.__origin__]
and [`__args__`][genericalias.__args__] attributes of [generic aliases][types-genericalias],
and the `parameter` item maps to the `__parameter__` attribute of generic classes.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.
27class FineGrainedReason(Response): 28 answer: list[Literal[ 29 "DangerousSituationDuel", "DangerousSituationAbduction", "DangerousSituationNatural", "DangerousSituationSupernatural", "DangerousSituationAmbush", "DangerousSituationHitchcock", "DangerousSituationOther", "NoDanger"]] 30 reason: str
!!! abstract "Usage Documentation" Models
A base class for creating Pydantic models.
Attributes:
__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The `origin` and `args` items map to the [`__origin__`][genericalias.__origin__]
and [`__args__`][genericalias.__args__] attributes of [generic aliases][types-genericalias],
and the `parameter` item maps to the `__parameter__` attribute of generic classes.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.
!!! abstract "Usage Documentation" Models
A base class for creating Pydantic models.
Attributes:
__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The `origin` and `args` items map to the [`__origin__`][genericalias.__origin__]
and [`__args__`][genericalias.__args__] attributes of [generic aliases][types-genericalias],
and the `parameter` item maps to the `__parameter__` attribute of generic classes.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.
!!! abstract "Usage Documentation" Models
A base class for creating Pydantic models.
Attributes:
__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: A dictionary containing metadata about generic Pydantic models.
The `origin` and `args` items map to the [`__origin__`][genericalias.__origin__]
and [`__args__`][genericalias.__args__] attributes of [generic aliases][types-genericalias],
and the `parameter` item maps to the `__parameter__` attribute of generic classes.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.
Built-in mutable sequence.
If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.
57class FineGrainedLLM(LLM): 58 def __init__(self, model: LLMArchitecture, system_prompt: str, post_prompt: str = None, cache_maxsize: int = 0, 59 seed=NOT_GIVEN, max_tries=3, reason: bool = False): 60 if model.openai: 61 super().__init__(model, system_prompt, post_prompt, cache_maxsize, seed=seed, 62 output_format=FineGrainedReason if reason else FineGrained) 63 else: 64 if not post_prompt or "yes" not in post_prompt: 65 logger.warning("Post prompt does not contain 'yes'. Adding instruction.") 66 post_prompt = post_prompt + ("Answer with the classes that are present.") \ 67 if not reason else "Answer with a list of the classes that are present followed by a reason" 68 super(FineGrainedLLM, self).__init__(model, system_prompt, 69 post_prompt, 70 cache_maxsize, 71 seed=seed, json=False) 72 self.max_tries = max_tries 73 self.reason = reason 74 75 def __call__(self, prompt) -> (bool, str): 76 if self.model.openai: 77 response = super().__call__(prompt) 78 return response 79 tries = 0 80 while tries < self.max_tries: 81 response = super().__call__(prompt) 82 response_lower = response.lower().strip("*").strip() 83 84 label_list = get_labels(response_lower) 85 if self.reason: 86 return FineGrainedReason(answer=label_list, reason=response) 87 else: 88 return FineGrained(answer=label_list) 89 tries += 1 90 raise ValueError(f"Could not get a yes/no answer after {self.max_tries} tries.")
58 def __init__(self, model: LLMArchitecture, system_prompt: str, post_prompt: str = None, cache_maxsize: int = 0, 59 seed=NOT_GIVEN, max_tries=3, reason: bool = False): 60 if model.openai: 61 super().__init__(model, system_prompt, post_prompt, cache_maxsize, seed=seed, 62 output_format=FineGrainedReason if reason else FineGrained) 63 else: 64 if not post_prompt or "yes" not in post_prompt: 65 logger.warning("Post prompt does not contain 'yes'. Adding instruction.") 66 post_prompt = post_prompt + ("Answer with the classes that are present.") \ 67 if not reason else "Answer with a list of the classes that are present followed by a reason" 68 super(FineGrainedLLM, self).__init__(model, system_prompt, 69 post_prompt, 70 cache_maxsize, 71 seed=seed, json=False) 72 self.max_tries = max_tries 73 self.reason = reason
93def annotate_doc(doc: UIMADocument, annotation_type: str, 94 annotation_unit: Type[UIMASpan] = UIMASystemScene, 95 model: LLMArchitecture = default_llm) -> DangerDocument: 96 if annotation_type == "Fear": 97 types = ("FearDescription",) 98 system_prompt = """The task is to detect whether there is fear in the current situation in the given text 99 unit. A dangerous situation is not enough. The situation must describe concrete fear.""" 100 101 elif annotation_type == "Danger": 102 types = ("DangerousSituation",) 103 system_prompt = """The task is to detect whether there is a concrete danger at the current time and location 104 in the given text unit. A suggestion, threat or potentially dangerous situation is not enough. Fear of a 105 situation is not enough. The situation must be dangerous and concrete.""" 106 else: 107 raise ValueError(f"Unknown annotation type: {annotation_type}") 108 109 llm = YesNoLLM(model, system_prompt, 110 post_prompt="Is there a dangerous situation as described above in this text?", seed=42, 111 reason=True, max_tries=2, 112 retry_prompt_add=" Note that the text may be empty. In this case, answer 'No, because the text is empty.' Do not include anything else in your response.") 113 114 doc: DangerDocument = DangerDocument.from_other_doc(doc) 115 116 units = doc._get_annos_of_type(annotation_unit) 117 118 for unit in units: 119 response = llm(unit.text) 120 121 has_dangerous_situation = response.answer if annotation_type == 'Danger' else None 122 has_fear_description = response.answer if annotation_type == 'Fear' else None 123 has_other_suspense = None 124 125 unit.additional_features[ 126 f"{annotation_type.lower()}_score"] = response.answer 127 unit.additional_features[f"{annotation_type.lower()}_reason"] = response.reason 128 129 doc.create_system_danger_paragraph(begin=unit.begin, end=unit.end, 130 has_dangerous_situation=has_dangerous_situation, 131 has_fear_description=has_fear_description, 132 has_other_suspense=has_other_suspense, add_to_document=True) 133 134 return doc
137class PromptingDangerProcessor(PipelineProcessor): 138 def __init__(self, anno_type: AnnoType, model: LLMArchitecture = default_llm): 139 self.anno_type = anno_type 140 self.model = model 141 142 def __call__(self, doc: UIMADocument, unit_type: Type[UIMASpan] = UIMASystemScene, 143 overwrite: bool = False, **kwargs): 144 return annotate_doc(doc, annotation_type=self.anno_type, annotation_unit=unit_type, model=self.model)
Base class for protocol classes.
Protocol classes are defined as::
class Proto(Protocol):
def meth(self) -> int:
...
Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).
For example::
class C:
def meth(self) -> int:
return 0
def func(x: Proto) -> int:
return x.meth()
func(C()) # Passes static type check
See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::
class GenProto[T](Protocol):
def meth(self) -> T:
...
Pipeline step PromptingDangerAnnotator (PromptingDangerProcessor).
unit type UIMASystemScene; additional features danger_score, danger_reason.
Pipeline step PromptingFearAnnotator (PromptingDangerProcessor).
unit type UIMASystemScene; additional features fear_score, fear_reason.