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