{
"tool_response": {
"filePath": "\/var\/www\/scripts\/pipeline\/analyzers\/entity_extractor.py",
"oldString": "def _build_prompt_from_yaml(yaml_content: str, text: str) -> str:\n \"\"\"Build prompt string from YAML-formatted prompt content.\n\n Parses YAML structure and builds a coherent prompt for the LLM.\n \"\"\"\n try:\n config = yaml.safe_load(yaml_content)\n except yaml.YAMLError:\n # Fallback: treat as plain text\n return yaml_content.replace(\"{{TEXT}}\", text[:3000])\n\n if not isinstance(config, dict):\n return yaml_content.replace(\"{{TEXT}}\", text[:3000])\n\n parts = []\n\n # System message\n if \"system\" in config:\n parts.append(config[\"system\"].strip())\n\n # Categories\n if \"categories\" in config:\n parts.append(\"\\nExtrahiere diese Kategorien:\")\n for cat_type, cat_desc in config[\"categories\"].items():\n parts.append(f\"- {cat_type}: {cat_desc}\")\n\n # Instructions\n if \"instructions\" in config:\n parts.append(\"\")\n for instruction in config[\"instructions\"]:\n parts.append(instruction)\n\n # Output format\n if \"output\" in config:\n output_cfg = config[\"output\"]\n if \"schema\" in output_cfg:\n parts.append(f\"\\nJSON-Format:\\n{output_cfg['schema'].strip()}\")\n\n # Text placeholder\n parts.append(\"\\nText:\")\n parts.append(text[:3000])\n\n return \"\\n\".join(parts)",
"newString": "def _build_prompt_from_yaml(yaml_content: str, text: str) -> str:\n \"\"\"Pass YAML prompt directly to LLM with text placeholder replaced.\"\"\"\n return yaml_content.replace(\"{{TEXT}}\", text[:3000])",
"originalFile": "\"\"\"\nEntity Extraction - Extract and store entities from text.\n\"\"\"\n\nimport json\nimport re\nimport sys\nimport time\nimport unicodedata\n\nimport requests\nimport yaml\n\nsys.path.insert(0, \"\/var\/www\/scripts\/pipeline\")\n\nfrom config import ANTHROPIC_MODEL, OLLAMA_HOST\nfrom db import db\nfrom protokoll import protokoll\n\n# Import normalizer for deterministic type correction\nfrom analyzers.entity_normalizer import normalize_entities as normalize_entity_types\n\n\ndef _build_prompt_from_yaml(yaml_content: str, text: str) -> str:\n \"\"\"Build prompt string from YAML-formatted prompt content.\n\n Parses YAML structure and builds a coherent prompt for the LLM.\n \"\"\"\n try:\n config = yaml.safe_load(yaml_content)\n except yaml.YAMLError:\n # Fallback: treat as plain text\n return yaml_content.replace(\"{{TEXT}}\", text[:3000])\n\n if not isinstance(config, dict):\n return yaml_content.replace(\"{{TEXT}}\", text[:3000])\n\n parts = []\n\n # System message\n if \"system\" in config:\n parts.append(config[\"system\"].strip())\n\n # Categories\n if \"categories\" in config:\n parts.append(\"\\nExtrahiere diese Kategorien:\")\n for cat_type, cat_desc in config[\"categories\"].items():\n parts.append(f\"- {cat_type}: {cat_desc}\")\n\n # Instructions\n if \"instructions\" in config:\n parts.append(\"\")\n for instruction in config[\"instructions\"]:\n parts.append(instruction)\n\n # Output format\n if \"output\" in config:\n output_cfg = config[\"output\"]\n if \"schema\" in output_cfg:\n parts.append(f\"\\nJSON-Format:\\n{output_cfg['schema'].strip()}\")\n\n # Text placeholder\n parts.append(\"\\nText:\")\n parts.append(text[:3000])\n\n return \"\\n\".join(parts)\n\n\ndef normalize_name(name: str) -> str:\n \"\"\"Generate canonical_name from entity name.\n\n Rules:\n - Lowercase\n - German umlauts: ä→ae, ö→oe, ü→ue, ß→ss\n - Replace spaces with underscores\n - Remove special characters except underscores\n - Collapse multiple underscores\n \"\"\"\n if not name:\n return \"\"\n\n result = name.lower()\n\n replacements = {\n \"ä\": \"ae\", \"ö\": \"oe\", \"ü\": \"ue\", \"ß\": \"ss\",\n \"Ä\": \"ae\", \"Ö\": \"oe\", \"Ü\": \"ue\",\n }\n for old, new in replacements.items():\n result = result.replace(old, new)\n\n result = unicodedata.normalize(\"NFKD\", result)\n result = result.encode(\"ascii\", \"ignore\").decode(\"ascii\")\n result = re.sub(r\"[\\s\\-]+\", \"_\", result)\n result = re.sub(r\"[^a-z0-9_]\", \"\", result)\n result = re.sub(r\"_+\", \"_\", result)\n result = result.strip(\"_\")\n\n return result\n\n\n# Category to type mapping for new format\nCATEGORY_TYPE_MAP = {\n \"persons\": \"PERSON\",\n \"roles\": \"ROLE\",\n \"organizations\": \"ORGANIZATION\",\n \"theories\": \"THEORY\",\n \"models\": \"MODEL\",\n \"concepts\": \"CONCEPT\",\n \"artifacts\": \"ARTIFACT\",\n \"metaphors\": \"METAPHOR\",\n \"locations\": \"LOCATION\",\n}\n\n\ndef _validate_entity_in_text(entity_name: str, source_text: str) -> bool:\n \"\"\"Strictly validate that entity appears EXACTLY in source text.\"\"\"\n if not entity_name or len(entity_name) < 3:\n return False\n # Exact match required\n return entity_name in source_text\n\n\ndef _normalize_entity_response(result: dict, source_text: str) -> list[dict]:\n \"\"\"Normalize entity response to standard format with validation.\n\n Handles two formats:\n 1. New: {\"persons\":[], \"roles\":[], ...}\n 2. Legacy: {\"entities\": [...]}\n\n Also validates entities against source text to filter hallucinations.\n \"\"\"\n entities = []\n\n # Check for legacy format\n if \"entities\" in result:\n legacy_entities = result.get(\"entities\", [])\n # Validate legacy entities too\n for e in legacy_entities:\n if isinstance(e, dict) and \"name\" in e:\n if _validate_entity_in_text(e[\"name\"], source_text):\n entities.append(e)\n return entities\n\n # New categorized format\n for category, items in result.items():\n if not isinstance(items, list):\n continue\n\n entity_type = CATEGORY_TYPE_MAP.get(category, category.upper())\n\n for item in items:\n if not item or not isinstance(item, str):\n continue\n\n # Strict validation: entity must appear EXACTLY in source text\n if not _validate_entity_in_text(item, source_text):\n continue # Skip hallucinations\n\n entities.append({\n \"name\": item,\n \"type\": entity_type,\n \"description\": None,\n })\n\n return entities\n\n\ndef _call_ollama(prompt: str, model: str, timeout: int = 120) -> tuple[str, int, int, int]:\n \"\"\"Call Ollama API and return (response_text, tokens_in, tokens_out, duration_ms).\"\"\"\n start_time = time.time()\n response = requests.post(\n f\"{OLLAMA_HOST}\/api\/generate\",\n json={\"model\": model, \"prompt\": prompt, \"stream\": False, \"format\": \"json\"},\n timeout=timeout,\n )\n response.raise_for_status()\n data = response.json()\n duration_ms = int((time.time() - start_time) * 1000)\n return (\n data.get(\"response\", \"{}\"),\n data.get(\"prompt_eval_count\", 0),\n data.get(\"eval_count\", 0),\n duration_ms,\n )\n\n\ndef extract_entities_ollama(text: str, model: str = \"gemma3:27b-it-qat\") -> list[dict]:\n \"\"\"Extract entities using 2-pass approach for better categorization.\n\n Pass 1: Extract entity names from text\n Pass 2: Categorize extracted entities\n Post: Normalize types using deterministic rules\n\n Falls back to single-pass if 2-pass prompts not available.\n \"\"\"\n # Try 2-pass approach first\n pass1_template = db.get_prompt(\"entity_extraction_pass1\")\n pass2_template = db.get_prompt(\"entity_extraction_pass2\")\n\n if pass1_template and pass2_template:\n entities = _extract_entities_2pass(text, pass1_template, pass2_template, model)\n else:\n # Fallback to single-pass\n entities = _extract_entities_single_pass(text, model)\n\n # Apply deterministic type normalization (fixes LLM categorization errors)\n return normalize_entity_types(entities)\n\n\ndef _extract_entities_2pass(\n text: str, pass1_template: str, pass2_template: str, model: str\n) -> list[dict]:\n \"\"\"2-pass entity extraction: extract then categorize.\"\"\"\n try:\n # PASS 1: Extract entity names\n prompt1 = pass1_template.replace(\"{text}\", text[:3000])\n resp1, tok_in1, tok_out1, dur1 = _call_ollama(prompt1, model)\n\n try:\n result1 = json.loads(resp1)\n raw_entities = result1.get(\"entities\", [])\n except json.JSONDecodeError:\n db.log(\"WARNING\", \"Failed to parse Pass 1 JSON\")\n return []\n\n # Validate: only keep entities that appear in text\n valid_entities = [e for e in raw_entities if _validate_entity_in_text(e, text)]\n\n if not valid_entities:\n return []\n\n protokoll.log_llm_call(\n request=f\"[entity_extraction_pass1] {len(valid_entities)} entities\",\n response=json.dumps(valid_entities[:10], ensure_ascii=False),\n model_name=f\"ollama:{model}\",\n tokens_input=tok_in1,\n tokens_output=tok_out1,\n duration_ms=dur1,\n status=\"completed\",\n )\n\n # PASS 2: Categorize entities\n entities_json = json.dumps(valid_entities, ensure_ascii=False)\n prompt2 = pass2_template.replace(\"{entities}\", entities_json)\n resp2, tok_in2, tok_out2, dur2 = _call_ollama(prompt2, model)\n\n try:\n result2 = json.loads(resp2)\n categorized = result2.get(\"kategorisiert\", [])\n except json.JSONDecodeError:\n db.log(\"WARNING\", \"Failed to parse Pass 2 JSON\")\n # Fallback: return uncategorized entities\n return [{\"name\": e, \"type\": \"CONCEPT\", \"description\": None} for e in valid_entities]\n\n protokoll.log_llm_call(\n request=f\"[entity_extraction_pass2] categorize {len(valid_entities)} entities\",\n response=resp2[:1000],\n model_name=f\"ollama:{model}\",\n tokens_input=tok_in2,\n tokens_output=tok_out2,\n duration_ms=dur2,\n status=\"completed\",\n )\n\n # Normalize output\n entities = []\n for e in categorized:\n if isinstance(e, dict) and \"name\" in e and \"type\" in e:\n # Final validation\n if _validate_entity_in_text(e[\"name\"], text):\n entities.append({\n \"name\": e[\"name\"],\n \"type\": e[\"type\"].upper(),\n \"description\": e.get(\"description\"),\n })\n\n return entities\n\n except Exception as e:\n db.log(\"ERROR\", f\"2-pass entity extraction failed: {e}\")\n return []\n\n\ndef _extract_entities_single_pass(text: str, model: str) -> list[dict]:\n \"\"\"Single-pass entity extraction (legacy fallback).\"\"\"\n prompt_data = db.get_prompt_by_use_case(\"entity_extraction\")\n prompt_content = prompt_data[\"content\"] if prompt_data else None\n\n if not prompt_content:\n db.log(\"WARNING\", \"entity_extraction prompt not found in DB, using fallback\")\n prompt_content = \"\"\"Analysiere den Text und extrahiere wichtige Entitäten.\nKategorisiere als: PERSON, ORGANIZATION, CONCEPT, LOCATION\nAntworte NUR im JSON-Format:\n{\"entities\": [{\"name\": \"...\", \"type\": \"...\", \"description\": \"...\"}]}\n\nText:\n{{TEXT}}\"\"\"\n\n # Build prompt from YAML or plain text\n prompt = _build_prompt_from_yaml(prompt_content, text)\n\n try:\n resp, tok_in, tok_out, dur = _call_ollama(prompt, model)\n\n protokoll.log_llm_call(\n request=f\"[entity_extraction] {prompt[:500]}...\",\n response=resp[:2000],\n model_name=f\"ollama:{model}\",\n tokens_input=tok_in,\n tokens_output=tok_out,\n duration_ms=dur,\n status=\"completed\",\n )\n\n ... [TRUNCATED-405e211c80849fc3]",
"structuredPatch": [
{
"oldStart": 22,
"oldLines": 50,
"newStart": 22,
"newLines": 10,
"lines": [
" ",
" ",
" def _build_prompt_from_yaml(yaml_content: str, text: str) -> str:",
"- \"\"\"Build prompt string from YAML-formatted prompt content.",
"+ \"\"\"Pass YAML prompt directly to LLM with text placeholder replaced.\"\"\"",
"+ return yaml_content.replace(\"{{TEXT}}\", text[:3000])",
" ",
"- Parses YAML structure and builds a coherent prompt for the LLM.",
"- \"\"\"",
"- try:",
"- config = yaml.safe_load(yaml_content)",
"- except yaml.YAMLError:",
"- # Fallback: treat as plain text",
"- return yaml_content.replace(\"{{TEXT}}\", text[:3000])",
" ",
"- if not isinstance(config, dict):",
"- return yaml_content.replace(\"{{TEXT}}\", text[:3000])",
"-",
"- parts = []",
"-",
"- # System message",
"- if \"system\" in config:",
"- parts.append(config[\"system\"].strip())",
"-",
"- # Categories",
"- if \"categories\" in config:",
"- parts.append(\"\\nExtrahiere diese Kategorien:\")",
"- for cat_type, cat_desc in config[\"categories\"].items():",
"- parts.append(f\"- {cat_type}: {cat_desc}\")",
"-",
"- # Instructions",
"- if \"instructions\" in config:",
"- parts.append(\"\")",
"- for instruction in config[\"instructions\"]:",
"- parts.append(instruction)",
"-",
"- # Output format",
"- if \"output\" in config:",
"- output_cfg = config[\"output\"]",
"- if \"schema\" in output_cfg:",
"- parts.append(f\"\\nJSON-Format:\\n{output_cfg['schema'].strip()}\")",
"-",
"- # Text placeholder",
"- parts.append(\"\\nText:\")",
"- parts.append(text[:3000])",
"-",
"- return \"\\n\".join(parts)",
"-",
"-",
" def normalize_name(name: str) -> str:",
" \"\"\"Generate canonical_name from entity name.",
" "
]
}
],
"userModified": false,
"replaceAll": false
}
}