群星(Star)- AI Agent 调度中心

1. 项目概述

1.1 项目名

  • 中文名:群星
  • 英文名:Star

1.2 命名寓意

“群星闪耀,各司其职。调度有序,光芒汇聚。”

多个 AI Agent 如同夜空中的群星,各自闪耀着独特的光芒。Star 调度中心就像星座的连线者,将这些独立的光芒串联成完整的图景,让每一颗星的输出都能汇聚成更强大的力量。

1.3 Slogan

群星(Star)—— 让每一个 AI Agent 都成为你的星座

1.4 核心目标

在 Windows 系统上,通过系统级 API(句柄、窗口消息、进程监控)实现对多个 AI 编程助手(Trae、CodeArts Atomcode、Cursor、Copilot 等)的统一调度与任务管理。

1.5 核心功能

  • 星图(Agent 发现):自动检测正在运行的 AI Agent 进程
  • 星轨(任务队列):创建、分配、修改、暂停、恢复任务
  • 星语(对话监控):截获 Agent 的输入/输出流
  • 星令(任务干预):向运行中的 Agent 注入新指令或修正方向
  • 星辉(结果收集):汇总各 Agent 输出,统一展示

2. 术语映射

通用术语 群星(Star)术语 说明
AI Agent 星(Star) 每个 Agent 是一颗星
Agent 进程 星体(Star Body) 运行中的 Agent 实例
任务队列 星轨(Orbit) 任务的流转路径
新任务 新星(Nova) 新创建的任务
任务分配 授星(Assign) 将任务交给某颗星
任务修改 调轨(Adjust Orbit) 修改运行中的任务方向
结果输出 星辉(Starlight) Agent 的产出
用户反馈 回响(Echo) 用户对 Agent 输出的回应
多 Agent 协同 星座(Constellation) 多个星协同完成复杂任务

3. 技术架构

┌─────────────────────────────────────────────────────────────┐
│                     群星(Star)前端                          │
│   (Web UI: React + Vite, 或 Electron 桌面端)                 │
│   星图面板 │ 星轨队列 │ 星语流 │ 星辉审查                      │
└───────────────────────────┬─────────────────────────────────┘
                            │ WebSocket / IPC
┌───────────────────────────▼─────────────────────────────────┐
│                    星核(Star Core)- Python 3.12             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ 寻星者        │  │ 星轨引擎      │  │ 星语路由         │  │
│  │ StarSeeker   │  │ OrbitEngine  │  │ StarlightRouter │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬─────────┘  │
│         │                 │                    │            │
│  ┌──────▼─────────────────▼────────────────────▼─────────┐  │
│  │              观星台(Observatory)                       │  │
│  │  · Win32 API (句柄/窗口/进程)                          │  │
│  │  · UI Automation (元素定位/文本注入)                   │  │
│  │  · 键盘/鼠标模拟 (SendInput/PostMessage)               │  │
│  │  · 剪贴板监控 (Clipboard API)                          │  │
│  └───────────────────────────────────────────────────────┘  │
└───────────────────────────┬─────────────────────────────────┘
                            │ 句柄 / UIA / 消息
┌───────────────────────────▼─────────────────────────────────┐
│                    群星闪耀(Agent 进程)                      │
│  ┌──────────┐  ┌──────────────┐  ┌──────────┐              │
│  │  Trae ☆  │  │CodeArts Atom☆│  │ Cursor ☆ │  ...         │
│  └──────────┘  └──────────────┘  └──────────┘              │
└─────────────────────────────────────────────────────────────┘

4. 核心模块命名与实现

4.1 寻星者(StarSeeker)- Agent 发现

