aiwiki/run_agent_loop.py at main · QuBrain/aiwiki · 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
"""Standalone script: run one agent loop cycle and exit.

Designed for Railway cron jobs. Shares the same database and env vars
as the main web service. Exits cleanly — no infinite loop, no sleeps.
"""

import logging
import sys

import core.database as db
from agents.base import validate_prompts
from agents.coordinator import Coordinator
from agents.critic import Critic
from agents.fact_checker import FactChecker
from agents.historian import Historian
from agents.quality_improver import QualityImprover
from agents.scientist import Scientist
from core import config

logging.basicConfig(
    level=getattr(logging, config.LOG_LEVEL, logging.INFO),
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger("aiwiki.agent_loop")


def main():
    prompt_errors = validate_prompts()
    if prompt_errors:
        for err in prompt_errors:
            logger.error("[Prompt Validation] %s", err)

    logger.info("Initializing database...")
    db.init_db()
    db.seed_topics_from_json()

    historian = Historian()
    scientist = Scientist()
    critic = Critic()
    fact_checker = FactChecker()
    quality_improver = QualityImprover(historian=historian, scientist=scientist)

    coordinator = Coordinator(
        historian=historian,
        scientist=scientist,
        critic=critic,
        fact_checker=fact_checker,
        quality_improver=quality_improver,
    )

    logger.info("Starting agent cycle...")
    try:
        result = coordinator.act({})
        action = result.get("action", "unknown")
        if action == "multi":
            steps = result.get("steps") or []
            for i, step in enumerate(steps, 1):
                step_action = step.get("action", "unknown")
                step_slug = step.get("slug", step.get("topic", ""))
                logger.info("[Step %d/%d] %s: %s", i, len(steps), step_action, step_slug)
            logger.info("Cycle complete: %d step(s)", len(steps))
        elif action == "noop":
            logger.info("Cycle complete: nothing to do (%s)", result.get("reason", ""))
        else:
            logger.info("Cycle complete: %s", action)
    except Exception as e:
        logger.error("Agent cycle failed: %s", e, exc_info=True)
        sys.exit(1)

    logger.info("Agent cycle finished successfully.")


if __name__ == "__main__":
    main()