113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
import httpx
|
|
|
|
IDENTIFY_SPEAKERS_PROMPT = """Du bekommst den Anfang eines Gesprächstranskripts mit Sprecher-Labels (SPEAKER_00, SPEAKER_01, ...).
|
|
Ermittle, welche echten Namen den Sprechern zugeordnet werden können — z.B. durch direkte Anrede ("Herr Möller", "Frank").
|
|
Antworte NUR mit einem JSON-Objekt: {"SPEAKER_00": "Name oder null", "SPEAKER_01": "Name oder null"}
|
|
Kein weiterer Text, keine Erklärung."""
|
|
|
|
SUMMARIZE_PROMPT = """Du bist ein präziser Assistent für Business-Kommunikation.
|
|
Du bekommst ein Gesprächstranskript mit Sprecher-Labels.
|
|
Erstelle eine strukturierte Zusammenfassung auf Deutsch mit:
|
|
1. Einem passenden H1-Titel
|
|
2. ## Wichtigste Punkte (Aufzählung)
|
|
3. ## Offene Fragen (Aufzählung, falls vorhanden)
|
|
4. ## Nächste Schritte / Ideen (Aufzählung, falls vorhanden)
|
|
Antworte NUR mit dem fertigen Markdown."""
|
|
|
|
SYSTEM_PROMPT = """Du bist ein präziser Schreibassistent.
|
|
Du bekommst einen rohen Sprachtranskript und optionale Instruktionen des Nutzers.
|
|
Deine Aufgabe:
|
|
1. Bereinige den Text (Füllwörter, Wiederholungen, Tippfehler)
|
|
2. Gliedere den Text in sinnvolle Absätze — trenne Gedanken durch Leerzeilen
|
|
3. Verwende Markdown-Überschriften (##) wenn der Text mehrere Themen hat
|
|
4. Verwende Aufzählungslisten (- ) für Aufzählungen oder Handlungsschritte
|
|
5. Erzeuge einen passenden deutschen Titel als H1
|
|
6. Beachte Instruktionen des Nutzers wenn vorhanden
|
|
7. Antworte NUR mit dem fertigen Markdown — kein Kommentar, keine Erklärung
|
|
|
|
Format:
|
|
# Titel
|
|
|
|
Erster Absatz...
|
|
|
|
Zweiter Absatz...
|
|
|
|
## Abschnitt (nur wenn sinnvoll)
|
|
|
|
- Punkt 1
|
|
- Punkt 2
|
|
"""
|
|
|
|
|
|
class OllamaClient:
|
|
def __init__(self, base_url: str = "http://localhost:11434"):
|
|
self.base_url = base_url
|
|
|
|
async def list_models(self) -> list[str]:
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(f"{self.base_url}/api/tags")
|
|
r.raise_for_status()
|
|
return [m["name"] for m in r.json().get("models", [])]
|
|
|
|
async def refine(
|
|
self,
|
|
raw_text: str,
|
|
instructions: str = "",
|
|
model: str = "gemma3:12b",
|
|
) -> str:
|
|
prompt = f"Transkript:\n{raw_text}"
|
|
if instructions.strip():
|
|
prompt += f"\n\nInstruktionen:\n{instructions.strip()}"
|
|
async with httpx.AsyncClient(timeout=120) as client:
|
|
r = await client.post(
|
|
f"{self.base_url}/api/generate",
|
|
json={"model": model, "prompt": prompt, "system": SYSTEM_PROMPT, "stream": False},
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()["response"]
|
|
|
|
async def identify_speakers(
|
|
self,
|
|
transcript_excerpt: str,
|
|
model: str = "gemma3:12b",
|
|
) -> dict[str, str]:
|
|
"""Try to map SPEAKER_XX labels to real names. Returns {} on failure."""
|
|
import json
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
r = await client.post(
|
|
f"{self.base_url}/api/generate",
|
|
json={
|
|
"model": model,
|
|
"prompt": f"Transkript-Anfang:\n{transcript_excerpt[:2000]}",
|
|
"system": IDENTIFY_SPEAKERS_PROMPT,
|
|
"stream": False,
|
|
},
|
|
)
|
|
r.raise_for_status()
|
|
raw = r.json()["response"].strip()
|
|
try:
|
|
data = json.loads(raw)
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
return {k: v for k, v in data.items() if v}
|
|
except Exception:
|
|
return {}
|
|
|
|
async def summarize(
|
|
self,
|
|
annotated_transcript: str,
|
|
model: str = "gemma3:12b",
|
|
) -> str:
|
|
async with httpx.AsyncClient(timeout=180) as client:
|
|
r = await client.post(
|
|
f"{self.base_url}/api/generate",
|
|
json={
|
|
"model": model,
|
|
"prompt": f"Transkript:\n{annotated_transcript}",
|
|
"system": SUMMARIZE_PROMPT,
|
|
"stream": False,
|
|
},
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()["response"].strip()
|