Coverage for core/langgraph/_llm.py: 60%
354 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-08 05:24 +0000
1"""
2_llm.py — OpenAI-compatible LLM invoker + Langfuse tracing + skill injection.
4OpenAIInvoker on callable, joka StepRunner kutsuu agent-stepille
5(`sequential`, `parallel`, `foreach`-body sequential/parallel). Se:
7- yhdistää AgentInstance-spec:n tiedoista täydellisen system_promptin
8 (skill-tiedostot luettu + `{{ skills }}` korvattu + params-substituutio)
9- serialisoi `step.input`-arvon user-messageksi
10- ajaa OpenAI-streamin DeepSeek-yhteensopivalla endpointilla
11- raportoi Langfuse-generationina pipeline-trace alle
12- parsii vastauksen JSON-objektiksi jos mahdollista (audit/triage/review),
13 muuten palauttaa raw-markdownin (ticket writer/refiner)
14- valvoo RunBudgetia (max LLM calls / wall-clock timeout)
16API:
17 invoker = OpenAIInvoker(config=..., budget=..., project_slug=..., pipeline_name=...)
18 new_state = invoker(resolved_spec, resolved_step, state)
19"""
20from __future__ import annotations
22import json
23import logging
24import os
25import re
26import threading
27import time
28from dataclasses import dataclass, field
29from pathlib import Path
30from typing import Any
32log = logging.getLogger(__name__)
34SKILL_DIR = Path("/skills")
37class SkillLoadError(ValueError):
38 """Skill-latauksen kova virhe (ADR-0012) — ei hiljaista fallbackia.
40 Erottaa skill-resoluution virheet muista ValueErrorista: runnerin
41 parallel-käsittelijät re-raisaavat tämän (kova ajo-failure kaikilla
42 step-polkuilla), kun muut step-virheet pysyvät state-virheinä.
43 """
45_LANGFUSE_INPUT_MAX = 6000
46_LANGFUSE_FLUSH_EVERY = 8
49# ---------------------------------------------------------------------------
50# RunBudget — sama semantiikka kuin doc-quality-pipeline.py:ssa
51# ---------------------------------------------------------------------------
54@dataclass
55class RunBudget:
56 max_llm_calls: int = 48
57 run_timeout_seconds: int = 240
58 start: float = field(default_factory=time.monotonic)
59 llm_calls: int = 0
60 stopped: bool = False
61 stop_reason: str = ""
62 _lock: threading.Lock = field(default_factory=threading.Lock)
64 def try_consume(self) -> bool:
65 with self._lock:
66 if self.stopped:
67 return False
68 if time.monotonic() - self.start > self.run_timeout_seconds:
69 self.stopped = True
70 self.stop_reason = "run_timeout"
71 return False
72 if self.llm_calls >= self.max_llm_calls:
73 self.stopped = True
74 self.stop_reason = "max_llm_calls"
75 return False
76 self.llm_calls += 1
77 return True
79 def as_stats(self) -> dict[str, Any]:
80 return {
81 "llm_calls": self.llm_calls,
82 "budget_stopped": self.stopped,
83 "budget_stop_reason": self.stop_reason or None,
84 }
87# ---------------------------------------------------------------------------
88# Skill loading
89# ---------------------------------------------------------------------------
92def _skill_repos_from_env() -> list[dict]:
93 """Lue SKILL_REPOS-env (JSON [{"name": ..., "url": ...}]) → lista entryistä.
95 Puuttuva env → [] (sopimus: SKILL_REPOS on valinnainen). Mikä tahansa
96 epäkelpo arvo on kova virhe (ADR-0012): malformi JSON, ei-lista, tai
97 entry ilman ei-tyhjää name- ja url-merkkijonoa → SkillLoadError.
98 """
99 raw = os.getenv("SKILL_REPOS")
100 if not raw:
101 return []
102 try:
103 repos = json.loads(raw)
104 except json.JSONDecodeError as exc:
105 raise SkillLoadError(f"SKILL_REPOS is not valid JSON: {exc}") from exc
106 if not isinstance(repos, list):
107 raise SkillLoadError(f"SKILL_REPOS must be a JSON list, got {type(repos).__name__}")
108 for idx, repo in enumerate(repos):
109 if not isinstance(repo, dict):
110 raise SkillLoadError(
111 f"SKILL_REPOS entry {idx} must be an object, got {type(repo).__name__}"
112 )
113 name = repo.get("name")
114 url = repo.get("url")
115 if not isinstance(name, str) or not name.strip():
116 raise SkillLoadError(f"SKILL_REPOS entry {idx} is missing a non-empty string 'name'")
117 if not isinstance(url, str) or not url.strip():
118 raise SkillLoadError(
119 f"SKILL_REPOS entry {idx} ('{name}') is missing a non-empty string 'url'"
120 )
121 return repos
124def _normalize_skill_libs() -> list[Path]:
125 """Skill-lähdelista: koodirepon skills/-kirjasto ensin, sitten skill-repo-mountit.
127 Koodirepon implisiittinen skills/-kirjasto (`PROJECT_ROOT/skills/`,
128 oletus `/project/skills`) on aina korkein prioriteetti skill-lähteenä
129 (koodirepon ylivalta, ADR-0018 D3), ja skill-repot tulevat SKILL_REPOS-
130 järjestyksessä (`/skills/<name>` -mountit, ADR-0018 D4). Puuttuvat
131 kansiot ohitetaan hiljaa resoluutiossa.
132 """
133 project_root = Path(os.getenv("PROJECT_ROOT", "/project"))
134 return [project_root / "skills"] + [
135 SKILL_DIR / repo["name"] for repo in _skill_repos_from_env()
136 ]
139def load_skills(spec: dict, *, max_bytes: int = 8000) -> str:
140 """
141 Lukee `spec.skills`-flat-listan (skill-nimiä) ja palauttaa yhdistetyn
142 tekstin (truncated per file).
144 Jokainen nimi on pelkkä skill-nimi: `lib / <nimi>/`-kansio.
145 Ensimmäinen skill-lähde, jonka kansio löytyy, voittaa (first-match):
146 koodirepon skills/-kirjasto (`PROJECT_ROOT/skills/`, koodirepon
147 ylivalta, ADR-0018 D3) ensin, sitten skill-repot SKILL_REPOS-
148 järjestyksessä (prioriteetti, ADR-0018 D4); puuttuvat lähteet
149 ohitetaan hiljaa. Kansion kaikki tiedostot injektoidaan erikseen
150 otsakkeella (SKILL.md ensin, sitten loput aakkosjärjestyksessä
151 suhteellisen polun mukaan).
153 Resolvoitumaton nimi → SkillLoadError (ADR-0012): viesti listaa etsityt
154 kirjastopolut ja saatavilla olevat skill-nimet. Ei hiljaista ohitusta.
155 """
156 skills = (spec.get("spec") or {}).get("skills") or []
157 if not skills:
158 return ""
159 libs = _normalize_skill_libs()
160 parts: list[str] = []
161 for entry in skills:
162 parts.extend(_load_bare_skill(entry, libs, max_bytes=max_bytes))
163 return "\n\n---\n\n".join(parts)
166def _load_bare_skill(entry: str, libs: list[Path], *, max_bytes: int) -> list[str]:
167 """Skill-nimi: kansion `lib / <nimi>/` kaikki tiedostot otsakkeilla.
169 Ensimmäinen skill-repo, jonka kansio löytyy, voittaa. Jos vastaava
170 `lib / entry` on tiedosto (ei kansio) tai puuttuu kaikista
171 skill-repoista → SkillLoadError.
172 """
173 for lib in libs:
174 folder = lib / entry
175 if folder.is_dir():
176 parts: list[str] = []
177 for path in _skill_folder_files(folder):
178 relpath = path.relative_to(folder)
179 parts.append(
180 f"## Skill: {entry} — {relpath}\n\n"
181 f"{_read_skill_file(entry, path, max_bytes=max_bytes)}"
182 )
183 return parts
184 if folder.is_file():
185 raise SkillLoadError(
186 f"skill '{entry}' resolves to a file, expected a skill folder "
187 f"(SKILL.md etc.): {folder}"
188 )
189 raise SkillLoadError(_unresolved_skill_error(entry, libs))
192def _skill_folder_files(folder: Path) -> list[Path]:
193 """Kansion kaikki tiedostot rekursiivisesti, deterministinen järjestys:
194 juuren SKILL.md ensin, sitten loput suhteellisen polun mukaan
195 aakkosjärjestyksessä."""
196 files = [p for p in folder.rglob("*") if p.is_file()]
197 files.sort(
198 key=lambda p: (
199 p.relative_to(folder) != Path("SKILL.md"),
200 str(p.relative_to(folder)),
201 )
202 )
203 return files
206def _read_skill_file(entry: str, path: Path, *, max_bytes: int) -> str:
207 try:
208 text = path.read_text(encoding="utf-8", errors="replace")
209 except OSError as exc:
210 raise SkillLoadError(
211 f"skill entry '{entry}' file unreadable: {path} ({exc})"
212 ) from exc
213 return text[:max_bytes]
216def _available_skill_names(libs: list[Path]) -> list[str]:
217 """Kirjastojen tarjonta: skill-kansiot (sisältävät SKILL.md) + top-tason
218 *.md-tiedostot. Sorted + deduped."""
219 names: set[str] = set()
220 for lib in libs:
221 if not lib.is_dir():
222 continue
223 for child in lib.iterdir():
224 if child.is_dir():
225 if (child / "SKILL.md").is_file():
226 names.add(child.name)
227 elif child.is_file() and child.name.endswith(".md"):
228 names.add(child.name)
229 return sorted(names)
232def _unresolved_skill_error(entry: str, libs: list[Path]) -> str:
233 available = ", ".join(_available_skill_names(libs)) or "(none)"
234 searched = "\n".join(f" - {lib}" for lib in libs)
235 return (
236 f"skill '{entry}' not found in any skill repo "
237 f"(available skills: {available}). Searched skill repos:\n{searched}"
238 )
241# ---------------------------------------------------------------------------
242# Template substitution (simple {{ key }} replacement against a dict)
243# ---------------------------------------------------------------------------
245_TMPL_RE = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_.]*)\s*\}\}")
248def _render(text: str, ctx: dict[str, Any]) -> str:
249 if not isinstance(text, str) or "{{" not in text:
250 return text
252 def _lookup(path: str) -> str:
253 cur: Any = ctx
254 for part in path.split("."):
255 if isinstance(cur, dict):
256 cur = cur.get(part)
257 else:
258 return ""
259 if cur is None:
260 return ""
261 return str(cur) if not isinstance(cur, str) else cur
263 return _TMPL_RE.sub(lambda m: _lookup(m.group(1)), text)
266# ---------------------------------------------------------------------------
267# Langfuse trace helpers
268# ---------------------------------------------------------------------------
271class LangfuseTracer:
272 """
273 Per-ajo Langfuse-tracen alustus + generation-spanit per LLM-kutsu.
274 Käyttää singleton-Langfuse-clientiä env-muuttujista.
275 """
277 def __init__(self, project_slug: str, pipeline_name: str, run_id: str):
278 self.project_slug = project_slug
279 self.pipeline_name = pipeline_name
280 self.run_id = run_id or "unknown"
281 self._lf: Any = None
282 self._trace_ctx: dict[str, Any] | None = None
283 self._flush_counter = 0
284 self._init_done = False
286 def _client(self) -> Any:
287 if self._lf is not None:
288 return self._lf
289 try:
290 from langfuse import Langfuse # type: ignore
291 except ImportError:
292 log.warning("[langfuse] package not installed — tracing disabled")
293 return None
294 prev = os.environ.get("LANGFUSE_TRACING_ENABLED")
295 os.environ["LANGFUSE_TRACING_ENABLED"] = "true"
296 try:
297 self._lf = Langfuse()
298 except Exception as exc: # noqa: BLE001
299 log.warning("[langfuse] init failed: %s — tracing disabled", exc)
300 self._lf = None
301 finally:
302 if prev is None:
303 os.environ["LANGFUSE_TRACING_ENABLED"] = "false"
304 else:
305 os.environ["LANGFUSE_TRACING_ENABLED"] = prev
306 return self._lf
308 @property
309 def trace_name(self) -> str:
310 return f"{self.project_slug}:{self.pipeline_name}"
312 @property
313 def session_id(self) -> str:
314 return f"{self.project_slug}-{self.pipeline_name}"
316 def _ctx(self) -> dict[str, Any] | None:
317 lf = self._client()
318 if lf is None:
319 return None
320 if self._trace_ctx is None:
321 self._trace_ctx = {"trace_id": lf.create_trace_id(seed=self.run_id)}
322 return self._trace_ctx
324 def init_trace(self) -> None:
325 if self._init_done:
326 return
327 lf = self._client()
328 ctx = self._ctx()
329 if lf is None or ctx is None:
330 self._init_done = True
331 return
332 try:
333 obs = lf.start_observation(
334 trace_context=ctx,
335 name=self.trace_name,
336 as_type="span",
337 input={
338 "pipeline": self.pipeline_name,
339 "project": self.project_slug,
340 "run_id": self.run_id,
341 },
342 )
343 self._apply_meta(obs)
344 obs.end()
345 self.flush(force=True)
346 except Exception as exc: # noqa: BLE001
347 log.warning("[langfuse] init_trace failed: %s", exc)
348 self._init_done = True
350 def _apply_meta(self, obs: Any) -> None:
351 try:
352 obs.update_trace(
353 name=self.trace_name,
354 user_id=self.project_slug,
355 session_id=self.session_id,
356 tags=[f"pipeline:{self.pipeline_name}"],
357 )
358 except Exception: # noqa: BLE001
359 pass
361 def generation(
362 self,
363 name: str,
364 *,
365 model: str,
366 temperature: float,
367 messages: list[dict[str, str]],
368 ) -> Any:
369 lf = self._client()
370 ctx = self._ctx()
371 if lf is None or ctx is None:
372 return _NoopObservation()
373 try:
374 obs = lf.start_observation(
375 trace_context=ctx,
376 name=name,
377 as_type="generation",
378 model=model,
379 model_parameters={"temperature": temperature},
380 input=_truncate_messages(messages),
381 )
382 self._apply_meta(obs)
383 return obs
384 except Exception as exc: # noqa: BLE001
385 log.warning("[langfuse] generation failed: %s", exc)
386 return _NoopObservation()
388 def flush(self, *, force: bool = False) -> None:
389 lf = self._client()
390 if lf is None:
391 return
392 self._flush_counter += 1
393 if force or self._flush_counter % _LANGFUSE_FLUSH_EVERY == 0:
394 try:
395 lf.flush()
396 except Exception: # noqa: BLE001
397 pass
400class _NoopObservation:
401 def update(self, **_: Any) -> None: ...
402 def end(self) -> None: ...
405def _truncate_messages(messages: list[dict[str, str]]) -> list[dict[str, str]]:
406 out: list[dict[str, str]] = []
407 for m in messages:
408 content = m.get("content", "")
409 if isinstance(content, str) and len(content) > _LANGFUSE_INPUT_MAX:
410 content = (
411 content[:_LANGFUSE_INPUT_MAX]
412 + f"\n… [{len(m.get('content', ''))} chars truncated for Langfuse]"
413 )
414 out.append({"role": m.get("role", "user"), "content": content})
415 return out
418# ---------------------------------------------------------------------------
419# JSON output parsing — sama semantiikka kuin doc-quality-pipeline._parse_json_findings
420# ---------------------------------------------------------------------------
423def _try_parse_json(text: str) -> tuple[Any, str | None]:
424 """Yritä parsia JSON. Palauttaa (data, error). data=None jos epäonnistui."""
425 if not text or not text.strip():
426 return None, "empty"
427 raw = text.strip()
428 fence = re.search(r"```(?:json)?\s*([\s\S]*?)```", raw)
429 if fence:
430 raw = fence.group(1).strip()
431 else:
432 m = re.search(r"\{[\s\S]*\}|\[[\s\S]*\]", raw)
433 if m:
434 raw = m.group(0)
435 try:
436 return json.loads(raw), None
437 except json.JSONDecodeError as e:
438 # Yritys: korjaa katkennut JSON
439 repaired = raw
440 if repaired.count('"') % 2 == 1:
441 repaired = repaired.rsplit('"', 1)[0] + '"'
442 repaired = re.sub(r",\s*$", "", repaired.rstrip())
443 for suffix in ("", "}", "]}", "\"]}"):
444 try:
445 return json.loads(repaired + suffix), None
446 except json.JSONDecodeError:
447 continue
448 return None, str(e)
451# ---------------------------------------------------------------------------
452# OpenAIInvoker
453# ---------------------------------------------------------------------------
456class OpenAIInvoker:
457 """
458 Default LLM invoker for agent-platform pipelines.
460 Yksi instanssi per pipeline-ajo. Thread-safe (semafori rinnakkaisuudelle).
461 """
463 def __init__(
464 self,
465 *,
466 providers: dict[str, dict[str, Any]],
467 max_parallel: int,
468 budget: RunBudget,
469 tracer: LangfuseTracer,
470 ) -> None:
471 self._providers = providers
472 self._clients: dict[str, Any] = {}
473 self.budget = budget
474 self.tracer = tracer
475 self._semaphore = threading.Semaphore(max(1, max_parallel))
477 def _openai_client(self, base_url: str, api_key: str) -> Any:
478 key = f"{base_url}|{api_key}"
479 if key not in self._clients:
480 from openai import OpenAI # type: ignore
481 self._clients[key] = OpenAI(base_url=base_url, api_key=api_key)
482 return self._clients[key]
484 # ---- Public callable -------------------------------------------------
486 def __call__(
487 self,
488 resolved_spec: dict,
489 resolved_step: dict,
490 state: dict,
491 ) -> dict:
492 """
493 StepRunner-yhteensopiva: (spec, step, state) → uusi state.
495 `resolved_step` = step jossa params/input on JO resolvoitu
496 StepRunnerissa template-engineä käyttäen. Tämä invoker ei enää
497 evaluoi `{{ }}`-lausekkeita state/run_context-vastaan, vain
498 substituoi `step.params` ja sisäänrakennetut feature-arvot
499 agentin `system_prompt`/`user`-mallineisiin.
500 """
501 step_id = resolved_step.get("id", "?")
502 agent_spec = resolved_spec.get("spec") or {}
504 model_spec = agent_spec.get("model") or {}
505 provider_name = model_spec.get("provider")
507 if not provider_name:
508 if self._providers:
509 provider_name = next(iter(self._providers))
510 else:
511 raise ValueError("No LLM provider configured and no provider specified in agent spec")
513 provider_cfg = self._providers.get(provider_name)
514 if provider_cfg is None:
515 raise ValueError(
516 f"Unknown LLM provider '{provider_name}'. "
517 f"Available: {', '.join(sorted(self._providers.keys()))}"
518 )
520 model_override = model_spec.get("model")
521 model = model_override if model_override else provider_cfg["model"]
523 # Budget check
524 if not self.budget.try_consume():
525 return {
526 **state,
527 step_id: {"error": f"budget:{self.budget.stop_reason}"},
528 }
530 # ---- Build system + user messages --------------------------------
531 skills_text = load_skills(resolved_spec)
532 step_params = resolved_step.get("params") or {}
533 input_value = resolved_step.get("input")
535 # Auto-feature: path verification block (filled by builtin path-verifier
536 # earlier in pipeline, or empty string if not present).
537 if "path_verification_block" not in step_params:
538 step_params = {**step_params, "path_verification_block": ""}
540 # Render system_prompt template with skills + params
541 prompt_ctx: dict[str, Any] = {
542 "skills": skills_text,
543 **{k: v for k, v in step_params.items() if isinstance(k, str)},
544 }
545 system_text = _render(
546 str(agent_spec.get("system_prompt") or ""),
547 prompt_ctx,
548 ).strip()
550 # User message: serialise input + step params
551 user_text = _format_user_message(input_value, step_params)
553 # Temperature: agent spec → provider default → step params
554 temperature = float(
555 model_spec.get("temperature", provider_cfg.get("temperature", 0.0))
556 )
557 if "temperature" in step_params:
558 try:
559 temperature = float(step_params["temperature"])
560 except (TypeError, ValueError):
561 pass
563 messages = [
564 {"role": "system", "content": system_text},
565 {"role": "user", "content": user_text},
566 ]
568 # ---- LLM call with Langfuse generation ---------------------------
569 agent_name = (resolved_spec.get("metadata") or {}).get("name", "agent")
570 gen_name = f"{step_id}:{agent_name}"
572 with self._semaphore:
573 obs = self.tracer.generation(
574 gen_name,
575 model=model,
576 temperature=temperature,
577 messages=messages,
578 )
579 content = ""
580 usage: dict[str, int] | None = None
581 err: str | None = None
582 try:
583 content, usage = self._stream_completion(messages, temperature, model, provider_cfg["base_url"], provider_cfg["api_key"])
584 except Exception as exc: # noqa: BLE001
585 err = f"{type(exc).__name__}: {exc}"
586 log.warning("[invoke] %s LLM call failed: %s", step_id, err)
587 try:
588 obs.update(
589 output=content[:8000] if len(content) > 8000 else content,
590 usage_details=usage,
591 )
592 except Exception: # noqa: BLE001
593 pass
594 try:
595 obs.end()
596 except Exception: # noqa: BLE001
597 pass
598 self.tracer.flush()
600 # ---- Parse output ------------------------------------------------
601 if err:
602 return {**state, step_id: {"error": err, "raw": content}}
604 output_format = ((agent_spec.get("output") or {}).get("format") or "").lower()
605 if output_format == "markdown":
606 return {**state, step_id: {"output": content.strip()}}
608 parsed, perr = _try_parse_json(content)
609 if parsed is not None:
610 return {**state, step_id: parsed}
611 # Fallback: raw text
612 return {**state, step_id: {"output": content.strip(), "parse_error": perr}}
614 # ---- Streaming completion -------------------------------------------
616 def _stream_completion(
617 self,
618 messages: list[dict[str, str]],
619 temperature: float,
620 model: str,
621 base_url: str,
622 api_key: str,
623 ) -> tuple[str, dict[str, int] | None]:
624 client = self._openai_client(base_url, api_key)
625 resp = client.chat.completions.create(
626 model=model,
627 temperature=temperature,
628 messages=messages,
629 stream=True,
630 stream_options={"include_usage": True},
631 )
632 content = ""
633 usage: dict[str, int] | None = None
634 for chunk in resp:
635 if chunk.choices and chunk.choices[0].delta.content:
636 content += chunk.choices[0].delta.content
637 if getattr(chunk, "usage", None):
638 u = chunk.usage
639 usage = {
640 "input": u.prompt_tokens or 0,
641 "output": u.completion_tokens or 0,
642 "total": u.total_tokens or 0,
643 }
644 return content, usage
647# ---------------------------------------------------------------------------
648# Helpers
649# ---------------------------------------------------------------------------
652def _format_user_message(input_value: Any, params: dict[str, Any]) -> str:
653 """
654 Muotoile agentin user-message resolvoidusta inputista + paramseista.
655 """
656 parts: list[str] = []
658 # Inputin esitysmuoto: yritetään olla informatiivisia
659 if input_value is None:
660 pass
661 elif isinstance(input_value, str):
662 parts.append(input_value)
663 elif isinstance(input_value, dict):
664 # Doc-file form: {path, content, ...} → human-readable
665 if "path" in input_value and "content" in input_value:
666 path = input_value.get("path", "")
667 content = input_value.get("content", "")
668 parts.append(f"Path: {path}\n\n---\n{content}")
669 else:
670 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
671 elif isinstance(input_value, list):
672 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
673 else:
674 parts.append(str(input_value))
676 # Append path verification block (auto-feature) jos asetettu
677 pvb = params.get("path_verification_block") or ""
678 if pvb:
679 parts.append(str(pvb))
681 # Lisää muut paramit (focus, finding jne.) jos eivät jo ole system_promptissa
682 extra: dict[str, Any] = {}
683 skip_keys = {"path_verification_block", "skills"}
684 for k, v in params.items():
685 if k in skip_keys:
686 continue
687 # Älä toista pieniä skalaareja - ne yleensä jo system_promptissa
688 if isinstance(v, (dict, list)):
689 extra[k] = v
690 if extra:
691 parts.append("## Additional context\n\n" + json.dumps(extra, ensure_ascii=False, indent=2))
693 return "\n\n".join(p for p in parts if p)