irishpulse/main.py at main · S-Shetty/irishpulse · 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
"""
IrishPulse — CLI entrypoint for interactive use.

Usage:
    python main.py                          # interactive REPL
    python main.py --query "What is Article 6?"
    python main.py --mode compliance --query "We use facial recognition for employee attendance"
"""
from __future__ import annotations

import sys
import typer
from rich import print as rprint
from rich.console import Console
from rich.panel import Panel
from rich.markdown import Markdown

from agents.orchestrator import run_query

app = typer.Typer()
console = Console()

BANNER = """
[bold green]
  ___      _     _     ____        _
 |_ _|_ __(_)___| |__ |  _ \ _  _| |___  ___
  | || '__| / __| '_ \| |_) | | | | / __|/ _ \\
  | || |  | \__ \ | | |  __/| |_| | \__ \  __/
 |___|_|  |_|___/_| |_|_|    \__,_|_|___/\___|

[/bold green]
[bold]EU AI Act Compliance & Market Intelligence for Irish SMEs[/bold]
Type [yellow]exit[/yellow] or [yellow]quit[/yellow] to stop.
"""


@app.command()
def main(
    query: str = typer.Option(None, "--query", "-q", help="Single query to run (non-interactive)"),
    mode: str = typer.Option("auto", "--mode", "-m", help="Force mode: auto | compliance | document | market"),
    session_id: str = typer.Option("cli", "--session", "-s", help="Session ID for memory"),
):
    console.print(BANNER)

    if query:
        _run_single(query, mode, session_id)
    else:
        _run_repl(session_id)


def _run_single(query: str, mode: str, session_id: str):
    if mode != "auto":
        prefixes = {
            "compliance": "Classify this AI system for EU AI Act compliance: ",
            "market": "Latest Irish AI news about: ",
            "document": "",
        }
        query = prefixes.get(mode, "") + query

    with console.status("[bold green]Thinking...[/bold green]"):
        state = run_query(query=query, session_id=session_id)

    console.print(Panel(Markdown(state.final_response), title=f"[bold]{state.route.title()} Agent[/bold]"))

    if state.compliance_result:
        cr = state.compliance_result
        console.print(
            f"[bold]Risk:[/bold] {cr.risk_category.value} | "
            f"[bold]Status:[/bold] {cr.compliance_flag.value} | "
            f"[bold]Confidence:[/bold] {cr.confidence_score:.0%}"
        )


def _run_repl(session_id: str):
    console.print("[dim]Enter your query below. The system will auto-route to the right agent.[/dim]\n")
    while True:
        try:
            query = console.input("[bold cyan]You:[/bold cyan] ").strip()
        except (KeyboardInterrupt, EOFError):
            console.print("\n[dim]Goodbye.[/dim]")
            break

        if query.lower() in {"exit", "quit", "q"}:
            console.print("[dim]Goodbye.[/dim]")
            break

        if not query:
            continue

        with console.status("[bold green]Thinking...[/bold green]"):
            try:
                state = run_query(query=query, session_id=session_id)
            except Exception as exc:
                console.print(f"[red]Error: {exc}[/red]")
                continue

        console.print(f"\n[bold dim]→ {state.route.title()} Agent[/bold dim]")
        console.print(Panel(Markdown(state.final_response)))
        console.print()


if __name__ == "__main__":
    app()