Coverage for agentos/cli/init.py: 9%
279 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09:59 +0800
1"""
2`agentos init` — 交互式配置向导。
4功能：
5 - 检测当前配置状态
6 - 引导选择 Provider + 输入 API Key
7 - 写入 ~/.agentos/config.yaml
8 - 可选写入 .env 文件（当前或全局）
10命令：
11 agentos init # 交互式引导
12 agentos init --quick # 跳过问答，直接生成 .env.example
13 agentos init --reset # 重置配置
14"""
16from __future__ import annotations
18import json
19import os
20import re
21import sys
22from pathlib import Path
23from typing import Optional
26CONFIG_DIR = Path.home() / ".agentos"
27CONFIG_FILE = CONFIG_DIR / "config.yaml"
28ENV_FILE = CONFIG_DIR / ".env"
30PROVIDERS = {
31 "openai": {
32 "label": "OpenAI",
33 "models": ["gpt-4o-mini", "gpt-4o", "o3-mini"],
34 "default_model": "gpt-4o-mini",
35 "env_var": "OPENAI_API_KEY",
36 "key_prefix": "sk-",
37 "website": "https://platform.openai.com/api-keys",
38 "cost": "低 ~ 中",
39 },
40 "deepseek": {
41 "label": "DeepSeek",
42 "models": ["deepseek-chat", "deepseek-reasoner"],
43 "default_model": "deepseek-chat",
44 "env_var": "DEEPSEEK_API_KEY",
45 "key_prefix": "sk-",
46 "website": "https://platform.deepseek.com/api_keys",
47 "cost": "低",
48 },
49 "anthropic": {
50 "label": "Anthropic (Claude)",
51 "models": ["claude-sonnet-4", "claude-haiku-3-5", "claude-opus-4"],
52 "default_model": "claude-sonnet-4",
53 "env_var": "ANTHROPIC_API_KEY",
54 "key_prefix": "sk-ant-",
55 "website": "https://console.anthropic.com/keys",
56 "cost": "中 ~ 高",
57 },
58}
61def _detect_current_config() -> dict:
62 """检测当前环境的配置状态。"""
63 config = {"providers": {}, "configured_providers": [], "active": None}
65 for name, info in PROVIDERS.items():
66 key = os.environ.get(info["env_var"]) or ""
67 masked = key[:8] + "..." + key[-4:] if len(key) > 20 else ""
68 config["providers"][name] = {
69 "env_set": bool(key),
70 "key_preview": masked,
71 }
72 if key:
73 config["configured_providers"].append(name)
75 # Check config file
76 if CONFIG_FILE.exists():
77 config["config_file_exists"] = True
78 try:
79 content = CONFIG_FILE.read_text()
80 for name in PROVIDERS:
81 if f"{PROVIDERS[name]['env_var']}:" in content:
82 config["providers"][name]["in_config"] = True
83 except Exception:
84 pass
86 # Determine active provider
87 for name in ["openai", "deepseek", "anthropic"]:
88 if config["providers"][name]["env_set"] or config["providers"].get(name, {}).get("in_config"):
89 config["active"] = name
90 break
92 return config
95def _print_banner():
96 """打印欢迎横幅。"""
97 from agentos import __version__
98 print(f"\n ╔══════════════════════════════════════════════╗")
99 print(f" ║ Nexus AgentOS v{__version__:8s} ║")
100 print(f" ║ 交互式配置向导 ║")
101 print(f" ╚══════════════════════════════════════════════╝")
102 print()
105def _print_status(config: dict):
106 """打印当前配置状态。"""
107 print(" ── 当前环境检测 ──")
108 print()
109 for name, info in config["providers"].items():
110 p = PROVIDERS[name]
111 status = "✅" if info["env_set"] else "⬜"
112 key_info = info.get("key_preview", "") or "未配置"
113 in_config = " (配置文件)" if info.get("in_config") else ""
114 print(f" {status} {p['label']:20s} {key_info:25s}{in_config}")
115 print()
118def _select_provider() -> str:
119 """交互选择 Provider。"""
120 print(" ── 选择 LLM 服务商 ──")
121 print()
122 names = list(PROVIDERS.keys())
123 for i, name in enumerate(names, 1):
124 p = PROVIDERS[name]
125 print(f" [{i}] {p['label']:20s} 模型: {p['default_model']:15s} 成本: {p['cost']}")
126 print()
128 while True:
129 try:
130 choice = input(" 请选择 (1-3) [1]: ").strip()
131 if not choice:
132 return "openai"
133 idx = int(choice) - 1
134 if 0 <= idx < len(names):
135 return names[idx]
136 except ValueError:
137 pass
138 print(" 输入无效，请输入数字 1-3。")
141def _input_api_key(provider_name: str) -> Optional[str]:
142 """交互输入 API Key。"""
143 p = PROVIDERS[provider_name]
144 print()
145 print(f" ── 配置 {p['label']} API Key ──")
146 print()
147 print(f" ① 打开 {p['website']}")
148 print(f" ② 创建或复制一个 API Key")
149 print(f" ③ 粘贴到下方（输入后按回车）")
150 print()
152 existing = os.environ.get(p["env_var"], "")
153 if existing:
154 preview = existing[:8] + "..." + existing[-4:] if len(existing) > 20 else existing
155 use_existing = input(f" 检测到环境变量已设置 ({preview})，直接使用？(Y/n): ").strip().lower()
156 if use_existing in ("", "y", "yes"):
157 return existing
159 while True:
160 key = input(f" 请输入 {p['label']} API Key: ").strip()
161 if not key:
162 print(" API Key 不能为空。输入 q 取消。")
163 continue
164 if key.lower() == "q":
165 return None
166 # Basic validation
167 prefix = p["key_prefix"]
168 if prefix and not key.startswith(prefix):
169 warn = input(f" 警告：{p['label']} 的 Key 通常以 '{prefix}' 开头，"
170 f"确认继续？(y/N): ").strip().lower()
171 if warn not in ("y", "yes"):
172 continue
173 return key
176def _test_connection(provider_name: str, api_key: str) -> bool:
177 """测试 API 连接（发一条最轻的请求）。"""
178 p = PROVIDERS[provider_name]
179 print(f"\n 正在测试 {p['label']} API 连接...", end=" ")
181 try:
182 if provider_name == "openai":
183 import httpx
184 resp = httpx.get(
185 "https://api.openai.com/v1/models",
186 headers={"Authorization": f"Bearer {api_key}"},
187 timeout=10,
188 )
189 if resp.status_code == 200:
190 print("✅ 成功")
191 return True
192 elif resp.status_code == 401:
193 print("❌ Key 无效（401 Unauthorized）")
194 return False
195 else:
196 print(f"⚠️ 返回 {resp.status_code}，Key 格式正确但不一定可用")
197 return True
198 elif provider_name == "deepseek":
199 import httpx
200 resp = httpx.post(
201 "https://api.deepseek.com/chat/completions",
202 headers={
203 "Authorization": f"Bearer {api_key}",
204 "Content-Type": "application/json",
205 },
206 json={"model": "deepseek-chat", "messages": [{"role": "user",
207 "content": "hi"}], "max_tokens": 1},
208 timeout=10,
209 )
210 if resp.status_code == 200:
211 print("✅ 成功")
212 return True
213 elif resp.status_code == 401:
214 print("❌ Key 无效（401）")
215 return False
216 else:
217 print(f"⚠️ 返回 {resp.status_code}")
218 return True
219 elif provider_name == "anthropic":
220 import httpx
221 resp = httpx.post(
222 "https://api.anthropic.com/v1/messages",
223 headers={
224 "x-api-key": api_key,
225 "anthropic-version": "2023-06-01",
226 "Content-Type": "application/json",
227 },
228 json={"model": "claude-sonnet-4-20250514", "max_tokens": 1,
229 "messages": [{"role": "user", "content": "hi"}]},
230 timeout=10,
231 )
232 if resp.status_code == 200:
233 print("✅ 成功")
234 return True
235 elif resp.status_code == 401:
236 print("❌ Key 无效（401）")
237 return False
238 else:
239 print(f"⚠️ 返回 {resp.status_code}")
240 return True
241 except Exception as e:
242 print(f"⚠️ 连接异常: {e}")
243 return False
244 return False
247def _save_config(provider_name: str, api_key: str):
248 """保存配置到 ~/.agentos/config.yaml。"""
249 CONFIG_DIR.mkdir(parents=True, exist_ok=True)
250 env_var = PROVIDERS[provider_name]["env_var"]
252 # Save .env file
253 env_content = f"# Nexus AgentOS — {PROVIDERS[provider_name]['label']} 配置\n"
254 env_content += f"{env_var}={api_key}\n\n"
255 env_content += "# 可选：配置多个 Provider 以实现自动回退\n"
256 env_content += "# OPENAI_API_KEY=sk-xxx\n"
257 env_content += "# DEEPSEEK_API_KEY=sk-xxx\n"
258 env_content += "# ANTHROPIC_API_KEY=sk-ant-xxx\n"
259 ENV_FILE.write_text(env_content)
261 # Save config.yaml
262 config = {
263 "version": "1.4.0",
264 "active_provider": provider_name,
265 "providers": {
266 provider_name: {
267 "env_var": env_var,
268 }
269 },
270 }
271 import yaml
272 with open(CONFIG_FILE, "w") as f:
273 yaml.dump(config, f, default_flow_style=False)
275 print(f"\n ✅ 配置已保存")
276 print(f" {CONFIG_FILE}")
277 print(f" {ENV_FILE}")
280def _show_completion_message(provider_name: str):
281 """显示配置完成后引导。"""
282 from agentos import __version__
283 p = PROVIDERS[provider_name]
285 print()
286 print(" ╔══════════════════════════════════════════════╗")
287 print(" ║ ✅ 配置就绪！ ║")
288 print(" ╚══════════════════════════════════════════════╝")
289 print()
290 print(f" 当前已配置: {p['label']} ({p['default_model']})")
291 print()
292 print(" ── 快速开始 ──")
293 print()
294 print(f" # 运行任务")
295 print(f" agentos \"列出当前目录的文件\"")
296 print()
297 print(f" # 运行端到端示例")
298 print(f" python -m examples.multi_agent_research --topic \"量子计算\"")
299 print()
300 if provider_name != "openai":
301 print(f" # 指定使用 {p['label']}")
302 print(f" agentos --provider {provider_name} \"写一个 Python 爬虫\"")
303 print()
304 print(" ── 多 Provider 配置（可选） ──")
305 print()
306 print(f" 编辑 {ENV_FILE}，添加其他 API Key 即可实现自动回退:")
307 print(f" OPENAI_API_KEY=sk-xxx # 默认使用")
308 print(f" DEEPSEEK_API_KEY=sk-xxx # 回退 1")
309 print(f" ANTHROPIC_API_KEY=sk-ant-xxx # 回退 2")
310 print()
311 print(f" 重新运行 agentos init 修改配置。")
312 print(f" 或 agentos config-panel 打开浏览器版配置面板。")
315# ── 配置加载接口 ────────────────────────────────────────────
318def load_config() -> dict:
319 """加载 ~/.agentos/config.yaml 和环境变量。
321 Returns:
322 dict: 包含 providers 和 active_provider 的配置字典。
323 """
324 config = {"providers": {}, "active_provider": None}
326 # 1. Load env vars
327 for name, info in PROVIDERS.items():
328 key = os.environ.get(info["env_var"])
329 if key:
330 config["providers"][name] = key
332 # 2. Load config file
333 if CONFIG_FILE.exists():
334 try:
335 import yaml
336 raw = yaml.safe_load(CONFIG_FILE.read_text())
337 if raw and "providers" in raw:
338 for name, pcfg in raw["providers"].items():
339 if name in PROVIDERS and name not in config["providers"]:
340 env_var = pcfg.get("env_var", PROVIDERS[name]["env_var"])
341 # Try to load from .env
342 if ENV_FILE.exists():
343 for line in ENV_FILE.read_text().splitlines():
344 if line.startswith(env_var + "="):
345 val = line.split("=", 1)[1].strip()
346 if val and val != "sk-xxx":
347 config["providers"][name] = val
348 break
349 if raw and "active_provider" in raw:
350 config["active_provider"] = raw["active_provider"]
351 except Exception:
352 pass
354 # 3. Determine active
355 for name in ["openai", "deepseek", "anthropic"]:
356 if config["providers"].get(name):
357 config["active_provider"] = config.get("active_provider") or name
358 break
360 return config
363def config_status_text() -> str:
364 """返回一行配置状态文本，给 CLI help 用。"""
365 config = load_config()
366 if config["active_provider"]:
367 name = config["active_provider"]
368 label = PROVIDERS.get(name, {}).get("label", name)
369 return f"✅ {label} 已配置"
370 return "⬜ 未配置（运行 agentos init）"
373# ── CLI ────────────────────────────────────────────────────
376def init_cli(args: list[str]):
377 """CLI 入口。"""
378 quick = "--quick" in args
379 reset = "--reset" in args
381 if reset:
382 if CONFIG_FILE.exists():
383 CONFIG_FILE.unlink()
384 if ENV_FILE.exists():
385 ENV_FILE.unlink()
386 print(" ✅ 配置已重置。运行 agentos init 重新配置。")
387 return
389 if quick:
390 # Quick mode: just create .env.example in current directory
391 example_path = Path.cwd() / ".env.example"
392 content = """# Nexus AgentOS 配置示例
393# 复制为 .env 并填入你的 API Key
394OPENAI_API_KEY=sk-xxx
395DEEPSEEK_API_KEY=sk-xxx
396ANTHROPIC_API_KEY=sk-ant-xxx
397"""
398 example_path.write_text(content)
399 print(f" 已生成 {example_path}")
400 print(" 复制为 .env 并填入你的 API Key 即可使用。")
401 return
403 # Interactive mode
404 _print_banner()
405 current = _detect_current_config()
406 _print_status(current)
408 if current["configured_providers"]:
409 print(f" 检测到已有 API Key 配置。")
410 reconfig = input(" 是否重新配置？(y/N): ").strip().lower()
411 if reconfig not in ("y", "yes"):
412 _show_completion_message(current["active"] or current["configured_providers"][0])
413 return
415 provider = _select_provider()
416 api_key = _input_api_key(provider)
417 if api_key is None:
418 print(" 配置已取消。")
419 return
421 test_result = _test_connection(provider, api_key)
422 if test_result is False:
423 retry = input(" Key 验证失败，重试？(Y/n): ").strip().lower()
424 if retry not in ("", "y", "yes"):
425 print(" 配置已取消。")
426 return
427 # Try again recursively for simplicity
428 return init_cli(args)
430 _save_config(provider, api_key)
431 _show_completion_message(provider)
434# ── 项目脚手架（兼容旧接口） ──────────────────────────────────
436TEMPLATES = {
437 "default": {
438 "agentos.yaml": """\
439# AgentOS v0.80 配置文件
440version: "0.80.0"
442models:
443 primary:
444 provider: openai
445 model_name: gpt-4o-mini
446 temperature: 0.7
447 max_tokens: 4096
449loop:
450 max_iterations: 10
451 step_timeout: 120
453observability:
454 tracer:
455 enabled: true
456 level: info
457""",
458 "main.py": """\
459\"""AgentOS — 我的 Agent 应用入口。\"""
461from agentos import AgentLoop, LoopConfig
464def main():
465 loop = AgentLoop(LoopConfig(max_iterations=5))
466 result = loop.run("你好，世界！")
467 print(result.output)
470if __name__ == "__main__":
471 main()
472""",
473 ".env.example": """\
474# AgentOS 环境变量
475OPENAI_API_KEY=sk-xxx
476ANTHROPIC_API_KEY=sk-ant-xxx
477GEMINI_API_KEY=AIza-xxx
478""",
479 },
480 "minimal": {
481 "agentos.yaml": """\
482version: "0.80.0"
483models:
484 primary:
485 provider: openai
486 model_name: gpt-4o-mini
487""",
488 "main.py": """\
489from agentos import AgentLoop, LoopConfig
491loop = AgentLoop(LoopConfig(max_iterations=3))
492result = loop.run("你好，世界！")
493print(result.output)
494""",
495 },
496}
499def scaffold(project_dir: str, template: str = "default") -> list[str]:
500 """初始化 AgentOS 项目脚手架。
502 Args:
503 project_dir: 项目根目录路径。
504 template: 模板名称（"default" 或 "minimal"）。
506 Returns:
507 创建的文件路径列表。
508 """
509 files = TEMPLATES.get(template, TEMPLATES["default"])
510 project_path = Path(project_dir).resolve()
511 project_path.mkdir(parents=True, exist_ok=True)
513 created = []
514 for filename, content in files.items():
515 filepath = project_path / filename
516 if filepath.exists():
517 continue
518 with open(filepath, "w") as f:
519 f.write(content)
520 created.append(str(filepath))
522 return created
525if __name__ == "__main__":
526 init_cli(sys.argv[1:])