class StarSeeker:
    """寻星者 - 发现并管理 AI Agent 进程"""
    
    # 已知星体特征
    STAR_SIGNATURES = {
        'trae': {
            'process_names': ['Trae.exe', 'trae.exe'],
            'window_class': ['Chrome_WidgetWin_1', 'TraeMainWindow'],
            'window_title_patterns': ['Trae', 'trae'],
        },
        'codearts_atomcode': {
            'process_names': ['CodeArts.exe', 'AtomCode.exe'],
            'window_class': ['Chrome_WidgetWin_1'],
            'window_title_patterns': ['CodeArts', 'AtomCode'],
        },
        'cursor': {
            'process_names': ['Cursor.exe'],
            'window_class': ['Chrome_WidgetWin_1'],
            'window_title_patterns': ['Cursor'],
        },
    }
    
    def scan_skies(self) -> list['StarBody']:
        """扫描天际,发现所有闪耀的星"""
        stars = []
        
        for proc in psutil.process_iter(['pid', 'name']):
            for star_type, sig in self.STAR_SIGNATURES.items():
                if proc.info['name'].lower() in [n.lower() for n in sig['process_names']]:
                    hwnd = self._locate_light(proc.info['pid'], sig)
                    if hwnd:
                        stars.append(StarBody(
                            star_type=star_type,
                            pid=proc.info['pid'],
                            hwnd=hwnd,
                            process=proc
                        ))
        return stars
    
    def _locate_light(self, pid: int, sig: dict) -> int:
        """定位星光 - 通过 PID 找到主窗口句柄"""
        def callback(hwnd, hwnds):
            if win32gui.IsWindowVisible(hwnd):
                _, found_pid = win32process.GetWindowThreadProcessId(hwnd)
                if found_pid == pid:
                    title = win32gui.GetWindowText(hwnd)
                    if any(p in title for p in sig['window_title_patterns']):
                        hwnds.append(hwnd)
            return True
        
        hwnds = []
        win32gui.EnumWindows(callback, hwnds)
        return hwnds[0] if hwnds else None

4.2 授星者(StarAssigner)- 文本注入

class StarAssigner:
    """授星者 - 向星体注入星令"""
    
    def send_starlight(self, star: 'StarBody', starlight: str) -> bool:
        """向星体发送星辉指令"""
        # 策略1: 观星术(UIA)- 最精准
        if self._try_observatory(star.hwnd, starlight):
            return True
        
        # 策略2: 星光传递(剪贴板)
        if self._try_starlight_transfer(star, starlight):
            return True
        
        # 策略3: 星波(消息)- 适用于原生控件
        if self._try_starwave(star.hwnd, starlight):
            return True
        
        return False
    
    def _try_observatory(self, hwnd: int, starlight: str) -> bool:
        """观星术 - 通过 UI Automation 精确定位输入框"""
        import uiautomation as uia
        
        try:
            window = uia.ControlFromHandle(hwnd)
            
            # 寻找星语输入口
            edit = window.EditControl(
                searchDepth=5,
                ClassName='TextArea'
            )
            
            if edit:
                edit.SetValue(starlight)
                edit.SendKeys('{Enter}')
                return True
        except Exception:
            pass
        return False
    
    def _try_starlight_transfer(self, star: 'StarBody', starlight: str) -> bool:
        """星光传递 - 剪贴板注入"""
        import pyperclip
        
        original = pyperclip.paste()
        try:
            pyperclip.copy(starlight)
            win32gui.SetForegroundWindow(star.hwnd)
            time.sleep(0.1)
            
            import pyautogui
            pyautogui.hotkey('ctrl', 'v')
            time.sleep(0.1)
            pyautogui.press('enter')
            
            return True
        finally:
            pyperclip.copy(original)

4.3 观星者(StarGazer)- 输出捕获

class StarGazer:
    """观星者 - 捕获星体的输出"""
    
    def gaze(self, star: 'StarBody') -> str:
        """凝视星辉 - 获取 Agent 最后一条回复"""
        import uiautomation as uia
        
        window = uia.ControlFromHandle(star.hwnd)
        
        output_controls = window.TextControl(
            searchDepth=10,
            ClassName='TextBlock'
        )
        
        if output_controls:
            return output_controls.Name
        
        return ""
    
    def continuous_gaze(self, star: 'StarBody', on_starlight_change):
        """持续观星 - 监控输出变化"""
        import threading
        
        def gaze_thread():
            last_starlight = ""
            while star.is_shining:
                current = self.gaze(star)
                if current != last_starlight:
                    on_starlight_change(star, current)
                    last_starlight = current
                time.sleep(1)
        
        threading.Thread(target=gaze_thread, daemon=True).start()

5. 星轨引擎(OrbitEngine)- 任务调度

