CellForge/run_method_design.py at main · gersteinlab/CellForge · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
#!/usr/bin/env python3
"""
Direct Method Design Runner
Run Method Design module directly with a specific task analysis file
"""

import os
import sys
from pathlib import Path

# Load environment variables from .env file
try:
    from dotenv import load_dotenv
    # 明确指定.env文件路径
    env_path = Path(__file__).parent / ".env"
    load_dotenv(env_path)
    print(f"Loaded .env from: {env_path}")
except ImportError:
    print("Warning: python-dotenv not installed, using system environment variables")

# Add cellforge to Python path
project_root = Path(__file__).parent
cellforge_path = project_root / "cellforge"
sys.path.insert(0, str(cellforge_path))

from Method_Design.main import load_task_analysis, RAGRetriever
from Method_Design import generate_research_plan

def main():
    """Run Method Design with specific task analysis file"""
    print("=== Direct Method Design Runner ===")

    # Specify the task analysis file path
    task_analysis_file = "cellforge/data/plans/research_plan_20240101_120000.jsonson"

    # Resolve path
    file_path = Path(task_analysis_file)
    if not file_path.is_absolute():
        file_path = project_root / file_path

    print(f"Loading task analysis from: {file_path}")

    try:
        # Load task analysis
        task_analysis = load_task_analysis(str(file_path))
        print("✅ Successfully loaded task analysis file")

        # Create RAG retriever
        print("Initializing RAG knowledge retriever...")
        rag_retriever = RAGRetriever()

        # Set output directory
        output_dir = "cellforge/data/results"

        # Generate research plan
        print("Generating research plan...")
        plan = generate_research_plan(
            task_analysis=task_analysis,
            rag_retriever=rag_retriever,
            task_type=task_analysis.get("task_type", "gene_knockout"),
            output_dir=output_dir
        )

        print("\n✅ Research plan generated successfully!")
        print(f"Output directory: {output_dir}")

        # Show summary
        if 'discussion_summary' in plan:
            summary = plan['discussion_summary']
            print(f"Discussion rounds: {summary.get('rounds', 'N/A')}")
            print(f"Consensus reached: {summary.get('consensus_reached', 'N/A')}")

        if 'expert_contributions' in plan:
            experts = plan['expert_contributions']
            print(f"Participating experts: {len(experts)}")

            # Show expert contributions
            print("\nExpert contributions:")
            for expert_name, contribution in list(experts.items())[:5]:  # Show top 5
                confidence = contribution.get('confidence', 0)
                print(f"  - {expert_name}: confidence {confidence:.2f}")

        # Show generated files with dynamic names
        if 'generated_files' in plan:
            files_info = plan['generated_files']
            base_filename = files_info['base_filename']
            print(f"\nGenerated files:")
            print(f"  - {output_dir}/{base_filename}.md (Research plan)")
            print(f"  - {output_dir}/{base_filename}.json (Detailed data)")
            print(f"  - {output_dir}/{base_filename}.mmd (Architecture diagram)")
            print(f"  - {output_dir}/{base_filename}_consensus.png (Consensus progress)")

            # Show code generation result
            if 'code_generation' in plan:
                code_info = plan['code_generation']
                if code_info['status'] == 'success':
                    print(f"  - {output_dir}/result.py (Generated code)")
                    print(f"✅ Code generation completed successfully")
                elif code_info['status'] == 'failed':
                    print(f"❌ Code generation failed: {code_info.get('error', 'Unknown error')}")
                elif code_info['status'] == 'error':
                    print(f"❌ Code generation error: {code_info.get('error', 'Unknown error')}")
        else:
            # Fallback to old format
            print(f"\nGenerated files:")
            print(f"  - {output_dir}/research_plan.md (Research plan)")
            print(f"  - {output_dir}/research_plan.json (Detailed data)")
            print(f"  - {output_dir}/architecture.mmd (Architecture diagram)")
            print(f"  - {output_dir}/consensus_progress.png (Consensus progress)")

        print("\n=== Complete ===")

    except Exception as e:
        print(f"\n❌ Generation failed: {e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()