Spaces:
Sleeping
Sleeping
| """ | |
| VentureForge CLI | |
| ================ | |
| Run the full multi-agent pipeline from the command line. | |
| Usage: | |
| python -m src.main --domain "developer tools" | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import logging | |
| from typing import Any | |
| from src.config import settings | |
| from src.graph import GRAPH | |
| from src.state.schema import VentureForgeState | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="[%(asctime)s] %(levelname)s [%(name)s] %(message)s", | |
| ) | |
| def make_initial_state( | |
| domain: str, | |
| max_pain_points: int | None = None, | |
| ideas_per_run: int | None = None, | |
| top_n_pitches: int | None = None, | |
| max_revisions: int | None = None, | |
| ) -> VentureForgeState: | |
| """Construct the initial VentureForgeState for a new run. | |
| Shared between the CLI entrypoint and the Gradio UI controller so | |
| that both use the same defaults from ``src.config.settings``. | |
| """ | |
| return VentureForgeState( | |
| domain=domain, | |
| max_pain_points=max_pain_points or settings.max_pain_points, | |
| ideas_per_run=ideas_per_run or settings.ideas_per_run, | |
| top_n_pitches=top_n_pitches or settings.top_n_pitches, | |
| max_revisions=max_revisions or settings.max_revisions, | |
| ) | |
| def run_pipeline( | |
| domain: str | None, | |
| max_pain_points: int | None = None, | |
| *, | |
| recursion_limit: int = 80, | |
| resume_run_id: str | None = None, | |
| force_refresh: bool = False, | |
| ) -> VentureForgeState | dict[str, Any]: | |
| """Execute the full end-to-end pipeline with real-time stage progress reporting. | |
| If ``resume_run_id`` is provided, the pipeline resumes from the latest | |
| checkpoint for that ``run_id`` (thread_id) using the LangGraph SQLite | |
| checkpointer and ignores ``domain``/``max_pain_points``. | |
| """ | |
| if resume_run_id is not None: | |
| thread_id = resume_run_id | |
| initial_input = None | |
| else: | |
| if domain is None: | |
| raise ValueError("domain is required when not resuming from a previous run") | |
| state = make_initial_state(domain, max_pain_points=max_pain_points) | |
| thread_id = state.run_id | |
| initial_input = state | |
| config = { | |
| "recursion_limit": recursion_limit, | |
| "configurable": {"thread_id": thread_id}, | |
| } | |
| print(f"\n{'='*70}\n[VentureForge] Pipeline Execution (Thread ID: {thread_id})\n{'='*70}\n", flush=True) | |
| for event in GRAPH.stream(initial_input, config=config): | |
| for node_name, patch in event.items(): | |
| if node_name == "orchestrator": | |
| stage = patch.get("current_stage", "") | |
| next_node = patch.get("next_node", "") | |
| print(f"[Stage] Orchestrator: {stage} -> Next Worker: '{next_node}'", flush=True) | |
| elif node_name == "pain_point_miner": | |
| pps = patch.get("pain_points", []) | |
| print(f"\n[Stage] Pain Point Miner: Extracted {len(pps)} validated pain points:", flush=True) | |
| for i, p in enumerate(pps): | |
| title = getattr(p, "title", p.get("title", "")) if isinstance(p, dict) else p.title | |
| ev_count = len(getattr(p, "evidence", p.get("evidence", []))) if isinstance(p, dict) else len(p.evidence) | |
| print(f" {i+1}. {title} ({ev_count} verified quotes)", flush=True) | |
| print("", flush=True) | |
| elif node_name == "idea_generator": | |
| ideas = patch.get("ideas", []) | |
| print(f"\n[Stage] Idea Generator: Generated {len(ideas)} startup concepts in parallel:", flush=True) | |
| for i, idea in enumerate(ideas): | |
| title = getattr(idea, "title", idea.get("title", "")) if isinstance(idea, dict) else idea.title | |
| target = getattr(idea, "target_user", idea.get("target_user", "")) if isinstance(idea, dict) else idea.target_user | |
| print(f" {i+1}. {title} (Target: {target})", flush=True) | |
| print("", flush=True) | |
| elif node_name == "scorer": | |
| scored = patch.get("scored_ideas", []) | |
| print(f"\n[Stage] Scorer: Evaluated {len(scored)} ideas against binary rubric:", flush=True) | |
| for si in scored: | |
| rank = getattr(si, "rank", si.get("rank", 0) if isinstance(si, dict) else 0) | |
| yes_count = getattr(si, "yes_count", si.get("yes_count", 0) if isinstance(si, dict) else 0) | |
| total = getattr(si, "total_checks", si.get("total_checks", 8) if isinstance(si, dict) else 8) | |
| verdict = getattr(si, "verdict", si.get("verdict", "") if isinstance(si, dict) else "") | |
| idea_id = getattr(si, "idea_id", si.get("idea_id", "") if isinstance(si, dict) else "") | |
| print(f" * Idea {idea_id} -> Rank {rank} [{verdict.upper()}]: {yes_count}/{total} binary checks passed", flush=True) | |
| print("", flush=True) | |
| elif node_name == "pitch_writer": | |
| briefs = patch.get("pitch_briefs", []) | |
| print(f"\n[Stage] Pitch Writer: Drafted {len(briefs)} pitch briefs in parallel:", flush=True) | |
| for i, pb in enumerate(briefs): | |
| title = getattr(pb, "title", pb.get("title", "")) if isinstance(pb, dict) else pb.title | |
| opp = getattr(pb, "market_opportunity", pb.get("market_opportunity", "")) if isinstance(pb, dict) else pb.market_opportunity | |
| print(f" * Brief #{i+1}: {title} | Opp: {opp[:80]}...", flush=True) | |
| print("", flush=True) | |
| elif node_name == "critic": | |
| critique = patch.get("critique") | |
| if critique: | |
| all_pass = getattr(critique, "all_pass", critique.get("all_pass", False) if isinstance(critique, dict) else False) | |
| status = getattr(critique, "approval_status", critique.get("approval_status", "revise") if isinstance(critique, dict) else "revise") | |
| failing = getattr(critique, "failing_checks", critique.get("failing_checks", []) if isinstance(critique, dict) else []) | |
| print(f"\n[Stage] Critic Evaluation: {status.upper()} (All pass: {all_pass}, Failing checks: {len(failing)})\n", flush=True) | |
| checkpoint = GRAPH.get_state(config) | |
| return checkpoint.values | |
| def main() -> None: | |
| parser = argparse.ArgumentParser(description="VentureForge — AI Startup Discovery") | |
| parser.add_argument( | |
| "--domain", | |
| type=str, | |
| required=False, | |
| help="Target domain, e.g. 'developer tools' (ignored when using --resume)", | |
| ) | |
| parser.add_argument( | |
| "--max-pain-points", | |
| type=int, | |
| default=None, | |
| help="Override max pain points to extract (new runs only)", | |
| ) | |
| parser.add_argument( | |
| "--resume", | |
| type=str, | |
| default=None, | |
| help="Existing run_id to resume from LangGraph checkpoints", | |
| ) | |
| parser.add_argument( | |
| "--output", | |
| type=str, | |
| default="output.json", | |
| help="Output JSON file path", | |
| ) | |
| parser.add_argument( | |
| "--force-refresh", | |
| action="store_true", | |
| help="Bypass the evidence cache and re-mine live sources (new runs only)", | |
| ) | |
| args = parser.parse_args() | |
| if args.resume: | |
| print(f"Resuming VentureForge run: run_id='{args.resume}'") | |
| result = run_pipeline( | |
| domain=None, | |
| max_pain_points=None, | |
| resume_run_id=args.resume, | |
| ) | |
| else: | |
| if not args.domain: | |
| parser.error("--domain is required for new runs (omit it when using --resume)") | |
| if args.force_refresh: | |
| import src.agents.pain_point_miner as ppm | |
| ppm.force_refresh = True | |
| print("Force-refresh enabled: bypassing evidence cache.") | |
| print(f"VentureForge starting: domain='{args.domain}'") | |
| result = run_pipeline(args.domain, args.max_pain_points) | |
| # Serialize final state | |
| if isinstance(result, dict): | |
| from src.state.schema import VentureForgeState | |
| try: | |
| state = VentureForgeState(**result) | |
| output = state.model_dump(mode="json", exclude_none=True) | |
| except Exception as e: | |
| print(f"Warning: Could not convert result to VentureForgeState: {e}") | |
| output = result | |
| else: | |
| output = result.model_dump(mode="json", exclude_none=True) | |
| with open(args.output, "w", encoding="utf-8") as f: | |
| json.dump(output, f, indent=2, ensure_ascii=False) | |
| print(f"\nPipeline finished in stage: {output.get('current_stage', 'unknown')}") | |
| print(f" Run ID : {output.get('run_id', 'unknown')}") | |
| print(f" Duration : {output.get('agent_timings', {})}") | |
| print(f" Pain points : {len(output.get('pain_points', []))}") | |
| print(f" Ideas : {len(output.get('ideas', []))}") | |
| print(f" Total Pitches : {len(output.get('pitch_briefs', []))}") | |
| print(f" Approved Pitches: {len(output.get('approved_pitches', []))}") | |
| if output.get("quarantined_pitches"): | |
| print( | |
| f" Quarantined : {len(output.get('quarantined_pitches', []))} (failed rubric at max revisions)" | |
| ) | |
| revision_counts = output.get("revision_counts", {}) | |
| total_revisions = sum(revision_counts.values()) | |
| print(f" Revisions : {total_revisions} (across {len(revision_counts)} pitches)") | |
| print(f"\nOutput written to: {args.output}") | |
| if __name__ == "__main__": | |
| main() | |