5.1 星轨任务模型

from dataclasses import dataclass
from enum import Enum
from datetime import datetime
from typing import Optional

class StarStatus(Enum):
    """星芒状态"""
    NASCENT = "nascent"           # 初生(新创建)
    ORBITING = "orbiting"         # 入轨(已分配)
    SHINING = "shining"           # 闪耀(Agent 正在处理)
    AWAITING_ECHO = "awaiting"    # 待回响(等待用户审查)
    CONSTELLATED = "constellated" # 成星(完成)
    FADED = "faded"               # 暗淡(失败)
    DARKENED = "darkened"         # 熄灭(取消)

class StarPriority(Enum):
    """星等(优先级)"""
    DIM = 0        # 暗星(低)
    NORMAL = 1     # 常星(正常)
    BRIGHT = 2     # 亮星(高)
    SUPERNOVA = 3  # 超新星(紧急)

@dataclass
class Nova:
    """新星 - 任务数据模型"""
    id: str
    title: str
    description: str
    starlight: str                # 发送给星的指令
    context_files: list[str]      # 上下文文件
    assigned_star: Optional[str]  # 分配的目标星
    status: StarStatus
    priority: StarPriority
    created_at: datetime
    updated_at: datetime
    result_starlight: Optional[str]  # 星辉(Agent 返回结果)
    starlight_log: list[dict]        # 星光日志(对话历史)
    echo: Optional[str]              # 回响(用户反馈)

5.2 星轨引擎

class OrbitEngine:
    """星轨引擎 - 任务调度核心"""
    
    def __init__(self, star_seeker: 'StarSeeker'):
        self.star_seeker = star_seeker
        self.orbit_queue = asyncio.Queue()
        self.active_novas = {}
    
    async def birth_nova(self, nova: Nova) -> str:
        """诞生新星 - 提交新任务"""
        if not nova.assigned_star:
            nova.assigned_star = self._calculate_orbit(nova)
        
        nova.status = StarStatus.NASCENT
        await self.orbit_queue.put(nova)
        return nova.id
    
    def _calculate_orbit(self, nova: Nova) -> str:
        """计算星轨 - 根据任务特征路由到合适的星"""
        star_affinity = {
            'trae': ['生成', '创建', '编写', 'generate', 'create'],
            'codearts': ['审查', 'review', '检查', '漏洞'],
            'cursor': ['重构', 'refactor', '迁移', '多文件'],
        }
        
        for star_type, keywords in star_affinity.items():
            if any(kw in nova.description for kw in keywords):
                available = self.star_seeker.get_idle_stars(star_type)
                if available:
                    return star_type
        
        return self.star_seeker.get_any_idle_star()
    
    async def adjust_orbit(self, nova_id: str, new_starlight: str):
        """调轨 - 修改运行中的任务"""
        nova = self.active_novas.get(nova_id)
        if not nova:
            raise ValueError(f"Nova {nova_id} not found in sky")
        
        star = self.star_seeker.get_star(nova.assigned_star)
        if star and nova.status == StarStatus.SHINING:
            correction_starlight = f"""
            [星核指令] 对当前星轨进行调整:
            原始星图:{nova.starlight}
            调轨指令:{new_starlight}
            请忽略之前的中间星光,基于调轨指令重新闪耀。
            """
            star.send_starlight(correction_starlight)
            nova.starlight = new_starlight
            nova.starlight_log.append({
                'role': 'star_core',
                'content': '星轨已调整'
            })

6. 前端界面设计

6.1 主界面布局

┌─────────────────────────────────────────────────┐
│  ✦ 群星 Star                      [⚙ 星图设置] [📊]│
├──────────┬───────────────────┬──────────────────┤
│ 星图      │    星轨面板        │   星语监控        │
│          │                   │                  │
│ 🟢 Trae  │ 🌟 新星待升 (3)   │ [星辉实时]       │
│ 🟡 Atom  │  · 生成API星图    │                  │
│ 🔴Cursor │  · 修复Bug #42   │ 正在闪耀...       │
│          │  · 优化星光查询   │                  │
│          │                   │                  │
│ [+ 唤星] │ 💫 闪耀中 (2)     │                  │
│          │  · 重构星云A     │                  │
│          │  · 撰写星光测试  │                  │
│          │                   │                  │
│          │ ✨ 已成星 (5)     │                  │
│          │                   │                  │
├──────────┴───────────────────┴──────────────────┤
│ 星令: [___________________] [授星: Trae ▼] [发送]│
└─────────────────────────────────────────────────┘

