diff --git a/.gitignore b/.gitignore index 1627f2f..3529f9c 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,8 @@ runs/ forge_output/ .DS_Store *.ipynb_checkpoints/ + +# coverage artifacts +.coverage +coverage.xml +htmlcov/ diff --git a/Makefile b/Makefile index 59bdf98..30d586e 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help install test lint run clean +.PHONY: help install test lint clean cli ui serve # Default command when just running 'make' help: @@ -32,7 +32,12 @@ clean: cli: @echo "Running CLI test query..." - epistemic-forge --query "Is RAG strictly better than Long-Context LLMs?" + epistemic-forge --title "RAG vs Long-Context" --question "Is RAG strictly better than Long-Context LLMs?" + +ui: + @echo "Launching Streamlit dashboard (requires the 'ui' extra)..." + pip install -e ".[ui]" + streamlit run epistemic_forge/ui/app.py serve: @echo "Booting Enterprise API Server..." diff --git a/docs/REVIEW_REMEDIATION.md b/docs/REVIEW_REMEDIATION.md new file mode 100644 index 0000000..50466e7 --- /dev/null +++ b/docs/REVIEW_REMEDIATION.md @@ -0,0 +1,82 @@ +# Review Remediation Plan — Epistemic Forge + +This document maps every finding from the July 22, 2026 technical review to a +concrete fix that has been (or will be) applied. The goal is to address the +**legitimate engineering issues** raised, while being honest about scope: +formal epistemic-logic research (Kripke models, AGM belief revision, trained +PRMs, true MCTS-UCB) is a multi-month research program and is documented here +as a roadmap item rather than overclaimed as "done". + +## Priority 1 — CI/CD is broken & non-enforcing (CRITICAL) +- [x] Rewrite `.github/workflows/ci.yml`: remove `|| echo`, make `pytest` blocking. +- [x] Add coverage reporting (`pytest-cov`) with a floor. +- [x] Make `ruff` a **blocking** lint gate (was `--exit-zero` warning only). +- [x] Add `mypy` (type checking) and `bandit` (security scan) to CI. +- [x] Fix `tests.yml` matrix + blocking tests + coverage. + +## Priority 2 — Critical correctness bugs +- [x] Fix `test_l2_conductor_routing`: it passed `spec.domain.value` (str) where + `SemanticConductor._route_experts(spec: ProjectSpec)` is expected. +- [x] `arsenal_run.py`: move `route_project` import to module top (was imported + inside `run()`). +- [x] `arsenal_run.py`: remove unreachable dead code after `return`. +- [x] `arsenal_run.py`: `run_pipeline` used `raise SystemExit(1)` in library code → + replaced with a re-raised `PipelineError` (CLI still does `sys.exit`). + +## Priority 3 — Code integrity (RED FLAG) +- [x] Delete `patch_run.py` (string-replacement hack). Its *intended* logging was + folded into `arsenal_run.py` properly via a normal edit. + +## Priority 4 — Error handling hygiene +- [x] `memory/economy.py`: bare `except: pass` → `except Exception`. +- [x] `memory/skill_library.py`: bare `except:` in `get_all_skills` → `except Exception`. +- [x] General: no new bare excepts introduced; ruff `E722` enforced in CI. + +## Priority 5 — Security +- [x] `llm.py`: stop writing API keys into `os.environ` globally (redundant — the + key is already passed per-call via `call_params["api_key"]`). +- [x] `llm.py`: add input validation (`validate_messages`) — reject empty/oversized + prompts before dispatch, capping token blow-up. +- [x] `l1_5_adas.py`: harden dynamic schema generation — bound field count, validate + identifiers, reject empty/unsafe blueprints (no arbitrary code execution; only + Pydantic `create_model` from sanitized names). + +## Priority 6 — Dependency hygiene +- [x] Add `ruff`, `pytest-cov`, `mypy`, `bandit`, `types-requests` to `dev`. +- [x] Move `streamlit` out of core deps into an optional `ui` extra (non-UI users + no longer pull a heavy web framework). +- [ ] Lock file (`uv.lock` / `pip-compile`) — tracked as follow-up; see CHANGELOG. + +## Priority 7 — Testing (was <10% coverage) +- [x] Fix existing test + big expansion: `llm` offline fallback, `router`, + `l1_optimizer`, `l2_conductor`, `l1_5_adas`, `l3_search`, `l4_refine`, + `l6_stages`, `skill_library` (with injected fake client), `economy`, + `models` (Pydantic constraints), `cli` arg parsing, and a **full offline + pipeline integration test**. +- [x] Coverage floor enforced in CI (start modest, raise as suite grows). + +## Priority 8 — Architecture honesty +- [x] `pipeline/machine.py`: explicit, data-driven **stage registry** (`L0→L6`) + with per-stage enable predicates, plus a typed `PipelineContext` + (replaces ad-hoc `Dict[str, Any]`), so routing is inspectable, not hidden + in `if` statements. +- [x] Async: `llm.agenerate_structured` (real `litellm.acompletion` + instructor), + parallel `conduct_async` (experts run concurrently via `asyncio.gather`), + and `arun_pipeline` entrypoint — addresses "no async / sequential experts". + +## Priority 9 — Documentation honesty +- [x] README: mark the "85% cost / 80% fewer tokens" figures as **illustrative + targets, not independently benchmarked**; add a "What 'Epistemic' Means + Here" section clarifying it is structured Toulmin prompting, not formal + logic; add architecture diagram. +- [x] `docs/ARCHITECTURE.md` with Mermaid diagram + data contracts. +- [x] `docs/BENCHMARKS.md` describing methodology + current caveats. +- [x] `CHANGELOG.md`. + +## Out of scope / roadmap (documented honestly, NOT overclaimed) +- Formal Dynamic Epistemic Logic / Kripke structures / AGM belief revision. +- Trained Process Reward Model (current "PRM" is LLM-as-Judge prompting). +- True MCTS with UCB / value backpropagation (current L3 is bounded beam enumeration). +- Full multi-agent message-passing communication layer (current "experts" are + sequential strategy-pattern nodes). +- These remain research roadmap items; the naming is now qualified in docs. diff --git a/epistemic_forge/__init__.py b/epistemic_forge/__init__.py index c50803f..878ccae 100644 --- a/epistemic_forge/__init__.py +++ b/epistemic_forge/__init__.py @@ -1,7 +1,7 @@ """Epistemic Forge — ARSENAL-powered research & writing kit.""" -from .pipeline.arsenal_run import ArsenalRun, run_pipeline from .models import Claim, ForgeResult, ProjectSpec +from .pipeline.arsenal_run import ArsenalRun, run_pipeline __version__ = "0.1.0" __all__ = [ diff --git a/epistemic_forge/benchmark/baseline.py b/epistemic_forge/benchmark/baseline.py index 46476b0..7f1cb52 100644 --- a/epistemic_forge/benchmark/baseline.py +++ b/epistemic_forge/benchmark/baseline.py @@ -2,10 +2,8 @@ from __future__ import annotations -from typing import List - -def baseline_answer(title: str, question: str, domain: str, keywords: List[str]) -> str: +def baseline_answer(title: str, question: str, domain: str, keywords: list[str]) -> str: """Produce a short unstructured answer — typical one-shot Q&A quality.""" kw = ", ".join(keywords) if keywords else "the main themes" # Deliberately thin: claim-ish sentence, weak support, no rebuttal structure diff --git a/epistemic_forge/benchmark/llm_judge.py b/epistemic_forge/benchmark/llm_judge.py index 42993c5..e5648db 100644 --- a/epistemic_forge/benchmark/llm_judge.py +++ b/epistemic_forge/benchmark/llm_judge.py @@ -1,35 +1,47 @@ """LLM-as-a-Judge for Automated, Scientific Epistemic Evaluation (Toulmin Model).""" -from typing import Dict, Any -from epistemic_forge.models import JudgeEvaluation + +from typing import Any + +from pydantic import BaseModel, Field + from epistemic_forge.llm import generate_structured from loguru import logger -def evaluate_artifact_quality(question: str, artifact_text: str) -> Dict[str, Any]: + +class JudgeEvaluation(BaseModel): + """Structured output schema for the LLM-as-Judge benchmark.""" + + logical_coherence_score: float = Field(..., ge=0.0, le=1.0) + hallucination_detected: bool = False + critique: str = "" + + +def evaluate_artifact_quality(question: str, artifact_text: str) -> dict[str, Any]: """Uses a stronger model to judge the output based strictly on Toulmin's Model of Argumentation.""" logger.info("⚖️ Initiating strict Toulmin-based evaluation of the final artifact...") - + messages = [ { - "role": "system", + "role": "system", "content": ( "You are an Elite Academic Peer Reviewer specializing in the Toulmin Model of Argumentation. " "Do NOT judge the artifact based on prose or formatting. You must ONLY evaluate the strength of the 'Warrants' (do they bridge the data to the claim?) " "and the validity of the 'Rebuttals/Falsifiers' (are they real weaknesses or just strawmen?)." - ) + ), }, - {"role": "user", "content": f"Core Inquiry: {question}\n\nSubmitted Artifact:\n{artifact_text}\n\nExecute the Toulmin Evaluation."} + {"role": "user", "content": f"Core Inquiry: {question}\n\nSubmitted Artifact:\n{artifact_text}\n\nExecute the Toulmin Evaluation."}, ] - + # We use a robust model for judging, maintaining temp 0.0 for deterministic grading evaluation: JudgeEvaluation = generate_structured( messages=messages, response_model=JudgeEvaluation, - model="openai/gpt-4o-mini", # Standardizing to openrouter/openai model format - api_base="https://openrouter.ai/api/v1" # Enforce OpenRouter for testing consistency + model="openai/gpt-4o-mini", # Standardizing to openrouter/openai model format + api_base="https://openrouter.ai/api/v1", # Enforce OpenRouter for testing consistency ) - + return { "score": evaluation.logical_coherence_score, "hallucination": evaluation.hallucination_detected, - "critique": evaluation.critique + "critique": evaluation.critique, } diff --git a/epistemic_forge/benchmark/metrics.py b/epistemic_forge/benchmark/metrics.py index 35e0644..dab1747 100644 --- a/epistemic_forge/benchmark/metrics.py +++ b/epistemic_forge/benchmark/metrics.py @@ -8,7 +8,7 @@ import re from dataclasses import asdict, dataclass -from typing import Any, Dict, List, Optional +from typing import Any @dataclass @@ -44,18 +44,18 @@ def overall(self) -> float: total += getattr(self, k) * w return round(total, 4) - def to_dict(self) -> Dict[str, Any]: + def to_dict(self) -> dict[str, Any]: d = asdict(self) d["overall"] = self.overall() return d -def _hit(text: str, patterns: List[str]) -> float: +def _hit(text: str, patterns: list[str]) -> float: t = text.lower() return 1.0 if any(p in t for p in patterns) else 0.0 -def _count_hits(text: str, patterns: List[str]) -> int: +def _count_hits(text: str, patterns: list[str]) -> int: t = text.lower() return sum(1 for p in patterns if p in t) @@ -63,7 +63,7 @@ def _count_hits(text: str, patterns: List[str]) -> int: def score_document( text: str, domain: str = "hybrid", - keywords: Optional[List[str]] = None, + keywords: list[str] | None = None, ) -> QualityScores: """Score a free-text answer for Toulmin completeness + packaging quality.""" keywords = keywords or [] diff --git a/epistemic_forge/benchmark/suite.py b/epistemic_forge/benchmark/suite.py index a5e827e..1b2a7e1 100644 --- a/epistemic_forge/benchmark/suite.py +++ b/epistemic_forge/benchmark/suite.py @@ -5,7 +5,7 @@ import json from dataclasses import asdict, dataclass from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any from epistemic_forge.benchmark.baseline import baseline_answer from epistemic_forge.benchmark.metrics import score_document, toulmin_coverage @@ -18,10 +18,10 @@ class BenchCase: title: str question: str domain: str - keywords: List[str] + keywords: list[str] -BENCHMARK_CASES: List[BenchCase] = [ +BENCHMARK_CASES: list[BenchCase] = [ BenchCase( "p1", "Predictive processing and blame", @@ -106,8 +106,8 @@ class CaseResult: forge_toulmin: float lift_overall: float lift_toulmin: float - baseline_scores: Dict[str, Any] - forge_scores: Dict[str, Any] + baseline_scores: dict[str, Any] + forge_scores: dict[str, Any] def _forge_text(case: BenchCase) -> str: @@ -125,9 +125,9 @@ def _forge_text(case: BenchCase) -> str: return "\n\n".join(a.content for a in result.artifacts) -def run_benchmark(cases: Optional[List[BenchCase]] = None) -> Dict[str, Any]: +def run_benchmark(cases: list[BenchCase] | None = None) -> dict[str, Any]: cases = cases or BENCHMARK_CASES - rows: List[CaseResult] = [] + rows: list[CaseResult] = [] for case in cases: base_txt = baseline_answer( case.title, case.question, case.domain, case.keywords @@ -187,7 +187,7 @@ def run_benchmark(cases: Optional[List[BenchCase]] = None) -> Dict[str, Any]: } -def write_benchmark_reports(out_dir: str | Path) -> Dict[str, Any]: +def write_benchmark_reports(out_dir: str | Path) -> dict[str, Any]: out = Path(out_dir) out.mkdir(parents=True, exist_ok=True) report = run_benchmark() diff --git a/epistemic_forge/cli.py b/epistemic_forge/cli.py index 18c8ff9..176936c 100644 --- a/epistemic_forge/cli.py +++ b/epistemic_forge/cli.py @@ -8,14 +8,15 @@ import argparse import sys + from rich.console import Console from rich.panel import Panel from rich.progress import Progress, SpinnerColumn, TextColumn from rich.tree import Tree +from epistemic_forge.memory.economy import budget_manager from epistemic_forge.models import ProjectSpec from epistemic_forge.pipeline.arsenal_run import run_pipeline -from epistemic_forge.memory.economy import budget_manager console = Console() @@ -107,12 +108,13 @@ def main(): "[cyan]L2 Conductor is dispatching semantic experts...", total=None ) - # Execute the core pipeline + # Execute the core pipeline (Hermes routing overrides are passed through) result = run_pipeline( title=spec.title, question=spec.question, domain=spec.domain.value, - # Note: To fully support Hermes, pipeline/arsenal_run.py must pass target_model down. + target_model=args.model, + api_base=args.api_base, ) progress.update( @@ -123,17 +125,20 @@ def main(): # Render output console.print("[bold green]✔ Pipeline Execution Successful.[/bold green]\n") - if hasattr(result, "claims"): + if hasattr(result, "claims") and result.claims: display_claim_lattice(result.claims) - - # SOTA EXPORT - from epistemic_forge.io.export import export_result - export_dir = f'runs/{spec.title.replace(" ", "_").lower()}' - export_result(result, export_dir) + else: console.print( "[yellow]Notice: No claims extracted in the final result.[/yellow]" ) + + # SOTA EXPORT + from epistemic_forge.io.export import export_result + + export_dir = f'runs/{spec.title.replace(" ", "_").lower()}' + export_result(result, export_dir) + console.print(f"\n[bold yellow]💰 {budget_manager.get_report()}[/bold yellow]") except Exception: diff --git a/epistemic_forge/errors.py b/epistemic_forge/errors.py new file mode 100644 index 0000000..9825d1f --- /dev/null +++ b/epistemic_forge/errors.py @@ -0,0 +1,42 @@ +"""Public error types for Epistemic Forge. + +Library code must never call ``sys.exit`` / ``SystemExit``. Callers (CLI, +notebooks, servers) decide how to surface failures; the library only raises +typed exceptions that are safe to catch. +""" + +from __future__ import annotations + + +class EpistemicForgeError(Exception): + """Base class for all library errors.""" + + +class PipelineError(EpistemicForgeError): + """Raised when the L0–L6 pipeline cannot complete. + + Wraps the underlying cause so callers can introspect ``__cause__``. + """ + + def __init__(self, message: str, *, stage: str | None = None) -> None: + self.stage = stage + super().__init__(message) + + +class BudgetExceededError(EpistemicForgeError): + """Raised when the cognitive economy budget is exhausted.""" + + def __init__(self, used: int, limit: int) -> None: + self.used = used + self.limit = limit + super().__init__( + f"Cognitive budget exceeded: {used} tokens used (limit {limit})." + ) + + +class LLMDispatchError(EpistemicForgeError): + """Raised when the universal LLM router cannot fulfil a request.""" + + +class InvalidInputError(EpistemicForgeError): + """Raised when user-supplied input fails validation.""" diff --git a/epistemic_forge/experts/base.py b/epistemic_forge/experts/base.py index 42ad1d2..4aca559 100644 --- a/epistemic_forge/experts/base.py +++ b/epistemic_forge/experts/base.py @@ -5,9 +5,12 @@ and predictable structured outputs within the L2 Conductor. """ +import asyncio from abc import ABC, abstractmethod -from typing import Any, Dict +from typing import Any + from pydantic import BaseModel + from epistemic_forge.models import ProjectSpec @@ -21,7 +24,7 @@ def expert_name(self) -> str: pass @abstractmethod - def analyze(self, spec: ProjectSpec, context: Dict[str, Any]) -> BaseModel: + def analyze(self, spec: ProjectSpec, context: dict[str, Any]) -> BaseModel: """ Executes the expert's specific neuro-symbolic logic. @@ -33,3 +36,12 @@ def analyze(self, spec: ProjectSpec, context: Dict[str, Any]) -> BaseModel: A strictly typed Pydantic BaseModel representing the expert's conclusion. """ pass + + async def analyze_async(self, spec: ProjectSpec, context: dict[str, Any]) -> BaseModel: + """Async counterpart used by the parallel conductor. + + The default implementation runs the sync ``analyze`` in a worker thread + so experts that do network I/O (LLM calls) can execute concurrently. + Experts that perform native async LLM calls should override this. + """ + return await asyncio.to_thread(self.analyze, spec, context) diff --git a/epistemic_forge/experts/claim_expert.py b/epistemic_forge/experts/claim_expert.py index 9ca36b3..0e287d2 100644 --- a/epistemic_forge/experts/claim_expert.py +++ b/epistemic_forge/experts/claim_expert.py @@ -1,10 +1,14 @@ -"""Claim Lattice Expert Implementation (Agentic RAG Grounded).""" -from typing import Dict, Any +"""Claim Lattice Expert Implementation (Grounded with Real-World Search).""" + +from typing import Any + +from loguru import logger + from epistemic_forge.experts.base import EpistemicExpert -from epistemic_forge.models import ProjectSpec, ClaimLatticeOutput from epistemic_forge.llm import generate_structured -from epistemic_forge.tools.search import multi_hop_search -from loguru import logger +from epistemic_forge.models import ClaimLatticeOutput, ProjectSpec +from epistemic_forge.tools.search import search_web + class ClaimLatticeExpert(EpistemicExpert): """Deconstructs the question into a structured, epistemically grounded claim lattice.""" @@ -13,13 +17,15 @@ class ClaimLatticeExpert(EpistemicExpert): def expert_name(self) -> str: return "Grounded_Claim_Lattice_Generator" - def analyze(self, spec: ProjectSpec, context: Dict[str, Any]) -> ClaimLatticeOutput: - """Uses Agentic Multi-Hop Web Search to ground the LLM's claims in reality.""" - - # 1. Fetch real-world context using Multi-Hop Agentic RAG - logger.debug("Gathering multi-hop empirical data (Thesis + Antithesis) from the web...") - live_evidence = multi_hop_search(spec.question, max_hops=2) - + def analyze(self, spec: ProjectSpec, context: dict[str, Any]) -> ClaimLatticeOutput: + """Uses Live Web Search to ground the LLM's claims in reality.""" + + # 1. Fetch real-world context before asking the LLM to build claims + logger.debug("Gathering live empirical data to prevent hallucination...") + search_query = f"{spec.question} scientific consensus" + live_evidence = search_web(search_query, max_results=3) + + messages = [ { "role": "system", diff --git a/epistemic_forge/experts/dialectic_expert.py b/epistemic_forge/experts/dialectic_expert.py index 752d579..11d356a 100644 --- a/epistemic_forge/experts/dialectic_expert.py +++ b/epistemic_forge/experts/dialectic_expert.py @@ -1,10 +1,10 @@ """Hegelian Synthesis Expert Implementation.""" -from typing import Dict, Any +from typing import Any from epistemic_forge.experts.base import EpistemicExpert -from epistemic_forge.models import ProjectSpec, HegelianDialecticOutput from epistemic_forge.llm import generate_structured +from epistemic_forge.models import HegelianDialecticOutput, ProjectSpec class HegelianExpert(EpistemicExpert): @@ -15,7 +15,7 @@ def expert_name(self) -> str: return "Hegelian_Dialectic_Engine" def analyze( - self, spec: ProjectSpec, context: Dict[str, Any] + self, spec: ProjectSpec, context: dict[str, Any] ) -> HegelianDialecticOutput: """Synthesizes the core question by forcing a steelmanned antithesis.""" messages = [ diff --git a/epistemic_forge/experts/freelance_expert.py b/epistemic_forge/experts/freelance_expert.py index a8133fd..5bf2b9e 100644 --- a/epistemic_forge/experts/freelance_expert.py +++ b/epistemic_forge/experts/freelance_expert.py @@ -2,14 +2,14 @@ from __future__ import annotations -from typing import Any, Dict +from typing import Any from epistemic_forge.models import ProjectSpec def build_client_pack( - spec: ProjectSpec, claims_bundle: Dict[str, Any] -) -> Dict[str, Any]: + spec: ProjectSpec, claims_bundle: dict[str, Any] +) -> dict[str, Any]: return { "client_brief": { "goal": spec.question, diff --git a/epistemic_forge/experts/kaggle_expert.py b/epistemic_forge/experts/kaggle_expert.py index e5b4798..66d9568 100644 --- a/epistemic_forge/experts/kaggle_expert.py +++ b/epistemic_forge/experts/kaggle_expert.py @@ -1,10 +1,10 @@ """Data Leakage and Scientific Rigor Expert Implementation.""" -from typing import Dict, Any +from typing import Any from epistemic_forge.experts.base import EpistemicExpert -from epistemic_forge.models import ProjectSpec, RigorSentinelOutput from epistemic_forge.llm import generate_structured +from epistemic_forge.models import ProjectSpec, RigorSentinelOutput class RigorSentinelExpert(EpistemicExpert): @@ -15,7 +15,7 @@ def expert_name(self) -> str: return "Rigor_And_Leakage_Sentinel" def analyze( - self, spec: ProjectSpec, context: Dict[str, Any] + self, spec: ProjectSpec, context: dict[str, Any] ) -> RigorSentinelOutput: """Identifies target leakage and establishes strict falsification metrics.""" messages = [ diff --git a/epistemic_forge/experts/semitic_expert.py b/epistemic_forge/experts/semitic_expert.py index 45ac93f..d6aa043 100644 --- a/epistemic_forge/experts/semitic_expert.py +++ b/epistemic_forge/experts/semitic_expert.py @@ -1,13 +1,14 @@ """L2 Epistemic Synthesis Engine: Semitic NLP & Arabic Logic Expert.""" -from epistemic_forge.models import ProjectSpec, HegelianDialecticOutput +from typing import Any + from epistemic_forge.llm import generate_structured -from typing import Dict, Any +from epistemic_forge.models import HegelianDialecticOutput, ProjectSpec def run_semitic_dialectic( - spec: ProjectSpec, claims_bundle: Dict[str, Any] -) -> Dict[str, Any]: + spec: ProjectSpec, claims_bundle: dict[str, Any] +) -> dict[str, Any]: """Execute dialectic reasoning specifically optimized for Arabic/Semitic morphological logic.""" thesis = spec.question diff --git a/epistemic_forge/experts/writing_expert.py b/epistemic_forge/experts/writing_expert.py index 623d97b..57c69af 100644 --- a/epistemic_forge/experts/writing_expert.py +++ b/epistemic_forge/experts/writing_expert.py @@ -1,13 +1,14 @@ """L2 Epistemic Synthesis Engine: Chain of Density Architect.""" -from epistemic_forge.models import ProjectSpec, ChainOfDensityOutput +from typing import Any + from epistemic_forge.llm import generate_structured -from typing import Dict, Any +from epistemic_forge.models import ChainOfDensityOutput, ProjectSpec def outline_and_draft( - spec: ProjectSpec, claims_bundle: Dict[str, Any], instruction: str -) -> Dict[str, Any]: + spec: ProjectSpec, claims_bundle: dict[str, Any], instruction: str +) -> dict[str, Any]: """Compress the dialectic into a hyper-dense, falsifiable artifact.""" # Gather previous context to compress diff --git a/epistemic_forge/io/export.py b/epistemic_forge/io/export.py index 3c72071..29e9804 100644 --- a/epistemic_forge/io/export.py +++ b/epistemic_forge/io/export.py @@ -5,23 +5,25 @@ - 📓 Jupyter Notebooks (.ipynb) for Kaggle Baselines. - 🧠 Obsidian-compatible JSON graphs for Personal Knowledge Management (PKM). """ -import os import json from pathlib import Path -from loguru import logger + import nbformat as nbf +from loguru import logger + from epistemic_forge.models import ForgeResult + def _generate_mermaid_graph(claims) -> str: """Generates a Mermaid.js flowchart from the Claim Lattice.""" lines = ["graph TD"] for c in claims: - # Pydantic safety + # Pydantic models expose model_dump(); fall back for raw dicts. c_dict = c.model_dump() if hasattr(c, "model_dump") else c c_id = c_dict.get("id", "Unknown") c_text = c_dict.get("text", "").replace('"', "'")[:50] + "..." lines.append(f' {c_id}["{c_id}: {c_text}"]') - + for s in c_dict.get("support", []): lines.append(f' {c_id} -->|Supports| S_{hash(s) % 1000}["{s[:40]}..."]') for o in c_dict.get("objections", []): @@ -32,11 +34,11 @@ def _export_jupyter_notebook(artifact, out_path: Path): """Converts a python/markdown artifact into a runnable Jupyter Notebook.""" nb = nbf.v4.new_notebook() cells = [] - + # Split artifact content heuristically (Markdown vs Code) chunks = artifact.content.split("```python") cells.append(nbf.v4.new_markdown_cell(chunks[0])) - + for chunk in chunks[1:]: if "```" in chunk: code, md = chunk.split("```", 1) @@ -45,7 +47,7 @@ def _export_jupyter_notebook(artifact, out_path: Path): cells.append(nbf.v4.new_markdown_cell(md.strip())) else: cells.append(nbf.v4.new_code_cell(chunk.strip())) - + nb.cells = cells with open(out_path, "w", encoding="utf-8") as f: nbf.write(nb, f) @@ -59,7 +61,7 @@ def export_result(result: ForgeResult, out_dir: str): # 1. Executive Summary & Memo (with Mermaid Graph) memo_path = out / "executive_summary.md" mermaid_code = _generate_mermaid_graph(result.claims) - + memo_content = f"""--- title: {result.spec.title} domain: {result.spec.domain} @@ -77,7 +79,7 @@ def export_result(result: ForgeResult, out_dir: str): **Verdict:** `{result.peer_review.get('verdict', 'Unknown').upper()}` **Critique:** {result.peer_review.get('final_comments', '')} """ - + # Add final synthesized text for art in result.artifacts: if art.name == "Final Synthesis Memo": @@ -90,12 +92,12 @@ def export_result(result: ForgeResult, out_dir: str): json_path = out / "claim_lattice_graph.json" with open(json_path, "w", encoding="utf-8") as f: # Convert claims to a node-edge graph format - nodes = [] - edges = [] + nodes: list = [] + edges: list = [] for c in result.claims: - c_dict = c.model_dump() if hasattr(c, "model_dump") else c + c_dict = c.model_dump() nodes.append({"id": c_dict.get("id"), "label": c_dict.get("text"), "warrant": c_dict.get("epistemic_warrant")}) - + graph_data = {"nodes": nodes, "edges": edges, "metadata": result.peer_review} json.dump(graph_data, f, indent=2) @@ -106,5 +108,5 @@ def export_result(result: ForgeResult, out_dir: str): if not str(nb_path).endswith(".ipynb"): nb_path = nb_path.with_suffix(".ipynb") _export_jupyter_notebook(art, nb_path) - + logger.success(f"💾 Export Complete. Files ready in {out.resolve()}") diff --git a/epistemic_forge/llm.py b/epistemic_forge/llm.py index 383a33f..2df87c1 100644 --- a/epistemic_forge/llm.py +++ b/epistemic_forge/llm.py @@ -2,33 +2,88 @@ Absolute flexibility: Use ANY model from ANY provider with zero code changes. Supports standard formats: 'openai/gpt-4o', 'anthropic/claude-3-sonnet', 'ollama/llama3', 'azure/...', etc. + +Security notes +-------------- +* API keys are passed **per call** via ``call_params["api_key"]`` and are never + written into ``os.environ`` (which would leak them to all child processes). +* User-supplied prompts are validated (non-empty, bounded size) before dispatch + to prevent runaway token usage / injection of malformed payloads. """ -from pydantic import BaseModel +from __future__ import annotations + +import os +from typing import Any, TypeVar + +import instructor +from litellm import acompletion, completion from loguru import logger +from pydantic import BaseModel from tenacity import retry, stop_after_attempt, wait_exponential -import instructor -from litellm import completion + +from epistemic_forge.errors import InvalidInputError, LLMDispatchError from epistemic_forge.memory.economy import budget_manager -import os + +T = TypeVar("T", bound=BaseModel) + +# Upper bound on total prompt characters to protect against accidental +# token blow-up / abusive inputs. ~50k chars is generous for our use case. +MAX_PROMPT_CHARS = int(os.getenv("EF_MAX_PROMPT_CHARS", "50000")) # We patch instructor to use LiteLLM's universal completion directly! # This is the "Hermes" way: we don't switch clients, we use one universal proxy. +client: Any = None try: client = instructor.from_litellm(completion) -except Exception as e: +except Exception as e: # pragma: no cover - depends on install logger.warning(f"LiteLLM/Instructor initialization failed: {e}") - client = None +aclient: Any = None +try: + aclient = instructor.from_litellm(acompletion) +except Exception as e: # pragma: no cover - depends on install + logger.warning(f"Async LiteLLM/Instructor initialization failed: {e}") + + +def validate_messages(messages: list[dict]) -> None: + """Validate prompt structure before dispatch. -def _offline_fallback(response_model: type[BaseModel], messages: list) -> BaseModel: + Raises :class:`InvalidInputError` on empty or oversized prompts. This is a + defense-in-depth measure: it bounds cost and rejects malformed payloads. + """ + if not messages: + raise InvalidInputError("Cannot dispatch an empty message list.") + total = 0 + for i, msg in enumerate(messages): + if not isinstance(msg, dict): + raise InvalidInputError(f"Message #{i} is not a mapping.") + if "role" not in msg: + raise InvalidInputError(f"Message #{i} is missing 'role'.") + content = msg.get("content") + if content is None: + continue + if not isinstance(content, str): + raise InvalidInputError(f"Message #{i} content must be a string.") + total += len(content) + if total > MAX_PROMPT_CHARS: + raise InvalidInputError( + f"Prompt too large ({total} chars > limit {MAX_PROMPT_CHARS})." + ) + + +def _offline_fallback(response_model: type[T], messages: list) -> T: """Deterministic fallback for CI/offline runs without provider credentials.""" - prompt = "" - if messages: - prompt = str(messages[-1].get("content", "")) - lower = prompt.lower() model_name = response_model.__name__ + if model_name == "RouteDecision": + return response_model( + families=["mock"], + activate={"l3_search": True, "l4_refine": True, "l6_stages": True}, + l1_mode="opro", + l3_mode="tot", + rationale="Offline deterministic routing (all heavy layers on).", + ) if model_name == "OptimizedInstruction": return response_model( meta_prompt=( @@ -142,7 +197,31 @@ def _offline_fallback(response_model: type[BaseModel], messages: list) -> BaseMo robust_baseline="Simple regularized model with strict CV and leakage audit.", ) - return response_model() + # Generic fallback: fill required fields with safe placeholders so the + # offline path never raises ValidationError (e.g. for ADAS-built models). + try: + data: dict = {} + for name, fld in response_model.model_fields.items(): + if not fld.is_required(): + continue + ann = fld.annotation + if ann is str or (isinstance(ann, type) and issubclass(ann, str)): + data[name] = f"[offline] {name}" + elif ann is float: + data[name] = 0.5 + elif ann is int: + data[name] = 0 + elif ann is bool: + data[name] = False + elif ann in (list, dict) or getattr(ann, "__origin__", None) in (list, dict): + data[name] = [] if ann in (list, list) or getattr( + ann, "__origin__", None + ) is list else {} + else: + data[name] = "" + return response_model(**data) + except Exception: + return response_model() def _missing_credentials(model: str, api_key: str | None) -> bool: @@ -161,50 +240,46 @@ def _missing_credentials(model: str, api_key: str | None) -> bool: @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def generate_structured( messages: list, - response_model: type[BaseModel], + response_model: type[T], model: str = "gpt-4o-mini", # Can be ANY litellm supported string, e.g., 'ollama/llama3' temperature: float = 0.0, seed: int = 42, - api_base: str = None, - api_key: str = None, + api_base: str | None = None, + api_key: str | None = None, **kwargs, -) -> BaseModel: +) -> T: """ Universal Hermes-style Structured Extraction. You can pass the provider in the model string (e.g., 'anthropic/claude-3-opus-20240229'). """ + validate_messages(messages) + if _missing_credentials(model, api_key): logger.warning( f"🌐 [Hermes Router] Missing credentials for [{model}], using deterministic fallback." ) return _offline_fallback(response_model, messages) - if not client: - raise ValueError("Universal LLM Router is not initialized.") + if client is None: + raise LLMDispatchError("Universal LLM Router is not initialized.") try: logger.debug( f"🌐 [Hermes Router] Dispatching to [{model}] for schema [{response_model.__name__}]..." ) - call_params = { + call_params: dict[str, Any] = { "model": model, "messages": messages, "response_model": response_model, "temperature": temperature, } - # Inject optional routing Overrides + # Inject optional routing Overrides (api_key is passed per-call only; + # we deliberately do NOT write it into os.environ). if api_base: call_params["api_base"] = api_base if api_key: - # Force it into environment for litellm - if "openrouter" in model: - os.environ["OPENROUTER_API_KEY"] = api_key - elif "gemini" in model: - os.environ["GEMINI_API_KEY"] = api_key - else: - os.environ["OPENAI_API_KEY"] = api_key call_params["api_key"] = api_key # Add any extra kwargs (like top_p, max_tokens) dynamically @@ -219,7 +294,52 @@ def generate_structured( return response except Exception as e: - logger.error( - f"🌐 [Hermes Router] Critical Failure for model '{model}': {str(e)}" - ) + logger.error(f"🌐 [Hermes Router] Critical Failure for model '{model}': {e}") + raise + + +@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) +async def agenerate_structured( + messages: list, + response_model: type[T], + model: str = "gpt-4o-mini", + temperature: float = 0.0, + seed: int = 42, + api_base: str | None = None, + api_key: str | None = None, + **kwargs, +) -> T: + """Async twin of :func:`generate_structured` using ``litellm.acompletion``. + + Used by the parallel expert conductor so multiple experts can dispatch LLM + calls concurrently instead of sequentially. + """ + validate_messages(messages) + + if _missing_credentials(model, api_key): + return _offline_fallback(response_model, messages) + + if aclient is None: + raise LLMDispatchError("Async Universal LLM Router is not initialized.") + + try: + call_params: dict[str, Any] = { + "model": model, + "messages": messages, + "response_model": response_model, + "temperature": temperature, + } + if api_base: + call_params["api_base"] = api_base + if api_key: + call_params["api_key"] = api_key + call_params.update(kwargs) + if "gpt" in model or "llama" in model: + call_params["seed"] = seed + + response = await aclient.chat.completions.create(**call_params) + budget_manager.add_usage(response._raw_response, model) + return response + except Exception as e: + logger.error(f"🌐 [Hermes Router] Async failure for model '{model}': {e}") raise diff --git a/epistemic_forge/memory/economy.py b/epistemic_forge/memory/economy.py index 493255c..ae5d01f 100644 --- a/epistemic_forge/memory/economy.py +++ b/epistemic_forge/memory/economy.py @@ -4,8 +4,8 @@ or halts if the API costs exceed the specified limits. """ -from loguru import logger import litellm +from loguru import logger class TokenBudgetManager: @@ -13,7 +13,7 @@ class TokenBudgetManager: def __new__(cls): if cls._instance is None: - cls._instance = super(TokenBudgetManager, cls).__new__(cls) + cls._instance = super().__new__(cls) cls._instance.reset() return cls._instance @@ -37,7 +37,8 @@ def add_usage(self, response_object, model: str): completion_response=response_object ) self.total_cost += cost - except: + except Exception: # noqa: S110 best-effort cost tracking + # Cost calculation is best-effort; ignore provider mismatches. pass except Exception as e: logger.debug(f"Could not track token usage: {e}") diff --git a/epistemic_forge/memory/reflexion_store.py b/epistemic_forge/memory/reflexion_store.py index 9c249db..05e13d5 100644 --- a/epistemic_forge/memory/reflexion_store.py +++ b/epistemic_forge/memory/reflexion_store.py @@ -2,15 +2,13 @@ from __future__ import annotations -from typing import List - from epistemic_forge.models import Reflection class ReflexionStore: def __init__(self, window: int = 3): self.window = window - self._items: List[Reflection] = [] + self._items: list[Reflection] = [] def add(self, reflection: Reflection) -> None: self._items.append(reflection) @@ -27,7 +25,7 @@ def as_prompt_block(self) -> str: ) return "\n".join(lines) - def all(self) -> List[Reflection]: + def all(self) -> list[Reflection]: return list(self._items) def reflect_on_failure(self, trial: int, score: float, notes: str) -> Reflection: diff --git a/epistemic_forge/memory/skill_library.py b/epistemic_forge/memory/skill_library.py index d2dc7ad..56000f8 100644 --- a/epistemic_forge/memory/skill_library.py +++ b/epistemic_forge/memory/skill_library.py @@ -1,30 +1,50 @@ """L5 Procedural Memory — SOTA Vector Database (ChromaDB) Skill Library. -Replaces hardcoded lists with a persistent semantic memory store. -The system learns over time by saving successful cognitive strategies and +Replaces hardcoded lists with a persistent semantic memory store. +The system learns over time by saving successful cognitive strategies and retrieving them dynamically for future similar inquiries (Voyager-style). """ -from typing import List, Optional -import os +from typing import Any, Optional, cast + import chromadb from loguru import logger + from epistemic_forge.models import Skill + class SkillLibrary: - """Persistent Vector Memory for Procedural Skills.""" - - def __init__(self, persist_dir: str = ".forge_memory"): + """Persistent Vector Memory for Procedural Skills. + + Accepts an optional ``client`` (a ``chromadb.Client`` / ``PersistentClient``) + so tests can inject an in-memory or fake client. When ``persist_dir`` is + ``None`` an ephemeral in-memory client is used (no disk writes). + """ + + def __init__( + self, + persist_dir: str | None = ".forge_memory", + client: Optional["chromadb.api.ClientAPI"] = None, + ): self.persist_dir = persist_dir # Initialize ChromaDB client (local persistent storage) try: - self.client = chromadb.PersistentClient(path=self.persist_dir) - self.collection = self.client.get_or_create_collection(name="epistemic_skills") + if client is not None: + self.client = client + elif persist_dir is None: + self.client = chromadb.Client() + else: + self.client = chromadb.PersistentClient(path=persist_dir) + self.collection = self.client.get_or_create_collection( + name="epistemic_skills" + ) logger.info("🧠 L5 Vector Memory (ChromaDB) successfully initialized.") except Exception as e: - logger.warning(f"ChromaDB initialization failed: {e}. Running with ephemeral memory.") + logger.warning( + f"ChromaDB initialization failed: {e}. Running with ephemeral memory." + ) self.client = chromadb.Client() self.collection = self.client.create_collection(name="epistemic_skills") - + def add_skill(self, skill: Skill): """Saves a new cognitive skill to the vector database.""" try: @@ -37,44 +57,49 @@ def add_skill(self, skill: Skill): except Exception as e: logger.debug(f"Failed to commit skill: {e}") - def retrieve_relevant_skills(self, query: str, n_results: int = 2) -> List[Skill]: + def retrieve_relevant_skills(self, query: str, n_results: int = 2) -> list[Skill]: """Performs semantic search to find skills relevant to the current inquiry.""" try: if self.collection.count() == 0: return [] - - results = self.collection.query( + + results = cast(dict[str, Any], self.collection.query( query_texts=[query], - n_results=min(n_results, self.collection.count()) - ) - - skills = [] - if results and 'metadatas' in results and results['metadatas'][0]: - for meta in results['metadatas'][0]: + n_results=min(n_results, self.collection.count()), + )) + + skills: list[Skill] = [] + metadatas = results.get("metadatas") or [] + if metadatas and metadatas[0]: + for meta in metadatas[0]: + meta = cast(dict[str, Any], meta) skills.append(Skill( - name=meta.get("name", "unknown"), - description="", # Recovered from doc if needed - code=meta.get("code", ""), - tags=meta.get("tags", "").split(",") + name=str(meta.get("name", "unknown")), + description="", # Recovered from doc if needed + code=str(meta.get("code", "")), + tags=str(meta.get("tags", "")).split(","), )) return skills except Exception as e: logger.warning(f"L5 Retrieval failed: {e}") return [] - def get_all_skills(self) -> List[Skill]: + def get_all_skills(self) -> list[Skill]: """Returns all skills (for debugging or exact matching).""" try: - results = self.collection.get() - skills = [] - if results and 'metadatas' in results: - for meta in results['metadatas']: + results = cast(dict[str, Any], self.collection.get()) + skills: list[Skill] = [] + metadatas = results.get("metadatas") or [] + if metadatas: + for meta in metadatas: + meta = cast(dict[str, Any], meta) skills.append(Skill( - name=meta.get("name", "unknown"), + name=str(meta.get("name", "unknown")), description="", - code=meta.get("code", ""), - tags=meta.get("tags", "").split(",") + code=str(meta.get("code", "")), + tags=str(meta.get("tags", "")).split(","), )) return skills - except: + except Exception as e: + logger.warning(f"L5 get_all_skills failed: {e}") return [] diff --git a/epistemic_forge/models.py b/epistemic_forge/models.py index eb1309c..4e73ed2 100644 --- a/epistemic_forge/models.py +++ b/epistemic_forge/models.py @@ -2,9 +2,10 @@ from __future__ import annotations -from pydantic import BaseModel, Field from enum import Enum -from typing import Any, Dict, List, Optional +from typing import Any + +from pydantic import BaseModel, Field class Domain(str, Enum): @@ -30,8 +31,8 @@ class ProjectSpec(BaseModel): question: str domain: Domain = Domain.HYBRID audience: str = "technical peer / client" - constraints: List[str] = Field(default_factory=list) - keywords: List[str] = Field(default_factory=list) + constraints: list[str] = Field(default_factory=list) + keywords: list[str] = Field(default_factory=list) budget_tokens: int = 8000 max_trials: int = 3 enable_opro_style: bool = True @@ -39,8 +40,8 @@ class ProjectSpec(BaseModel): # Hermes Universal Routing Overrides target_model: str = "gpt-4o-mini" - api_base: str = None - api_key: str = None + api_base: str | None = None + api_key: str | None = None class Claim(BaseModel): @@ -54,10 +55,10 @@ class Claim(BaseModel): potential_falsifier: str = Field( description="What specific evidence or scenario would prove this claim wrong?" ) - support: List[str] = Field( + support: list[str] = Field( default_factory=list, description="Sub-arguments supporting this claim." ) - objections: List[str] = Field( + objections: list[str] = Field( default_factory=list, description="Valid counter-arguments against this claim." ) confidence: Confidence = Field(default=Confidence.LIKELY) @@ -66,8 +67,8 @@ class Claim(BaseModel): class RouteDecision(BaseModel): """L0 router output.""" - families: List[str] - activate: Dict[str, bool] + families: list[str] + activate: dict[str, bool] rationale: str l1_mode: str = "ape" # ape | opro | cascade | off l3_mode: str = "tot" # tot | lats | cascade | off @@ -79,9 +80,9 @@ class SearchNode(BaseModel): id: str thought: str value: float - parent_id: Optional[str] = None - children: List[str] = Field(default_factory=list) - meta: Dict[str, Any] = Field(default_factory=dict) + parent_id: str | None = None + children: list[str] = Field(default_factory=list) + meta: dict[str, Any] = Field(default_factory=dict) class Reflection(BaseModel): @@ -99,7 +100,7 @@ class Skill(BaseModel): name: str description: str code: str - tags: List[str] = Field(default_factory=list) + tags: list[str] = Field(default_factory=list) class StageArtifact(BaseModel): @@ -115,13 +116,13 @@ class ForgeResult(BaseModel): spec: ProjectSpec route: RouteDecision instruction: str - claims: List[Claim] - search_trace: List[SearchNode] - reflections: List[Reflection] - skills_used: List[str] - artifacts: List[StageArtifact] - peer_review: Dict[str, Any] - trial_log: List[Dict[str, Any]] = Field(default_factory=list) + claims: list[Claim] + search_trace: list[SearchNode] + reflections: list[Reflection] + skills_used: list[str] + artifacts: list[StageArtifact] + peer_review: dict[str, Any] + trial_log: list[dict[str, Any]] = Field(default_factory=list) final_score: float = 0.0 @@ -143,7 +144,7 @@ class KaggleExpertOutput(BaseModel): baseline_architecture: str = Field( description="Simple, robust baseline model recommendation." ) - critical_flaws: List[str] = Field( + critical_flaws: list[str] = Field( description="Potential pitfalls in the feature engineering." ) @@ -154,7 +155,7 @@ class DialecticExpertOutput(BaseModel): core_thesis: str antithesis: str synthesis: str - logical_fallacies_avoided: List[str] + logical_fallacies_avoided: list[str] class WritingExpertOutput(BaseModel): @@ -162,19 +163,18 @@ class WritingExpertOutput(BaseModel): tone_consistency_score: float structural_flow: str - draft_paragraphs: List[str] + draft_paragraphs: list[str] # ========================================== # L2 SYNTHESIS ENGINE SCHEMAS (NEURO-SYMBOLIC) # ========================================== -from pydantic import BaseModel, Field class RigorSentinelOutput(BaseModel): """Strict schema for the Leakage & Rigor Sentinel (formerly Kaggle Expert).""" - epistemic_blind_spots: List[str] = Field( + epistemic_blind_spots: list[str] = Field( description="Hidden assumptions or target leakage risks in the user's premise." ) falsification_metric: str = Field( @@ -194,7 +194,7 @@ class HegelianDialecticOutput(BaseModel): synthesis_resolution: str = Field( description="The nuanced truth that reconciles the thesis and the antithesis." ) - remaining_uncertainties: List[str] = Field( + remaining_uncertainties: list[str] = Field( description="Questions that still lack sufficient evidence." ) epistemic_confidence: float = Field( @@ -221,7 +221,7 @@ class ChainOfDensityOutput(BaseModel): class ClaimLatticeOutput(BaseModel): """A strict output schema containing multiple grounded claims.""" - claims: List[Claim] + claims: list[Claim] lattice_summary: str = Field( description="A short summary of how these claims interlock." ) @@ -236,7 +236,7 @@ class OptimizedInstruction(BaseModel): rationale: str = Field( description="Why this instruction will yield better results than a generic prompt." ) - expected_failure_modes: List[str] = Field( + expected_failure_modes: list[str] = Field( description="What the LLM might get wrong if not guided properly." ) @@ -250,7 +250,7 @@ class ThoughtProposal(BaseModel): class ThoughtProposalsOutput(BaseModel): """Collection of proposed thoughts for branching.""" - proposals: List[ThoughtProposal] + proposals: list[ThoughtProposal] class ThoughtEvaluation(BaseModel): @@ -275,7 +275,7 @@ class RefinementFeedback(BaseModel): le=1.0, description="Presence of explicit boundaries, assumptions, and falsifiers.", ) - critical_flaws: List[str] = Field( + critical_flaws: list[str] = Field( description="List of logical leaps, hallucinations, or unsupported claims." ) passes_threshold: bool = Field( @@ -289,7 +289,7 @@ class RefinedArtifact(BaseModel): improved_text: str = Field( description="The heavily revised, flawless version of the text." ) - changes_made: List[str] = Field(description="What was fixed based on the critique.") + changes_made: list[str] = Field(description="What was fixed based on the critique.") class PeerReviewScores(BaseModel): @@ -305,7 +305,7 @@ class FinalPeerReview(BaseModel): scores: PeerReviewScores overall_score: float = Field(ge=0.0, le=1.0, description="Average of all metrics.") - revision_needed: List[str] = Field(description="Areas that still need work if any.") + revision_needed: list[str] = Field(description="Areas that still need work if any.") verdict: str = Field( description="Must be one of: 'accept', 'accept_with_minor_revisions', 'major_revisions', 'reject'" ) @@ -321,7 +321,7 @@ class DynamicExpertSchema(BaseModel): description="Name of the expert, e.g., 'QuantumMechanicsExpert'" ) expert_description: str = Field(description="What this expert analyzes.") - fields_to_extract: List[Dict[str, str]] = Field( + fields_to_extract: list[dict[str, str]] = Field( description="List of fields the expert must extract. Format: {'field_name': 'description'}" ) system_prompt: str = Field( @@ -331,6 +331,6 @@ class DynamicExpertSchema(BaseModel): class SearchResult(BaseModel): best_thought: str - nodes: List[SearchNode] + nodes: list[SearchNode] mode_used: str score: float diff --git a/epistemic_forge/pipeline/arsenal_run.py b/epistemic_forge/pipeline/arsenal_run.py index ba45a50..e5f6714 100644 --- a/epistemic_forge/pipeline/arsenal_run.py +++ b/epistemic_forge/pipeline/arsenal_run.py @@ -1,15 +1,20 @@ from __future__ import annotations -from loguru import logger + +import asyncio from dataclasses import dataclass -from typing import List, Optional +from loguru import logger + +from epistemic_forge.errors import PipelineError from epistemic_forge.memory.reflexion_store import ReflexionStore from epistemic_forge.memory.skill_library import SkillLibrary -from epistemic_forge.models import Domain, ForgeResult, ProjectSpec +from epistemic_forge.models import Domain, ForgeResult, ProjectSpec, RouteDecision from epistemic_forge.pipeline.l1_optimizer import optimize_instruction -from epistemic_forge.pipeline.l2_conductor import conduct +from epistemic_forge.pipeline.l2_conductor import conduct_async from epistemic_forge.pipeline.l3_search import explore from epistemic_forge.pipeline.l6_stages import produce_artifacts +from epistemic_forge.pipeline.machine import PipelineContext, execute_pipeline +from epistemic_forge.pipeline.router import route_project @dataclass @@ -20,68 +25,79 @@ class ArsenalRun: reflexion: ReflexionStore @classmethod - def create(cls) -> "ArsenalRun": + def create(cls) -> ArsenalRun: return cls(skills=SkillLibrary(), reflexion=ReflexionStore(window=3)) - def run(self, spec: ProjectSpec, out_dir: Optional[str] = None) -> ForgeResult: + def run(self, spec: ProjectSpec, out_dir: str | None = None) -> ForgeResult: logger.info(f"Starting ArsenalRun for: {spec.title}") - from epistemic_forge.models import RouteDecision - - # L0: Semantic Router - route = route_project(spec) - logger.info(f"Pipeline dynamically configured: {route.rationale}") - - # L1: OPRO Optimizer - instruction = optimize_instruction(spec) - - # L2: Conductor & Experts - conducted = conduct(spec, {'instruction': instruction, 'skills': []}) - - # L3: Tree Search with PRM (Only if activated by L0) - search_nodes = [] - best_thought = str(conducted) - final_score = 0.5 - - if route.activate.get("l3_search", True): - search = explore(spec, conducted, beam=3, steps=2) - search_nodes = search.nodes - best_thought = search.best_thought - final_score = search.score - - # L6: Stage Artifacts and Review (incorporates L4 Self-Refine internally) - artifacts, review, score = produce_artifacts(spec, best_thought, conducted, final_score) - + ctx = PipelineContext(spec=spec) + try: + ctx = execute_pipeline(ctx) + except PipelineError: + raise + except Exception as exc: + raise PipelineError( + f"Pipeline failed for '{spec.title}': {exc}", stage="run" + ) from exc + + route = ctx.route or RouteDecision( + families=["mock"], activate={}, rationale="mock" + ) return ForgeResult( spec=spec, route=route, - instruction=instruction, - claims=conducted.get('ClaimLatticeExpert', {}).get('claims', []), - search_trace=search_nodes, + instruction=ctx.instruction, + claims=ctx.conducted.get("Grounded_Claim_Lattice_Generator", {}).get( + "claims", [] + ), + search_trace=list(ctx.search_result.nodes) if ctx.search_result else [], reflections=self.reflexion.all(), skills_used=[], - artifacts=artifacts, - peer_review=review, - final_score=score + artifacts=ctx.artifacts, + peer_review=ctx.review or {}, + final_score=ctx.final_score, ) - instruction = optimize_instruction(spec) - conducted = conduct(spec, {"instruction": instruction, "skills": []}) - search = explore(spec, conducted, beam=3, steps=2) - artifacts, review, score = produce_artifacts( - spec, search.best_thought, conducted, search.score - ) + + async def arun(self, spec: ProjectSpec) -> ForgeResult: + """Async variant: runs L2 experts concurrently via asyncio.gather.""" + logger.info(f"Starting async ArsenalRun for: {spec.title}") + ctx = PipelineContext(spec=spec) + try: + ctx.route = await asyncio.to_thread(route_project, spec) + ctx.instruction = await asyncio.to_thread(optimize_instruction, spec) + ctx.conducted = await conduct_async( + spec, {"instruction": ctx.instruction, "skills": []} + ) + if ctx.route.activate.get("l3_search", True): + ctx.search_result = await asyncio.to_thread( + explore, spec, ctx.conducted, 3, 2 + ) + + best = ctx.search_result.best_thought if ctx.search_result else str(ctx.conducted) + prior = ctx.search_result.score if ctx.search_result else 0.5 + artifacts, review, score = await asyncio.to_thread( + produce_artifacts, spec, best, ctx.conducted, prior + ) + ctx.artifacts, ctx.review, ctx.final_score = artifacts, review, score + except Exception as exc: + raise PipelineError( + f"Async pipeline failed for '{spec.title}': {exc}", stage="arun" + ) from exc return ForgeResult( spec=spec, - route=RouteDecision(families=["mock"], activate={}, rationale="mock"), - instruction=instruction, - claims=conducted.get('ClaimLatticeExpert', {}).get('claims', []), - search_trace=search.nodes, + route=ctx.route, + instruction=ctx.instruction, + claims=ctx.conducted.get("Grounded_Claim_Lattice_Generator", {}).get( + "claims", [] + ), + search_trace=list(ctx.search_result.nodes) if ctx.search_result else [], reflections=self.reflexion.all(), skills_used=[], - artifacts=artifacts, - peer_review=review, - final_score=score, + artifacts=ctx.artifacts, + peer_review=ctx.review or {}, + final_score=ctx.final_score, ) @@ -90,11 +106,18 @@ async def run_pipeline( question: str, domain: str = "hybrid", audience: str = "technical peer / client", - keywords: Optional[List[str]] = None, - constraints: Optional[List[str]] = None, + keywords: list[str] | None = None, + constraints: list[str] | None = None, max_trials: int = 3, + target_model: str = "gpt-4o-mini", + api_base: str | None = None, + async_run: bool = False, ) -> ForgeResult: - """Convenience API.""" + """Convenience API for library and CLI users. + + Library code must not call ``sys.exit``; on failure we raise a typed + :class:`PipelineError` so callers decide how to surface it. + """ try: dom = Domain(domain) except ValueError: @@ -107,13 +130,17 @@ async def run_pipeline( keywords=keywords or [], constraints=constraints or [], max_trials=max_trials, + target_model=target_model, + api_base=api_base, ) + logger.info(f"Starting Epistemic Forge Pipeline for: '{title}'") + runner = ArsenalRun.create() try: - logger.info(f"Starting Epistemic Forge Pipeline for: '{title}'") - async for event in ArsenalRun.create().run(spec): - yield event - logger.success("Pipeline execution completed successfully.") - return result - except Exception as e: - logger.exception(f"Critical Pipeline Failure: {str(e)}") - raise SystemExit(1) + result = asyncio.run(runner.arun(spec)) if async_run else runner.run(spec) + except PipelineError: + raise + except Exception as exc: + logger.exception(f"Critical Pipeline Failure: {exc}") + raise PipelineError(f"Pipeline failed: {exc}") from exc + logger.success("Pipeline execution completed successfully.") + return result diff --git a/epistemic_forge/pipeline/l1_5_adas.py b/epistemic_forge/pipeline/l1_5_adas.py index c175601..df5b97c 100644 --- a/epistemic_forge/pipeline/l1_5_adas.py +++ b/epistemic_forge/pipeline/l1_5_adas.py @@ -3,20 +3,65 @@ Self-Evolving Architecture: If the static experts (Hegelian, Rigor Sentinel) are insufficient for a highly specific query, this layer dynamically writes a custom Pydantic Schema and instantiates a new Expert Node on the fly. + +Safety +------ +The LLM-designed blueprint is **not** executed as code. We only use it to build +a Pydantic model via ``pydantic.create_model`` from *sanitized* field names, and +we bound the number of fields and validate every identifier. A malformed +blueprint raises instead of silently producing a broken expert. """ -from typing import Dict, Any -from pydantic import BaseModel, create_model, Field +from typing import Any + from loguru import logger +from pydantic import BaseModel, Field, create_model -from epistemic_forge.models import ProjectSpec, DynamicExpertSchema -from epistemic_forge.llm import generate_structured +from epistemic_forge.errors import InvalidInputError from epistemic_forge.experts.base import EpistemicExpert +from epistemic_forge.llm import agenerate_structured, generate_structured +from epistemic_forge.models import DynamicExpertSchema, ProjectSpec + +# Hard caps to keep a dynamically generated expert cheap and safe. +MAX_DYNAMIC_FIELDS = 12 + + +def _safe_identifier(name: str) -> str: + """Return a valid Python identifier derived from ``name``, or '' if none.""" + cleaned = "".join(c for c in name if c.isalnum() or c == "_").lower() + if not cleaned: + return "" + if cleaned[0].isdigit(): + cleaned = "_" + cleaned + return cleaned + + +def _build_field_definitions(blueprint: DynamicExpertSchema) -> dict[str, Any]: + """Sanitize the LLM-provided field list into Pydantic field definitions.""" + field_definitions: dict[str, Any] = {} + seen: set[str] = set() + raw = blueprint.fields_to_extract or [] + if len(raw) > MAX_DYNAMIC_FIELDS: + logger.warning( + f"ADAS blueprint had {len(raw)} fields; truncating to {MAX_DYNAMIC_FIELDS}." + ) + raw = raw[:MAX_DYNAMIC_FIELDS] + + for f in raw: + for fname, fdesc in f.items(): + safe = _safe_identifier(fname) + if not safe or safe in seen: + continue + seen.add(safe) + field_definitions[safe] = (str, Field(description=fdesc or fname)) + + if not field_definitions: + raise InvalidInputError("ADAS blueprint produced no valid fields.") + return field_definitions def generate_dynamic_expert(spec: ProjectSpec) -> EpistemicExpert: """Uses LLM to design a custom expert class and Pydantic schema.""" - logger.info( "🧬 L1.5 ADAS: Generating a custom Self-Evolving Expert tailored to this query..." ) @@ -24,7 +69,11 @@ def generate_dynamic_expert(spec: ProjectSpec) -> EpistemicExpert: messages = [ { "role": "system", - "content": "You are a Meta-Architect (ADAS). Your job is to design a highly specialized 'AI Expert Node' that is perfectly tailored to solve the user's specific problem. Define its output schema and its system prompt.", + "content": ( + "You are a Meta-Architect (ADAS). Your job is to design a highly " + "specialized 'AI Expert Node' that is perfectly tailored to solve the " + "user's specific problem. Define its output schema and its system prompt." + ), }, { "role": "user", @@ -41,29 +90,22 @@ def generate_dynamic_expert(spec: ProjectSpec) -> EpistemicExpert: logger.debug(f"🧬 Blueprint acquired: {blueprint.expert_class_name}") - # Dynamically create the Pydantic Model based on the LLM's design - field_definitions = {} - for f in blueprint.fields_to_extract: - for fname, fdesc in f.items(): - # Clean field name to be a valid python identifier - safe_fname = "".join(c for c in fname if c.isalnum() or c == "_").lower() - if safe_fname: - field_definitions[safe_fname] = (str, Field(description=fdesc)) + class_name = _safe_identifier(blueprint.expert_class_name) or "DynamicExpert" + field_definitions = _build_field_definitions(blueprint) - DynamicModel = create_model( - f"{blueprint.expert_class_name}Output", **field_definitions - ) + DynamicModel = create_model(f"{class_name}Output", **field_definitions) + system_prompt = blueprint.system_prompt or "Analyze the problem rigorously." # Create the Expert Class dynamically class DynamicallyGeneratedExpert(EpistemicExpert): @property def expert_name(self) -> str: - return blueprint.expert_class_name + return class_name - def analyze(self, spec: ProjectSpec, context: Dict[str, Any]) -> BaseModel: + def analyze(self, spec: ProjectSpec, context: dict[str, Any]) -> BaseModel: logger.debug(f"Activating dynamically generated expert: {self.expert_name}") msgs = [ - {"role": "system", "content": blueprint.system_prompt}, + {"role": "system", "content": system_prompt}, { "role": "user", "content": f"Problem: {spec.question}\nContext: {context}\nAnalyze this.", @@ -76,4 +118,24 @@ def analyze(self, spec: ProjectSpec, context: Dict[str, Any]) -> BaseModel: api_base=spec.api_base, ) + async def analyze_async( + self, spec: ProjectSpec, context: dict[str, Any] + ) -> BaseModel: + logger.debug( + f"Activating dynamically generated expert (async): {self.expert_name}" + ) + msgs = [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": f"Problem: {spec.question}\nContext: {context}\nAnalyze this.", + }, + ] + return await agenerate_structured( + messages=msgs, + response_model=DynamicModel, + model=spec.target_model, + api_base=spec.api_base, + ) + return DynamicallyGeneratedExpert() diff --git a/epistemic_forge/pipeline/l1_optimizer.py b/epistemic_forge/pipeline/l1_optimizer.py index 35483d4..c1d89fd 100644 --- a/epistemic_forge/pipeline/l1_optimizer.py +++ b/epistemic_forge/pipeline/l1_optimizer.py @@ -1,11 +1,11 @@ """L1 — Instruction Optimizer (APE + OPRO fallback friendly).""" from dataclasses import dataclass -from typing import List + +from loguru import logger from epistemic_forge.llm import generate_structured from epistemic_forge.models import OptimizedInstruction, ProjectSpec -from loguru import logger @dataclass @@ -47,7 +47,7 @@ def _score_instruction( return round(max(0.01, min(score, 0.99)), 4) -def ape_generate(spec: ProjectSpec) -> List[InstructionCandidate]: +def ape_generate(spec: ProjectSpec) -> list[InstructionCandidate]: """Generate deterministic APE-style instruction seeds with heuristic scoring.""" seeds = _seed_instructions(spec) ranked = [ @@ -59,8 +59,8 @@ def ape_generate(spec: ProjectSpec) -> List[InstructionCandidate]: def opro_evolve( - candidates: List[InstructionCandidate], spec: ProjectSpec, steps: int = 2 -) -> List[InstructionCandidate]: + candidates: list[InstructionCandidate], spec: ProjectSpec, steps: int = 2 +) -> list[InstructionCandidate]: """Deterministically evolve instructions OPRO-style while preserving stability.""" pool = list(candidates) if candidates else ape_generate(spec) for step in range(max(1, steps)): diff --git a/epistemic_forge/pipeline/l2_conductor.py b/epistemic_forge/pipeline/l2_conductor.py index 8148b3b..de14a27 100644 --- a/epistemic_forge/pipeline/l2_conductor.py +++ b/epistemic_forge/pipeline/l2_conductor.py @@ -4,14 +4,16 @@ It uses the Strategy Pattern to dynamically load and run experts based on the domain. """ -from typing import Dict, Any +import asyncio +from typing import Any + from loguru import logger -from epistemic_forge.models import ProjectSpec from epistemic_forge.experts.base import EpistemicExpert from epistemic_forge.experts.claim_expert import ClaimLatticeExpert from epistemic_forge.experts.dialectic_expert import HegelianExpert from epistemic_forge.experts.kaggle_expert import RigorSentinelExpert +from epistemic_forge.models import ProjectSpec from epistemic_forge.pipeline.l1_5_adas import generate_dynamic_expert @@ -25,7 +27,7 @@ def __init__(self): def _route_experts(self, spec: ProjectSpec) -> list[EpistemicExpert]: """Determines which experts are required based on the domain.""" domain = spec.domain - active_experts = [ClaimLatticeExpert()] + active_experts: list[EpistemicExpert] = [ClaimLatticeExpert()] # 🧬 ADAS: Inject a dynamically generated expert specific to this domain! try: @@ -42,9 +44,9 @@ def _route_experts(self, spec: ProjectSpec) -> list[EpistemicExpert]: active_experts.append(RigorSentinelExpert()) return active_experts - def conduct(self, spec: ProjectSpec, context: Dict[str, Any]) -> Dict[str, Any]: + def conduct(self, spec: ProjectSpec, context: dict[str, Any]) -> dict[str, Any]: """ - Executes the active experts and aggregates their Pydantic outputs. + Executes the active experts sequentially and aggregates their outputs. Returns: A dictionary mapping expert names to their structured outputs. @@ -57,7 +59,7 @@ def conduct(self, spec: ProjectSpec, context: Dict[str, Any]) -> Dict[str, Any]: for expert in active_experts: logger.debug(f"Activating node: {expert.expert_name}") try: - # Executes the polymophic analyze() method + # Executes the polymorphic analyze() method structured_output = expert.analyze(spec, context) results[expert.expert_name] = structured_output.model_dump() except Exception as e: @@ -66,8 +68,31 @@ def conduct(self, spec: ProjectSpec, context: Dict[str, Any]) -> Dict[str, Any]: return results + async def conduct_async( + self, spec: ProjectSpec, context: dict[str, Any] + ) -> dict[str, Any]: + """Async variant that runs experts concurrently via asyncio.gather.""" + logger.info(f"L2 Conductor (async): Routing inquiry for domain [{spec.domain}]") + active_experts = self._route_experts(spec) + + async def _run(expert: EpistemicExpert) -> tuple[str, dict[str, Any]]: + try: + structured_output = await expert.analyze_async(spec, context) + return expert.expert_name, structured_output.model_dump() + except Exception as e: + logger.error(f"Cognitive fault in {expert.expert_name}: {e}") + return expert.expert_name, {"error": str(e)} + + paired = await asyncio.gather(*(_run(e) for e in active_experts)) + return dict(paired) + # Expose a functional interface for backward compatibility with the pipeline -def conduct(spec: ProjectSpec, claims_bundle: Dict[str, Any]) -> Dict[str, Any]: +def conduct(spec: ProjectSpec, claims_bundle: dict[str, Any]) -> dict[str, Any]: conductor = SemanticConductor() return conductor.conduct(spec, claims_bundle) + + +async def conduct_async(spec: ProjectSpec, claims_bundle: dict[str, Any]) -> dict[str, Any]: + conductor = SemanticConductor() + return await conductor.conduct_async(spec, claims_bundle) diff --git a/epistemic_forge/pipeline/l3_search.py b/epistemic_forge/pipeline/l3_search.py index 7082a5e..69e6e2d 100644 --- a/epistemic_forge/pipeline/l3_search.py +++ b/epistemic_forge/pipeline/l3_search.py @@ -1,20 +1,22 @@ """L3 — Tree Search (True LLM-Based Tree of Thoughts / LATS escalation).""" +import uuid +from typing import Any + +from loguru import logger + +from epistemic_forge.llm import generate_structured +from epistemic_forge.memory.economy import budget_manager from epistemic_forge.models import ( ProjectSpec, - SearchResult, SearchNode, - ThoughtProposalsOutput, + SearchResult, ThoughtEvaluation, + ThoughtProposalsOutput, ) -from epistemic_forge.llm import generate_structured -from loguru import logger -from epistemic_forge.memory.economy import budget_manager -from typing import Dict, Any, List -import uuid -def _generate_thoughts(spec: ProjectSpec, context: str, beam: int) -> List[str]: +def _generate_thoughts(spec: ProjectSpec, context: str, beam: int) -> list[str]: messages = [ { "role": "system", @@ -60,12 +62,12 @@ def _evaluate_thought(spec: ProjectSpec, thought: str) -> float: def explore( - spec: ProjectSpec, bundle: Dict[str, Any], beam: int = 3, steps: int = 2 + spec: ProjectSpec, bundle: dict[str, Any], beam: int = 3, steps: int = 2 ) -> SearchResult: logger.info( f"L3 Search: Initiating genuine LLM Tree Search (Beam={beam}, Steps={steps})..." ) - nodes: List[SearchNode] = [] + nodes: list[SearchNode] = [] current_context = f"Initial constraints for {spec.domain}." best_thought_overall = "" highest_score = -1.0 @@ -118,12 +120,12 @@ def explore( def tot_search( - spec: ProjectSpec, bundle: Dict[str, Any], beam: int = 3, steps: int = 2 + spec: ProjectSpec, bundle: dict[str, Any], beam: int = 3, steps: int = 2 ) -> SearchResult: return explore(spec, bundle, beam, steps) def lats_search( - spec: ProjectSpec, bundle: Dict[str, Any], rollouts: int = 3 + spec: ProjectSpec, bundle: dict[str, Any], rollouts: int = 3 ) -> SearchResult: return explore(spec, bundle, beam=rollouts, steps=2) diff --git a/epistemic_forge/pipeline/l4_refine.py b/epistemic_forge/pipeline/l4_refine.py index d8587f4..c70130d 100644 --- a/epistemic_forge/pipeline/l4_refine.py +++ b/epistemic_forge/pipeline/l4_refine.py @@ -5,10 +5,11 @@ it rewrites the draft iteratively until perfection or max retries are reached. """ -from epistemic_forge.models import ProjectSpec, RefinementFeedback, RefinedArtifact -from epistemic_forge.llm import generate_structured + from loguru import logger -from typing import Tuple + +from epistemic_forge.llm import generate_structured +from epistemic_forge.models import ProjectSpec, RefinedArtifact, RefinementFeedback def _generate_critique(spec: ProjectSpec, draft: str) -> RefinementFeedback: @@ -57,7 +58,7 @@ def _rewrite_draft( def refine_document( spec: ProjectSpec, draft: str, max_iterations: int = 2 -) -> Tuple[str, float]: +) -> tuple[str, float]: """Iterative Self-Refine Loop (Generate -> Critique -> Rewrite).""" logger.info( diff --git a/epistemic_forge/pipeline/l6_stages.py b/epistemic_forge/pipeline/l6_stages.py index 4ebffbd..86978f3 100644 --- a/epistemic_forge/pipeline/l6_stages.py +++ b/epistemic_forge/pipeline/l6_stages.py @@ -4,11 +4,13 @@ that evaluates the final crystallized artifact against scientific standards. """ -from epistemic_forge.models import ProjectSpec, FinalPeerReview, StageArtifact +from typing import Any + +from loguru import logger + from epistemic_forge.llm import generate_structured +from epistemic_forge.models import FinalPeerReview, ProjectSpec, StageArtifact from epistemic_forge.pipeline.l4_refine import refine_document -from loguru import logger -from typing import Dict, Any, List def _peer_review(spec: ProjectSpec, doc: str, prior_score: float) -> FinalPeerReview: @@ -38,9 +40,9 @@ def _peer_review(spec: ProjectSpec, doc: str, prior_score: float) -> FinalPeerRe def produce_artifacts( spec: ProjectSpec, final_draft: str, - claims_bundle: Dict[str, Any], + claims_bundle: dict[str, Any], prior_score: float, -) -> tuple[List[StageArtifact], Dict[str, Any], float]: +) -> tuple[list[StageArtifact], dict[str, Any], float]: """Wraps the pipeline in progressive evaluation stages and produces the final deliverables.""" logger.info("L6 Stages: Initializing final crystallization and review sequence.") diff --git a/epistemic_forge/pipeline/machine.py b/epistemic_forge/pipeline/machine.py new file mode 100644 index 0000000..a72a1d5 --- /dev/null +++ b/epistemic_forge/pipeline/machine.py @@ -0,0 +1,134 @@ +"""Explicit pipeline stage machine (L0–L6). + +The original pipeline was a linear chain of function calls with the layer +activation logic hidden inside ``if`` statements in ``arsenal_run.py``. This +module makes the *state machine* explicit and inspectable: + +* :class:`PipelineContext` is a single typed object that carries state between + stages (replacing ad-hoc ``Dict[str, Any]`` context passing). +* :data:`STAGES` is a registry describing every stage, its function, and the + predicate that decides whether it runs for a given context. This is what lets + the L0 router *actually* toggle layers instead of the activation map being + ignored. +* :func:`execute_pipeline` runs only the enabled stages, in order, recording + which ran. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +from epistemic_forge.errors import PipelineError +from epistemic_forge.models import ( + ProjectSpec, + RouteDecision, + SearchResult, + StageArtifact, +) +from epistemic_forge.pipeline.l1_optimizer import optimize_instruction +from epistemic_forge.pipeline.l2_conductor import conduct +from epistemic_forge.pipeline.l3_search import explore +from epistemic_forge.pipeline.l6_stages import produce_artifacts +from epistemic_forge.pipeline.router import route_project + + +@dataclass +class PipelineContext: + """Typed, versioned carrier of intermediate pipeline state.""" + + spec: ProjectSpec + route: RouteDecision | None = None + instruction: str = "" + conducted: dict[str, Any] = field(default_factory=dict) + search_result: SearchResult | None = None + artifacts: list[StageArtifact] = field(default_factory=list) + review: dict[str, Any] | None = None + final_score: float = 0.0 + stages_run: list[str] = field(default_factory=list) + + def assert_route(self) -> RouteDecision: + if self.route is None: + raise PipelineError("PipelineContext has no RouteDecision.", stage="L0") + return self.route + + +# Each entry: (stage_name, callable(ctx) -> ctx, enabled?(ctx) -> bool) +_STAGE_FN = Callable[["PipelineContext"], "PipelineContext"] + +STAGES: list[dict[str, Any]] = [ + { + "name": "L0_router", + "run": lambda ctx: _set(ctx, "route", route_project(ctx.spec)), + "enabled": lambda ctx: ctx.route is None, + }, + { + "name": "L1_optimizer", + "run": lambda ctx: _set(ctx, "instruction", optimize_instruction(ctx.spec)), + "enabled": lambda ctx: True, + }, + { + "name": "L2_conductor", + "run": lambda ctx: _set( + ctx, + "conducted", + conduct(ctx.spec, {"instruction": ctx.instruction, "skills": []}), + ), + "enabled": lambda ctx: True, + }, + { + "name": "L3_search", + "run": lambda ctx: _set( + ctx, "search_result", explore(ctx.spec, ctx.conducted, beam=3, steps=2) + ), + "enabled": lambda ctx: ctx.assert_route().activate.get("l3_search", True), + }, + { + "name": "L6_review", + "run": lambda ctx: _finalize(ctx), + "enabled": lambda ctx: True, + }, +] + + +def _set(ctx: PipelineContext, attr: str, value: Any) -> PipelineContext: + setattr(ctx, attr, value) + return ctx + + +def _finalize(ctx: PipelineContext) -> PipelineContext: + best = ctx.search_result.best_thought if ctx.search_result else str(ctx.conducted) + prior = ctx.search_result.score if ctx.search_result else 0.5 + artifacts, review, score = produce_artifacts( + ctx.spec, best, ctx.conducted, prior + ) + ctx.artifacts = artifacts + ctx.review = review + ctx.final_score = score + return ctx + + +def planned_stages(spec: ProjectSpec) -> list[str]: + """Return the ordered list of stage names that *would* run for a spec. + + Useful for introspection, tests, and honest documentation of what the + router actually activated. + """ + ctx = PipelineContext(spec=spec) + ctx.route = route_project(spec) + return [s["name"] for s in STAGES if s["enabled"](ctx)] + + +def execute_pipeline(ctx: PipelineContext) -> PipelineContext: + """Run every enabled stage in order, mutating and returning ``ctx``.""" + for stage in STAGES: + if stage["enabled"](ctx): + try: + ctx = stage["run"](ctx) + ctx.stages_run.append(stage["name"]) + except Exception as exc: # pragma: no cover - surfaced to caller + raise PipelineError( + f"Stage {stage['name']} failed: {exc}", stage=stage["name"] + ) from exc + return ctx diff --git a/epistemic_forge/pipeline/router.py b/epistemic_forge/pipeline/router.py index a979c7d..d9cb6ee 100644 --- a/epistemic_forge/pipeline/router.py +++ b/epistemic_forge/pipeline/router.py @@ -1,21 +1,23 @@ """L0 — Semantic Technique Router (SOTA LLM-Based Routing). -Replaces rigid heuristics with a Semantic Router that analyzes the -epistemic complexity of the query to dynamically toggle architectural layers +Replaces rigid heuristics with a Semantic Router that analyzes the +epistemic complexity of the query to dynamically toggle architectural layers (L1-L6) to save tokens (Cognitive Economy) while maintaining rigor. """ -from epistemic_forge.models import ProjectSpec, RouteDecision -from epistemic_forge.llm import generate_structured from loguru import logger +from epistemic_forge.llm import generate_structured +from epistemic_forge.models import ProjectSpec, RouteDecision + + def route_project(spec: ProjectSpec) -> RouteDecision: """Dynamically routes the project through the optimal cognitive layers.""" - + logger.info("L0 Router: Analyzing epistemic complexity to dynamically route execution...") - + messages = [ { - "role": "system", + "role": "system", "content": ( "You are an Elite L0 Architectural Router. Analyze the user's inquiry and determine exactly which cognitive layers are required to solve it. " "If it's a simple query, turn off heavy layers (like L3 Tree Search) to save compute. " @@ -24,7 +26,7 @@ def route_project(spec: ProjectSpec) -> RouteDecision: }, {"role": "user", "content": f"Inquiry: {spec.question}\nDomain: {spec.domain}\n\nDetermine the optimal routing architecture."} ] - + try: decision: RouteDecision = generate_structured( messages=messages, diff --git a/epistemic_forge/ui/app.py b/epistemic_forge/ui/app.py index fef9e56..1211edd 100644 --- a/epistemic_forge/ui/app.py +++ b/epistemic_forge/ui/app.py @@ -5,6 +5,7 @@ """ import streamlit as st + from epistemic_forge.pipeline.arsenal_run import run_pipeline st.set_page_config(page_title="Epistemic Forge", page_icon="🧠", layout="wide") @@ -46,7 +47,8 @@ title=title, question=question, domain=domain, - # Further deep integration needed to pass model_choice strictly to the run. + target_model=model_choice, + api_base=api_base or None, ) st.success("Synthesis Complete.") @@ -55,11 +57,7 @@ if hasattr(result, "claims"): st.markdown("### 🌳 Epistemic Claim Lattice") for claim in result.claims: - c_dict = ( - claim.model_dump() - if hasattr(claim, "model_dump") - else claim - ) + c_dict = claim.model_dump() with st.expander(f"Claim: {c_dict.get('text', '')}"): st.write("**Supports:**") for s in c_dict.get("support", []): diff --git a/examples/run_demo.py b/examples/run_demo.py index a569e78..ab9a9bf 100644 --- a/examples/run_demo.py +++ b/examples/run_demo.py @@ -7,27 +7,27 @@ from epistemic_forge.pipeline.arsenal_run import run_pipeline DEMOS = [ - dict( - title="Predictive minds and blame", - question="If the brain is a prediction machine, what happens to moral responsibility?", - domain="philosophy", - keywords=["predictive processing", "responsibility"], - out="runs/demo_philosophy", - ), - dict( - title="Imbalanced tabular baseline", - question="What is an honest baseline plan for a noisy imbalanced Kaggle table?", - domain="kaggle", - keywords=["imbalance", "baseline", "cv", "leakage"], - out="runs/demo_kaggle", - ), - dict( - title="Climate-tech research sprint", - question="How do I scope a 2-week research sprint for a climate-tech founder?", - domain="freelance", - keywords=["sprint", "scope", "founder"], - out="runs/demo_freelance", - ), + { + "title": "Predictive minds and blame", + "question": "If the brain is a prediction machine, what happens to moral responsibility?", + "domain": "philosophy", + "keywords": ["predictive processing", "responsibility"], + "out": "runs/demo_philosophy", + }, + { + "title": "Imbalanced tabular baseline", + "question": "What is an honest baseline plan for a noisy imbalanced Kaggle table?", + "domain": "kaggle", + "keywords": ["imbalance", "baseline", "cv", "leakage"], + "out": "runs/demo_kaggle", + }, + { + "title": "Climate-tech research sprint", + "question": "How do I scope a 2-week research sprint for a climate-tech founder?", + "domain": "freelance", + "keywords": ["sprint", "scope", "founder"], + "out": "runs/demo_freelance", + }, ] diff --git a/patch_run.py b/patch_run.py deleted file mode 100644 index 817dd39..0000000 --- a/patch_run.py +++ /dev/null @@ -1,21 +0,0 @@ -with open("epistemic_forge/pipeline/arsenal_run.py", "r") as f: - content = f.read() - -if "from loguru import logger" not in content: - content = "from loguru import logger\n" + content - -old_run = " return ArsenalRun.create().run(spec)" -new_run = """ try: - logger.info(f"Starting Epistemic Forge Pipeline for: '{title}'") - result = ArsenalRun.create().run(spec) - logger.success("Pipeline execution completed successfully.") - return result - except Exception as e: - logger.exception(f"Critical Pipeline Failure: {str(e)}") - raise SystemExit(1)""" - -if old_run in content: - content = content.replace(old_run, new_run) - -with open("epistemic_forge/pipeline/arsenal_run.py", "w") as f: - f.write(content) diff --git a/pyproject.toml b/pyproject.toml index f743914..d0d717d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,10 +31,20 @@ classifiers = [ "Topic :: Scientific/Engineering :: Artificial Intelligence", "Topic :: Text Processing :: Linguistic", ] -dependencies = ["litellm>=1.40.0", "duckduckgo-search>=5.0.0", "chromadb>=0.4.0", "nbformat>=5.0.0", "fastapi>=0.100.0", "uvicorn>=0.20.0", "sse-starlette>=2.0.0", "pydantic>=2.0.0", "loguru>=0.7.0", "tenacity>=8.0.0", "instructor>=1.3.0", "rich>=13.0.0", "streamlit>=1.35.0"] +dependencies = ["litellm>=1.40.0", "duckduckgo-search>=5.0.0", "chromadb>=0.4.0", "nbformat>=5.0.0", "fastapi>=0.100.0", "uvicorn>=0.20.0", "sse-starlette>=2.0.0", "pydantic>=2.0.0", "loguru>=0.7.0", "tenacity>=8.0.0", "instructor>=1.3.0", "rich>=13.0.0"] [project.optional-dependencies] -dev = ["pytest>=7.0", "pytest-mock"] +dev = [ + "pytest>=7.0,<9", + "pytest-mock", + "pytest-asyncio>=0.23,<0.25", + "pytest-cov>=4.0,<5", + "ruff>=0.4,<0.16", + "mypy>=1.8,<2", + "bandit>=1.7,<2", + "types-requests", +] +ui = ["streamlit>=1.35.0"] [project.scripts] epistemic-forge = "epistemic_forge.cli:main" @@ -54,3 +64,29 @@ epistemic_forge = ["data/samples/*.json", "data/samples/*.md"] [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] +asyncio_mode = "auto" +addopts = "-q" + +[tool.coverage.run] +source = ["epistemic_forge"] +omit = ["epistemic_forge/ui/*", "epistemic_forge/benchmark/*"] + +[tool.coverage.report] +show_missing = true + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "B", "C4", "UP", "SIM"] +ignore = ["E501", "B008", "C401", "SIM105"] + +[tool.ruff.lint.per-file-ignores] +"__init__.py" = ["F401"] + +[tool.mypy] +python_version = "3.10" +ignore_missing_imports = true +warn_unused_ignores = false +disallow_untyped_defs = false diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..832e4e6 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,42 @@ +"""Shared pytest fixtures. + +Forces the universal LLM router into its deterministic offline fallback (no +network / no API keys) and prevents ChromaDB from writing to disk, so the test +suite is fast, hermetic, and CI-friendly. +""" + +from __future__ import annotations + +import pytest + +import epistemic_forge.llm as llm_mod +from epistemic_forge.memory.economy import budget_manager +from epistemic_forge.pipeline import arsenal_run as arsenal_run_mod + + +class _FakeSkillLibrary: + """In-memory stand-in for the ChromaDB-backed SkillLibrary.""" + + def __init__(self, *args, **kwargs): + self._skills = [] + + def retrieve_relevant_skills(self, *args, **kwargs): + return [] + + def add_skill(self, skill): + self._skills.append(skill) + + def get_all_skills(self): + return list(self._skills) + + +@pytest.fixture(autouse=True) +def _offline_and_memory(monkeypatch): + # The TokenBudgetManager is a process-wide singleton; reset it so a test + # that intentionally exhausts the budget can't poison later tests. + budget_manager.reset() + # Force the router into deterministic offline mode for every test. + monkeypatch.setattr(llm_mod, "_missing_credentials", lambda model, api_key: True) + # Avoid touching ChromaDB on disk inside ArsenalRun.create(). + monkeypatch.setattr(arsenal_run_mod, "SkillLibrary", _FakeSkillLibrary) + yield diff --git a/tests/test_extra.py b/tests/test_extra.py new file mode 100644 index 0000000..bc92041 --- /dev/null +++ b/tests/test_extra.py @@ -0,0 +1,146 @@ +"""Additional coverage: memory helpers, I/O export, standalone experts, CLI, errors.""" + + +import pytest + +from epistemic_forge.errors import PipelineError +from epistemic_forge.experts import freelance_expert, semitic_expert, writing_expert +from epistemic_forge.memory.reflexion_store import ReflexionStore +from epistemic_forge.models import ( + Claim, + Confidence, + ForgeResult, + ProjectSpec, + RouteDecision, + StageArtifact, +) +from epistemic_forge.pipeline import arsenal_run + +# ----------------------------- reflexion store ----------------------------- + + +def test_reflexion_windowing(): + store = ReflexionStore(window=3) + for i in range(5): + store.add(store.reflect_on_failure(trial=i, score=0.1, notes=f"fail {i}")) + assert len(store.all()) == 3 + assert store.all()[-1].trial == 4 + + +def test_reflexion_prompt_block_empty(): + assert "No prior" in ReflexionStore().as_prompt_block() + + +def test_reflexion_reflect_on_failure(): + store = ReflexionStore() + r = store.reflect_on_failure(trial=1, score=0.2, notes="bad") + assert r.lesson + assert r in store.all() + + +# ------------------------------- I/O export -------------------------------- + + +def _sample_result(): + spec = ProjectSpec(title="Sample", question="Q?", domain="research") + claims = [ + Claim( + id="C1", + text="A baseline-first plan is reliable.", + epistemic_warrant="Baselines expose errors early.", + potential_falsifier="If baseline fails while alt succeeds.", + confidence=Confidence.LIKELY, + ) + ] + artifacts = [ + StageArtifact(name="Final Synthesis Memo", content="memo body", kind="markdown"), + StageArtifact( + name="Baseline Notebook", + content="```python\nprint('hi')\n```", + kind="python", + path_hint="baseline", + ), + ] + return ForgeResult( + spec=spec, + route=RouteDecision(families=["mock"], activate={}, rationale="r"), + instruction="do", + claims=claims, + search_trace=[], + reflections=[], + skills_used=[], + artifacts=artifacts, + peer_review={ + "verdict": "accept", + "final_comments": "looks good", + "overall_score": 0.8, + }, + final_score=0.8, + ) + + +def test_export_writes_files(tmp_path): + from epistemic_forge.io.export import export_result + + export_result(_sample_result(), str(tmp_path)) + assert (tmp_path / "executive_summary.md").exists() + assert (tmp_path / "claim_lattice_graph.json").exists() + assert (tmp_path / "baseline.ipynb").exists() + + +def test_export_mermaid_contains_claim(tmp_path): + from epistemic_forge.io.export import _generate_mermaid_graph + + graph = _generate_mermaid_graph(_sample_result().claims) + assert "C1" in graph + assert "graph TD" in graph + + +# --------------------------- standalone experts ---------------------------- + + +def test_freelance_pack(): + spec = ProjectSpec(title="Pitch", question="Help me price", domain="freelance") + pack = freelance_expert.build_client_pack(spec, {}) + assert pack["client_brief"]["title"] == "Pitch" + assert pack["acceptance_criteria"] + + +def test_writing_expert_offline(): + spec = ProjectSpec(title="t", question="q") + out = writing_expert.outline_and_draft(spec, {"claims": []}, "instr") + assert "draft_markdown" in out + assert "Density Score" in out["draft_markdown"] + + +def test_semitic_expert_offline(): + spec = ProjectSpec(title="t", question="q") + out = semitic_expert.run_semitic_dialectic(spec, {}) + assert out["thesis"] == "q" + assert "synthesis_arabic" in out + + +# ------------------------------- CLI / errors ------------------------------ + + +def test_cli_main_runs(monkeypatch, capsys): + from epistemic_forge.cli import main + + class FakeResult: + claims = [] + + monkeypatch.setattr("epistemic_forge.cli.run_pipeline", lambda **k: FakeResult()) + monkeypatch.setattr("epistemic_forge.io.export.export_result", lambda *a, **k: None) + monkeypatch.setattr("sys.argv", ["ef", "--title", "T", "--question", "Q?"]) + main() # must not raise (and must not call sys.exit) + captured = capsys.readouterr() + assert "Pipeline Execution Successful" in captured.out + + +def test_run_pipeline_raises_typed_error_not_systemexit(monkeypatch): + def boom(ctx): + raise RuntimeError("kaboom") + + monkeypatch.setattr(arsenal_run, "execute_pipeline", boom) + with pytest.raises(PipelineError): + arsenal_run.run_pipeline(title="t", question="q") diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..cc8fe30 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,74 @@ +"""Integration + architecture (machine) + CLI + async tests.""" + +import asyncio + +from epistemic_forge.cli import build_parser +from epistemic_forge.models import ProjectSpec, RouteDecision +from epistemic_forge.pipeline import arsenal_run +from epistemic_forge.pipeline.l2_conductor import conduct_async +from epistemic_forge.pipeline.machine import PipelineContext, execute_pipeline, planned_stages + + +def test_full_pipeline_offline_runs(): + result = arsenal_run.run_pipeline( + title="Integration", question="Does this work end to end?", domain="research" + ) + assert result.spec.title == "Integration" + assert isinstance(result.final_score, float) + assert 0.0 <= result.final_score <= 1.0 + assert result.peer_review + assert len(result.artifacts) >= 1 + # Claims come back as dicts from the offline fallback; ensure shape is sane. + assert isinstance(result.claims, list) + + +def test_planned_stages_includes_heavy_layers(): + stages = planned_stages(ProjectSpec(title="t", question="q", domain="research")) + assert "L3_search" in stages + assert "L6_review" in stages + + +def test_execute_pipeline_respects_l3_toggle(): + ctx = PipelineContext( + spec=ProjectSpec(title="t", question="q", domain="research"), + route=RouteDecision( + families=["mock"], activate={"l3_search": False}, rationale="disable search" + ), + ) + out = execute_pipeline(ctx) + assert "L3_search" not in out.stages_run + assert out.search_result is None + assert "L6_review" in out.stages_run + + +def test_async_pipeline_runs(): + result = arsenal_run.run_pipeline( + title="Async", question="Async end to end?", domain="philosophy", async_run=True + ) + assert isinstance(result.final_score, float) + assert 0.0 <= result.final_score <= 1.0 + + +def test_conduct_async_runs_experts_concurrently(): + spec = ProjectSpec(title="t", question="q", domain="research") + results = asyncio.run( + conduct_async(spec, {"instruction": "x", "skills": []}) + ) + assert "Grounded_Claim_Lattice_Generator" in results + assert "Hegelian_Dialectic_Engine" in results + + +def test_cli_parser_requires_args(): + parser = build_parser() + args = parser.parse_args(["--title", "T", "--question", "Q?"]) + assert args.title == "T" + assert args.question == "Q?" + assert args.domain == "hybrid" + + +def test_cli_parser_custom_model(): + parser = build_parser() + args = parser.parse_args( + ["--title", "T", "--question", "Q?", "--model", "anthropic/claude-3-opus"] + ) + assert args.model == "anthropic/claude-3-opus" diff --git a/tests/test_llm.py b/tests/test_llm.py new file mode 100644 index 0000000..cc5bf76 --- /dev/null +++ b/tests/test_llm.py @@ -0,0 +1,110 @@ +"""Tests for the universal LLM router (offline fallback, validation, async).""" + +import pytest + +from epistemic_forge.errors import InvalidInputError +from epistemic_forge.llm import ( + _offline_fallback, + agenerate_structured, + generate_structured, + validate_messages, +) +from epistemic_forge.models import ( + ClaimLatticeOutput, + DynamicExpertSchema, + FinalPeerReview, + HegelianDialecticOutput, + OptimizedInstruction, + RefinedArtifact, + RefinementFeedback, + RigorSentinelOutput, + RouteDecision, + ThoughtEvaluation, + ThoughtProposalsOutput, +) + +OFFLINE_MODELS = [ + OptimizedInstruction, + ThoughtProposalsOutput, + ThoughtEvaluation, + RefinementFeedback, + RefinedArtifact, + FinalPeerReview, + DynamicExpertSchema, + ClaimLatticeOutput, + HegelianDialecticOutput, + RigorSentinelOutput, + RouteDecision, +] + + +@pytest.mark.parametrize("model", OFFLINE_MODELS) +def test_offline_fallback_builds_every_model(model): + out = _offline_fallback(model, [{"role": "user", "content": "test prompt"}]) + assert isinstance(out, model) + + +def test_offline_fallback_fills_unknown_required_fields(): + from pydantic import BaseModel + + class WeirdModel(BaseModel): + a: str + b: float + c: int + d: bool + e: list + + out = _offline_fallback(WeirdModel, [{"role": "user", "content": "x"}]) + assert out.a + assert out.b == 0.5 + assert out.c == 0 + assert out.d is False + assert out.e == [] + + +def test_validate_messages_rejects_empty(): + with pytest.raises(InvalidInputError): + validate_messages([]) + + +def test_validate_messages_rejects_non_mapping(): + with pytest.raises(InvalidInputError): + validate_messages(["not a dict"]) + + +def test_validate_messages_rejects_oversized(monkeypatch): + monkeypatch.setattr( + "epistemic_forge.llm.MAX_PROMPT_CHARS", 10 + ) + with pytest.raises(InvalidInputError): + validate_messages([{"role": "user", "content": "x" * 50}]) + + +def test_validate_messages_accepts_valid(): + validate_messages( + [ + {"role": "system", "content": "be good"}, + {"role": "user", "content": "hello"}, + ] + ) + + +def test_generate_structured_offline_returns_fallback(): + out = generate_structured( + messages=[{"role": "user", "content": "hi"}], + response_model=OptimizedInstruction, + model="gpt-4o-mini", + ) + assert isinstance(out, OptimizedInstruction) + assert "Toulmin" in out.meta_prompt or "claim" in out.meta_prompt.lower() + + +@pytest.mark.asyncio +async def test_agenerate_structured_offline_returns_fallback(): + out = await agenerate_structured( + messages=[{"role": "user", "content": "hi"}], + response_model=ClaimLatticeOutput, + model="gpt-4o-mini", + ) + assert isinstance(out, ClaimLatticeOutput) + assert out.claims diff --git a/tests/test_memory.py b/tests/test_memory.py new file mode 100644 index 0000000..3b4cef1 --- /dev/null +++ b/tests/test_memory.py @@ -0,0 +1,107 @@ +"""Tests for the memory layer: TokenBudgetManager and SkillLibrary.""" + +from epistemic_forge.memory.economy import TokenBudgetManager, budget_manager +from epistemic_forge.memory.skill_library import SkillLibrary +from epistemic_forge.models import Skill + + +class _FakeCollection: + def __init__(self): + self._store = {} + + def add(self, documents, metadatas, ids): + for _id, doc, meta in zip(ids, documents, metadatas, strict=False): + self._store[_id] = (doc, meta) + + def count(self): + return len(self._store) + + def query(self, query_texts, n_results): + ids = list(self._store.keys())[:n_results] + return { + "metadatas": [[self._store[i][1] for i in ids]], + "documents": [[self._store[i][0] for i in ids]], + "ids": [ids], + } + + def get(self): + ids = list(self._store.keys()) + return { + "metadatas": [self._store[i][1] for i in ids], + "documents": [self._store[i][0] for i in ids], + "ids": ids, + } + + +class _FakeClient: + def get_or_create_collection(self, name): + return _FakeCollection() + + def create_collection(self, name): + return _FakeCollection() + + +def test_budget_reset_and_set(): + mgr = TokenBudgetManager() + mgr.reset() + mgr.set_budget(100) + assert mgr.budget_limit == 100 + assert not mgr.is_budget_exceeded() + + +def test_budget_exceeded(mocker): + mgr = TokenBudgetManager() + mgr.reset() + mgr.set_budget(100) + + class MockResponse: + class Usage: + total_tokens = 150 + + usage = Usage() + + mgr.add_usage(MockResponse(), "mock-model") + assert mgr.is_budget_exceeded() is True + + +def test_budget_add_usage_no_usage(mocker): + mgr = TokenBudgetManager() + mgr.reset() + mgr.set_budget(1000) + # Object without .usage should not raise + mgr.add_usage(object(), "mock-model") + assert mgr.total_tokens == 0 + + +def test_singleton_identity(): + assert TokenBudgetManager() is TokenBudgetManager() + assert budget_manager is TokenBudgetManager() + + +def test_skill_library_add_and_retrieve(): + lib = SkillLibrary(client=_FakeClient()) + lib.add_skill( + Skill(name="baseline", description="use a baseline", code="x", tags=["ml"]) + ) + skills = lib.retrieve_relevant_skills("training", n_results=1) + assert len(skills) == 1 + assert skills[0].name == "baseline" + + +def test_skill_library_empty_retrieve(): + lib = SkillLibrary(client=_FakeClient()) + assert lib.retrieve_relevant_skills("anything") == [] + + +def test_skill_library_get_all(): + lib = SkillLibrary(client=_FakeClient()) + lib.add_skill(Skill(name="a", description="da", code="ca")) + lib.add_skill(Skill(name="b", description="db", code="cb")) + all_skills = lib.get_all_skills() + assert {s.name for s in all_skills} == {"a", "b"} + + +def test_skill_library_accepts_ephemeral(monkeypatch): + # persist_dir=None must not touch disk and must work. + lib = SkillLibrary(persist_dir=None) + assert lib.collection is not None diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..3aad529 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,94 @@ +"""Tests for the core Pydantic data models and their constraints.""" + +import pytest +from pydantic import ValidationError + +from epistemic_forge.models import ( + Claim, + ClaimLatticeOutput, + Confidence, + Domain, + FinalPeerReview, + ForgeResult, + PeerReviewScores, + ProjectSpec, + RouteDecision, + Skill, +) + + +def test_domain_coercion_and_default(): + spec = ProjectSpec(title="t", question="q", domain="research") + assert spec.domain is Domain.RESEARCH + default = ProjectSpec(title="t", question="q") + assert default.domain is Domain.HYBRID + + +def test_domain_invalid_string_rejected(): + with pytest.raises(ValidationError): + ProjectSpec(title="t", question="q", domain="not_a_domain") + + +def test_claim_requires_warrant_and_falsifier(): + with pytest.raises(ValidationError): + Claim(id="C1", text="a claim") + + +def test_claim_valid(): + c = Claim( + id="C1", + text="x", + epistemic_warrant="because y", + potential_falsifier="if z", + confidence=Confidence.LIKELY, + ) + assert c.confidence is Confidence.LIKELY + + +def test_route_decision_constraints(): + rd = RouteDecision( + families=["a"], activate={"l3_search": True}, rationale="r" + ) + assert rd.l1_mode == "ape" + assert rd.l3_mode == "tot" + + +def test_peer_review_scores_bounded(): + with pytest.raises(ValidationError): + PeerReviewScores(clarity=2.0, structure=0.0, soundness=0.0, actionability=0.0, humility=0.0) + + +def test_final_peer_review_overall_bounded(): + with pytest.raises(ValidationError): + FinalPeerReview( + scores=PeerReviewScores( + clarity=0.1, structure=0.1, soundness=0.1, actionability=0.1, humility=0.1 + ), + overall_score=1.5, + verdict="accept", + final_comments="ok", + ) + + +def test_forge_result_roundtrip(): + spec = ProjectSpec(title="t", question="q") + res = ForgeResult( + spec=spec, + route=RouteDecision(families=["a"], activate={}, rationale="r"), + instruction="do it", + claims=[], + search_trace=[], + reflections=[], + skills_used=[], + artifacts=[], + peer_review={}, + final_score=0.5, + ) + dumped = res.model_dump() + assert dumped["final_score"] == 0.5 + assert isinstance(ClaimLatticeOutput(claims=[], lattice_summary="s"), ClaimLatticeOutput) + + +def test_skill_model(): + s = Skill(name="x", description="d", code="c", tags=["t1"]) + assert s.name == "x" diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 20ee24d..2fa3714 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,12 +1,16 @@ -"""Test suite for Epistemic Forge Core Pipeline. +"""Classic pipeline smoke tests (mirrors the original suite, now corrected). -Uses Pytest and Mocks to ensure CI/CD passes without requiring +Uses Pytest and Mocks to ensure CI/CD passes without requiring live API keys or consuming credits. """ + import pytest -from unittest.mock import MagicMock -from epistemic_forge.models import ProjectSpec, ClaimLatticeOutput, Claim, Confidence, RouteDecision -from epistemic_forge.pipeline.arsenal_run import ArsenalRun + +from epistemic_forge.models import ( + ProjectSpec, +) +from epistemic_forge.pipeline.l2_conductor import SemanticConductor + @pytest.fixture def dummy_spec(): @@ -14,42 +18,46 @@ def dummy_spec(): title="Test Project", question="Is this a test?", domain="research", - target_model="mock-model" + target_model="mock-model", ) + def test_l0_router_fallback(dummy_spec, mocker): """Ensure L0 Router falls back safely if LLM fails.""" - # Mock the LLM to raise an exception - mocker.patch("epistemic_forge.pipeline.router.generate_structured", side_effect=Exception("API Down")) + mocker.patch( + "epistemic_forge.pipeline.router.generate_structured", + side_effect=Exception("API Down"), + ) from epistemic_forge.pipeline.router import route_project - + decision = route_project(dummy_spec) assert decision.l3_mode == "tot" assert decision.activate["l4_refine"] is True + def test_l2_conductor_routing(dummy_spec, mocker): - """Ensure L2 Conductor routes to the correct experts.""" - from epistemic_forge.pipeline.l2_conductor import SemanticConductor - + """Ensure L2 Conductor routes to the correct experts for the research domain.""" conductor = SemanticConductor() - active = conductor._route_experts(dummy_spec.domain.value) - - # Research domain should activate Hegelian and Rigor Sentinel + # NOTE: the contract requires a ProjectSpec, not a bare domain string. + active = conductor._route_experts(dummy_spec) + names = [e.expert_name for e in active] assert "Hegelian_Dialectic_Engine" in names assert "Rigor_And_Leakage_Sentinel" in names + def test_economy_manager(): """Ensure the Token Budget Manager correctly halts operations.""" from epistemic_forge.memory.economy import budget_manager - + budget_manager.reset() budget_manager.set_budget(100) - + class MockResponse: class Usage: total_tokens = 150 + usage = Usage() - + budget_manager.add_usage(MockResponse(), "mock-model") assert budget_manager.is_budget_exceeded() is True diff --git a/tests/test_units.py b/tests/test_units.py new file mode 100644 index 0000000..8e9eff8 --- /dev/null +++ b/tests/test_units.py @@ -0,0 +1,199 @@ +"""Unit tests for individual pipeline stages (L0–L6).""" + + +import pytest + +from epistemic_forge.errors import InvalidInputError +from epistemic_forge.models import ( + DynamicExpertSchema, + FinalPeerReview, + PeerReviewScores, + ProjectSpec, + RefinedArtifact, + RefinementFeedback, + ThoughtEvaluation, + ThoughtProposalsOutput, +) +from epistemic_forge.pipeline.l1_5_adas import _safe_identifier, generate_dynamic_expert +from epistemic_forge.pipeline.l1_optimizer import optimize_instruction +from epistemic_forge.pipeline.l2_conductor import SemanticConductor +from epistemic_forge.pipeline.l3_search import explore +from epistemic_forge.pipeline.l4_refine import refine_document +from epistemic_forge.pipeline.l6_stages import produce_artifacts +from epistemic_forge.pipeline.router import route_project + + +@pytest.fixture +def spec(): + return ProjectSpec(title="T", question="Is this testable?", domain="research") + + +def test_l0_router_fallback_on_failure(mocker): + mocker.patch( + "epistemic_forge.pipeline.router.generate_structured", + side_effect=Exception("API Down"), + ) + decision = route_project(ProjectSpec(title="t", question="q", domain="research")) + assert decision.l3_mode == "tot" + assert decision.activate["l4_refine"] is True + + +def test_l0_router_offline_returns_decision(): + d = route_project(ProjectSpec(title="t", question="q", domain="philosophy")) + assert isinstance(d, type(d)) + assert "l3_search" in d.activate + + +def test_l1_optimizer_offline_returns_string(spec): + inst = optimize_instruction(spec) + assert isinstance(inst, str) and len(inst) > 0 + + +def test_l1_optimizer_fallback_on_error(mocker, spec): + mocker.patch( + "epistemic_forge.pipeline.l1_optimizer.generate_structured", + side_effect=RuntimeError("boom"), + ) + inst = optimize_instruction(spec) + assert isinstance(inst, str) and len(inst) > 0 + + +def test_l2_conductor_routing_research(spec): + conductor = SemanticConductor() + active = conductor._route_experts(spec) + names = [e.expert_name for e in active] + assert "Hegelian_Dialectic_Engine" in names + assert "Rigor_And_Leakage_Sentinel" in names + + +def test_l2_conductor_routing_kaggle(): + spec = ProjectSpec(title="t", question="q", domain="kaggle") + names = [e.expert_name for e in SemanticConductor()._route_experts(spec)] + assert "Rigor_And_Leakage_Sentinel" in names + assert "Hegelian_Dialectic_Engine" not in names + + +def test_l2_conductor_routing_writing_has_no_hegelian_or_rigor(): + spec = ProjectSpec(title="t", question="q", domain="writing") + names = [e.expert_name for e in SemanticConductor()._route_experts(spec)] + assert "Hegelian_Dialectic_Engine" not in names + assert "Rigor_And_Leakage_Sentinel" not in names + + +def test_l2_conductor_type_contract(): + # The router test originally passed a string; the contract requires a ProjectSpec. + with pytest.raises(AttributeError): + SemanticConductor()._route_experts("research") # type: ignore[arg-type] + + +def test_safe_identifier(): + assert _safe_identifier("My Expert!") == "myexpert" + assert _safe_identifier("9bad") == "_9bad" + assert _safe_identifier("") == "" + + +def test_adas_creates_working_expert(mocker): + def fake_gen(messages, response_model, **kwargs): + if response_model.__name__ == "DynamicExpertSchema": + return DynamicExpertSchema( + expert_class_name="RiskExpert", + expert_description="risk analysis", + fields_to_extract=[{"risk": "main risk"}, {"check": "validation"}], + system_prompt="analyze risks", + ) + return response_model(risk="r", check="c") + + mocker.patch( + "epistemic_forge.pipeline.l1_5_adas.generate_structured", side_effect=fake_gen + ) + expert = generate_dynamic_expert( + ProjectSpec(title="t", question="q", domain="research") + ) + assert expert.expert_name == "riskexpert" + out = expert.analyze( + ProjectSpec(title="t", question="q", domain="research"), {"instruction": "x"} + ) + assert out.risk == "r" + + +def test_adas_rejects_empty_blueprint(mocker): + mocker.patch( + "epistemic_forge.pipeline.l1_5_adas.generate_structured", + return_value=DynamicExpertSchema( + expert_class_name="X", expert_description="d", fields_to_extract=[], system_prompt="p" + ), + ) + with pytest.raises(InvalidInputError): + generate_dynamic_expert(ProjectSpec(title="t", question="q")) + + +def test_l3_search_evaluates_and_rolls_back(mocker): + scores = iter([0.2, 0.9, 0.5]) + calls = {"n": 0} + + def fake_gen(messages, response_model, **kwargs): + calls["n"] += 1 + if response_model.__name__ == "ThoughtProposalsOutput": + return ThoughtProposalsOutput( + proposals=[ + {"thought_text": "weak thought"}, + {"thought_text": "strong thought"}, + ] + ) + return ThoughtEvaluation(epistemic_score=next(scores), critique="c") + + mocker.patch("epistemic_forge.pipeline.l3_search.generate_structured", side_effect=fake_gen) + result = explore(ProjectSpec(title="t", question="q"), {"x": 1}, beam=2, steps=1) + assert result.mode_used + assert len(result.nodes) == 2 # both thoughts become nodes + # best thought is the strong one (0.9) + assert "strong" in result.best_thought + assert result.score == 0.9 + + +def test_l4_refine_iterates_until_pass(mocker): + state = {"i": 0} + + def fake_gen(messages, response_model, **kwargs): + state["i"] += 1 + if response_model.__name__ == "RefinementFeedback": + if state["i"] == 1: + return RefinementFeedback( + clarity_score=0.4, + epistemic_humility_score=0.4, + critical_flaws=["vague"], + passes_threshold=False, + ) + return RefinementFeedback( + clarity_score=0.9, + epistemic_humility_score=0.9, + critical_flaws=[], + passes_threshold=True, + ) + return RefinedArtifact(improved_text="FINAL", changes_made=["fixed vagueness"]) + + mocker.patch("epistemic_forge.pipeline.l4_refine.generate_structured", side_effect=fake_gen) + text, score = refine_document(ProjectSpec(title="t", question="q"), "draft", max_iterations=3) + assert text == "FINAL" + assert score > 0.8 + + +def test_l6_stages_produces_artifacts(mocker): + mocker.patch( + "epistemic_forge.pipeline.l6_stages.generate_structured", + return_value=FinalPeerReview( + scores=PeerReviewScores( + clarity=0.8, structure=0.8, soundness=0.8, actionability=0.8, humility=0.8 + ), + overall_score=0.8, + revision_needed=[], + verdict="accept", + final_comments="good", + ), + ) + artifacts, review, score = produce_artifacts( + ProjectSpec(title="t", question="q"), "final draft", {}, 0.5 + ) + assert len(artifacts) >= 1 + assert review["overall_score"] == 0.8 + assert score == 0.8