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