6.2 颜色系统

  • 主色调:深空蓝黑(#0a0e27 背景,#1a1f3a 面板)
  • 强调色:星光金(#ffd700
  • 星等色
    • 超新星:炽白(#ffffff
    • 亮星:亮蓝(#4fc3f7
    • 常星:银白(#b0bec5
    • 暗星:暗灰(#616161
  • 星芒色
    • 闪耀中:翠绿(#66bb6a
    • 待回响:琥珀(#ffa726
    • 暗淡:赤红(#ef5350

6.3 图标系统

使用星星相关的 SVG 图标:

  • ⭐ 普通星
  • 🌟 新星
  • 💫 闪耀中
  • ✨ 已成星
  • 🔭 观星台
  • 🌌 星图

7. 项目结构

star/
├── star_core/                    # 星核(核心引擎)
│   ├── __init__.py
│   ├── star_seeker.py           # 寻星者(Agent 发现)
│   ├── star_assigner.py         # 授星者(文本注入)
│   ├── star_gazer.py            # 观星者(输出捕获)
│   ├── orbit_engine.py          # 星轨引擎(任务调度)
│   └── observatory.py           # 观星台(Windows API 封装)
│
├── star_api/                     # 星光接口(后端 API)
│   ├── __init__.py
│   ├── main.py                  # FastAPI 入口
│   ├── routes/
│   │   ├── stars.py             # 星管理接口
│   │   ├── novas.py             # 新星(任务)接口
│   │   └── constellations.py    # 星座(协同)接口
│   └── websocket/
│       └── starlight_stream.py  # 星光流推送
│
├── star_web/                     # 星图(前端)
│   ├── src/
│   │   ├── components/
│   │   │   ├── StarMap.tsx       # 星图面板
│   │   │   ├── OrbitPanel.tsx    # 星轨面板
│   │   │   ├── StarlightMonitor.tsx  # 星语监控
│   │   │   └── StarCommand.tsx   # 星令输入
│   │   ├── pages/
│   │   │   └── Observatory.tsx   # 观星台主页
│   │   └── assets/
│   │       └── star_icons/       # 星星图标
│   └── package.json
│
├── pyproject.toml
└── README.md

8. 星语(日志)系统

from loguru import logger

# 星语级别
logger.add(
    "logs/star_core.log",
    format="⭐ {time:YYYY-MM-DD HH:mm:ss} | {level} | {message}",
    level="INFO",
    rotation="1 day",
    retention="7 days"
)

# 关键事件星语
def log_nova_birth(nova_id: str, star_type: str):
    logger.info(f"🌟 新星诞生 | ID: {nova_id} | 目标星: {star_type}")

def log_star_shine(nova_id: str, star_type: str):
    logger.info(f"💫 星光闪耀 | 任务: {nova_id} | 星: {star_type}")

def log_constellation_complete(nova_id: str):
    logger.info(f"✨ 星座完成 | 任务: {nova_id}")

def log_star_fade(nova_id: str, error: str):
    logger.error(f"🌑 星光暗淡 | 任务: {nova_id} | 原因: {error}")

9. 实现路线图(星图里程碑)

阶段 里程碑 内容
🌑 朔月 星核初现 进程发现、窗口句柄获取、基础 UI
🌒 娥眉 星光通联 UI Automation 文本注入/读取
🌓 上弦 星轨流转 任务队列、状态管理、星轨引擎
🌔 盈凸 星语交响 WebSocket 实时推送、对话监控
🌕 满月 群星闪耀 多星协同(星座)、结果对比
🌖 亏凸 星图完善 插件系统、自定义星体
🌗 下弦 星辉永驻 历史记录、统计分析
🌘 残月 星辰大海 正式发布、文档完善

这个名字体系将整个项目包装成一个完整的“星空叙事”,每个技术概念都有对应的诗意表达,让项目不仅实用,更具有独特的品牌识别度。

Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