Coverage for core/langgraph/discover.py: 0%
78 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-08 05:24 +0000
1"""
2discover.py — dynaaminen pipeline-lataaja.
4Käynnistyksessä:
5 1. Skannaa /project/.agent-platform/*-pipeline.py (legacy, backward-compat)
6 2. Skannaa /project/.agent-platform/**/*.yaml (uusi manifest-pohjainen)
7 3. Generoi /tmp/langgraph.json (graph-nimet tiedostonimistä / manifest-nimistä)
8 4. Tulostaa löydetyt pipelinet
10YAML-manifest-pipelinet:
11 - Löydetyt Pipeline-manifestit rekisteröidään nimellä metadata.name
12 - Graph-polku osoittaa runner.py:hin: /project/langgraph/runner.py:workflow_<name>
13 - runner.py:ssä on thin wrapper workflow_<name> per pipeline
14 - Dynaaminen registry: /tmp/pipeline_registry.json
16Legacy-pipelinet (*-pipeline.py):
17 - Toimivat edelleen kuten ennen (backward-compatible)
19Graphin nimi: metadata.name (YAML) tai tiedostonimi ilman "-pipeline.py".
20Tukee PostgreSQL-storea (LANGGRAPH_STORE_POSTGRES_URI env-var).
21"""
23import json
24import os
25import sys
26from pathlib import Path
28PIPELINE_DIR = Path(os.getenv("PROJECT_ROOT", "/project")) / ".agent-platform"
29PIPELINE_PATTERN = "*-pipeline.py"
30MANIFEST_PATTERN = "**/*.yaml"
31OUTPUT = Path("/tmp/langgraph.json")
32REGISTRY_OUTPUT = Path("/tmp/pipeline_registry.json")
33STORE_URI = os.getenv("LANGGRAPH_STORE_POSTGRES_URI", "")
35# Runner.py sijaitsee samassa hakemistossa kuin discover.py
36RUNNER_PATH = Path(__file__).parent / "runner.py"
37RUNNER_MODULE_PATH = "/project/langgraph/runner.py"
40def _load_yaml_pipelines(pipeline_dir: Path) -> list[dict]:
41 """
42 Skannaa *.yaml manifestit ja palauttaa löydettyjen Pipeline-manifestien
43 listan: [{name: str, source: str, budget: dict}].
44 Ei tarvita pyyaml:ia jos se ei ole asennettuna — graceful degradation.
45 """
46 try:
47 import yaml # noqa: PLC0415
48 except ImportError:
49 print("[discover] pyyaml not installed — skipping YAML manifest scan")
50 return []
52 pipelines: list[dict] = []
53 if not pipeline_dir.exists():
54 return pipelines
56 for yaml_path in sorted(pipeline_dir.glob(MANIFEST_PATTERN)):
57 try:
58 content = yaml_path.read_text(encoding="utf-8")
59 except OSError:
60 continue
61 for raw in content.split("\n---"):
62 raw_clean = "\n".join(
63 line for line in raw.splitlines()
64 if not line.strip().startswith("#")
65 )
66 try:
67 doc = yaml.safe_load(raw_clean)
68 except yaml.YAMLError:
69 continue
70 if (
71 isinstance(doc, dict)
72 and doc.get("apiVersion") == "agent-platform/v1"
73 and doc.get("kind") == "Pipeline"
74 ):
75 name = (doc.get("metadata") or {}).get("name")
76 if name:
77 budget = (doc.get("spec") or {}).get("budget") or {}
78 pipelines.append({
79 "name": name,
80 "source": str(yaml_path),
81 "budget": budget,
82 })
83 print(f"[discover] YAML Pipeline '{name}' ← {yaml_path.name}")
85 return pipelines
88def _generate_wrapper(name: str, source_yaml: str, budget: dict) -> str:
89 """
90 Tuottaa thin-wrapperin /tmp/<name>-pipeline.py.
92 Wrapper:
93 - lataa config.yaml
94 - alustaa LangfuseTracer (per ajo, uuid-seed)
95 - alustaa RunBudget pipeline-manifestin budget-asetuksista
96 - alustaa OpenAIInvoker
97 - kutsuu build_workflow(name, invoke_agent_fn=invoker, run_context={budget, config})
98 """
99 import json as _json
100 # repr() tuottaa validia Python-dict-literalia (True/False/None) toisin
101 # kuin json.dumps() joka tuottaa JSON-syntaksia (true/false/null).
102 budget_literal = repr(budget)
103 return f'''# Auto-generated by discover.py — do not edit
104"""Per-pipeline wrapper using LangGraph factory pattern.
106LangGraph calls ``make_workflow(config)`` ONCE PER RUN, ensuring fresh
107RunBudget / LangfuseTracer / OpenAIInvoker per invocation. This avoids the
108singleton-state bug where module-level instances accumulated across runs
109(exhausting the LLM-call budget after the first run).
111The factory is async because blockbuster (LangGraph local_dev) detects
112blocking I/O on the event loop. All file/network work runs in a thread.
113"""
114import asyncio
115import os
116import sys
117import uuid
119import yaml
121sys.path.insert(0, "/project/langgraph")
123from runner import build_workflow # noqa: E402
124from _llm import OpenAIInvoker, LangfuseTracer, RunBudget # noqa: E402
126# --- Static (safe at module scope: no side effects, no mutable state) -----
127_PIPELINE_NAME = {name!r}
128_BUDGET_SPEC = {budget_literal}
131def _build_workflow_sync():
132 """All blocking work (file I/O, manifest load, tracer init) in one place."""
133 # Provider registry — K8s Job path sets LLM_CONFIGS env; dev/local path
134 # passes an empty registry (no config.yaml). By ADR-0016 D6, K8s is the
135 # only path; config.yaml is obsolete.
136 import json as _json_mod
137 _providers_raw = os.getenv("LLM_CONFIGS", "[]")
138 try:
139 _PROVIDERS = _json_mod.loads(_providers_raw)
140 except _json_mod.JSONDecodeError:
141 _PROVIDERS = []
143 _prov_registry = {{}}
144 for _c in _PROVIDERS:
145 _n = _c.get("name")
146 if _n and _n not in _prov_registry:
147 _prov_registry[_n] = {{
148 "base_url": _c.get("base_url", ""),
149 "model": _c.get("model", ""),
150 "api_key": _c.get("api_key", ""),
151 "temperature": float(_c.get("temperature", 0.0)),
152 }}
154 max_parallel = int(_BUDGET_SPEC.get("max_parallel", 8))
156 budget = RunBudget(
157 max_llm_calls=int(_BUDGET_SPEC.get("max_llm_calls", 48)),
158 run_timeout_seconds=int(_BUDGET_SPEC.get("timeout_seconds", 240)),
159 )
161 tracer = LangfuseTracer(
162 project_slug=os.getenv("PROJECT_SLUG", "unknown"),
163 pipeline_name=_PIPELINE_NAME,
164 run_id=uuid.uuid4().hex,
165 )
166 tracer.init_trace()
168 invoker = OpenAIInvoker(
169 providers=_prov_registry,
170 max_parallel=max_parallel,
171 budget=budget,
172 tracer=tracer,
173 )
175 run_context = {{
176 "budget": dict(_BUDGET_SPEC),
177 "config": {{}},
178 }}
180 return build_workflow(
181 _PIPELINE_NAME,
182 invoke_agent_fn=invoker,
183 run_context=run_context,
184 )
187async def make_workflow(config):
188 """Factory invoked by LangGraph per run. Builds fresh runtime objects."""
189 return await asyncio.to_thread(_build_workflow_sync)
190'''
193def main() -> None:
194 graphs = {}
196 # --- Legacy: *-pipeline.py ---
197 legacy_names: set[str] = set()
198 if PIPELINE_DIR.exists():
199 for f in sorted(PIPELINE_DIR.glob(PIPELINE_PATTERN)):
200 name = f.stem.replace("-pipeline", "")
201 legacy_names.add(name)
202 api_path = f"/project/.agent-platform/{f.name}"
203 graphs[name] = f"{api_path}:workflow"
204 print(f"[discover] legacy '{name}' ← {api_path}")
206 # --- Uusi: YAML-manifest-pipelinet ---
207 yaml_pipelines = _load_yaml_pipelines(PIPELINE_DIR)
209 # Kirjoita pipeline registry jotta runner.py voi käyttää sitä
210 registry_data = {
211 "manifest_dir": str(PIPELINE_DIR),
212 "pipelines": [p["name"] for p in yaml_pipelines],
213 }
214 REGISTRY_OUTPUT.write_text(json.dumps(registry_data, indent=2))
216 # Luo thin-wrapper-tiedostot /tmp/:iin per YAML-pipeline.
217 # YAML voittaa legacyn jos samanniminen löytyy — uusi totuus.
218 for p in yaml_pipelines:
219 name = p["name"]
220 if name in legacy_names:
221 print(f"[discover] YAML Pipeline '{name}' OVERRIDES legacy *-pipeline.py")
222 wrapper_path = Path(f"/tmp/{name}-pipeline.py")
223 wrapper_path.write_text(
224 _generate_wrapper(name, p["source"], p["budget"])
225 )
226 graphs[name] = f"{wrapper_path}:make_workflow"
227 print(f"[discover] YAML '{name}' → {wrapper_path}")
229 config: dict = {
230 "dependencies": ["."],
231 "graphs": graphs,
232 }
234 # PostgreSQL store
235 if STORE_URI:
236 config["store"] = {
237 "postgres": {"uri": STORE_URI},
238 }
239 print("[discover] Store: postgres")
241 OUTPUT.write_text(json.dumps(config, indent=2))
243 total = len(graphs)
244 yaml_count = len(yaml_pipelines)
245 legacy_count = len(legacy_names)
246 if graphs:
247 print(
248 f"[discover] Total pipelines: {total} "
249 f"({legacy_count} legacy, {yaml_count} YAML)"
250 )
251 print(f"[discover] Config → {OUTPUT}")
252 else:
253 print("[discover] No pipelines found — LangGraph starts with empty graphs")
256if __name__ == "__main__":
257 main()