mutiagent-startup/test.py at main · ssynb/mutiagent-startup · 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
import asyncio
import json

from agents.critic import CriticAgent
from agents.decision import DecisionAgent
from agents.finance import FinanceAgent
from agents.market import MarketAgent
from agents.product import ProductAgent
from agents.tech import TechAgent
from evals.quality import evaluate_startup_artifacts
from runtime.agent import AgentContext
from runtime.human_gate import append_human_review_request, create_human_review_request
from runtime.scheduler import AgentScheduler
from runtime.state import RuntimeState


async def main() -> None:
    user_input = """
I want to build an AI-powered technical feasibility analysis product.
The user gives a topic, target audience, and desired style.
The system researches the topic, creates an outline, generates slide content,
identifies technical risks, and exports a feasibility brief.
"""

    state = RuntimeState(
        run_id="run_001",
        user_input=user_input,
        data={
            "user_input": user_input,
        },
        pending_agents=["market"],
    )

    context = AgentContext(
        run_id="run_001",
        user_id="user_001",
        metadata={
            "workflow": "startup_planning",
        },
    )

    scheduler = AgentScheduler(
        agents={
            "market": MarketAgent(),
            "product": ProductAgent(),
            "tech": TechAgent(),
            "finance": FinanceAgent(),
            "critic": CriticAgent(),
            "decision": DecisionAgent(),
        }
    )

    final_state = await scheduler.run(state, context)
    eval_result = evaluate_startup_artifacts(final_state.data)
    final_state.data["_eval"] = eval_result.model_dump()

    if not eval_result.passed:
        append_human_review_request(
            final_state.data,
            create_human_review_request(
                run_id=final_state.run_id,
                reason="Final artifact evaluation did not pass",
                questions=[
                    "Should the workflow revise failed artifacts?",
                    "Which artifact should be revised first?",
                ],
                suggested_actions=[
                    "revise_market",
                    "revise_product",
                    "revise_tech",
                    "revise_finance",
                    "stop",
                ],
                related_agent="eval",
                related_data=eval_result.model_dump(),
                blocking=False,
            ),
        )

    print(json.dumps(
        final_state.model_dump(),
        ensure_ascii=False,
        indent=2,
    ))


if __name__ == "__main__":
    asyncio.run(main())