diff --git a/.env b/.env new file mode 100644 index 0000000..081a5b3 --- /dev/null +++ b/.env @@ -0,0 +1,10 @@ +# 数据存储配置 +# 设置为 true 启用数据库存储，false 则使用JSON文件（默认） +DB_ENABLED=true + +# MariaDB连接信息 +DB_HOST=127.0.0.1 +DB_PORT=33039 +DB_USER=root +DB_PASSWORD=root +DB_NAME=rocom_data diff --git a/battle.py b/battle.py index 473ef05..a84b153 100644 --- a/battle.py +++ b/battle.py @@ -27,6 +27,7 @@ list_teams, build_team, add_team, delete_team, rename_team, get_team_def ) from sim.mcts_agent import MCTSAgent +from sim.human_agent import HumanAgent # MCTS 每回合迭代次数（可调整：20 快速 / 100 标准 / 200 强力） _MCTS_ITERS_BATTLE = 100 # 单局对战 @@ -58,12 +59,17 @@ def _status_flags(p: Pokemon) -> str: return " ".join(parts) +def _ability_tag(p: Pokemon) -> str: + """精灵特性标签，无特性则返回空串""" + return f" [{p.ability}]" if p.ability else "" + + def _print_field(state: BattleState, label_a: str, label_b: str) -> None: pa, pb = state.get_current("a"), state.get_current("b") weather_str = f" 天气:{state.weather.value}" if state.weather.value != "none" else "" print(f"\n{LINE} 回合 {state.turn}{weather_str}") - print(f" {label_a}: {pa.name:<10} {_hp_bar(pa.current_hp, pa.hp)} 能量:{pa.energy:2} {_status_flags(pa)}") - print(f" {label_b}: {pb.name:<10} {_hp_bar(pb.current_hp, pb.hp)} 能量:{pb.energy:2} {_status_flags(pb)}") + print(f" {label_a}: {pa.name:<10}{_ability_tag(pa)} {_hp_bar(pa.current_hp, pa.hp)} 能量:{pa.energy:2} {_status_flags(pa)}") + print(f" {label_b}: {pb.name:<10}{_ability_tag(pb)} {_hp_bar(pb.current_hp, pb.hp)} 能量:{pb.energy:2} {_status_flags(pb)}") def _print_team_summary(label: str, team: List[Pokemon]) -> None: @@ -102,7 +108,7 @@ def _pick_team(prompt: str = "选择队伍序号") -> Optional[str]: # ============================================================ -# 核心：单场对战 +# 核心：单场对战 — 接受任意 AgentProtocol 组合 # ============================================================ def run_battle( team_a: List[Pokemon], @@ -110,33 +116,58 @@ def run_battle( label_a: str = "A队", label_b: str = "B队", verbose: bool = True, + agent_a=None, # AgentProtocol 实例；None=自动创建 MCTSAgent + agent_b=None, # AgentProtocol 实例；None=自动创建 MCTSAgent ) -> Optional[str]: """ - 运行一场对战（随机 AI），verbose=True 时实时打印日志。 - 返回胜者 "a" / "b" 或 None（平局/超时）。 + 运行一场对战，支持任意智能体组合。 + + Parameters + ---------- + agent_a / agent_b : AgentProtocol — 自定义智能体（MCTS/Human/LLM） + None 时自动创建 MCTSAgent + + Returns + ------- + "a" / "b" / None（平局/超时） """ + from sim.agent_base import AgentProtocol + state = BattleState(team_a=team_a, team_b=team_b) - engine = BattleEngine(state, verbose=verbose) + engine = BattleEngine(state, verbose=True) # 引擎日志始终打开 + engine.label_a = label_a + engine.label_b = label_b - if verbose: + if agent_a is None: + agent_a = MCTSAgent("a", label_a, iterations=_MCTS_ITERS_BATTLE) + if agent_b is None: + agent_b = MCTSAgent("b", label_b, iterations=_MCTS_ITERS_BATTLE) + + history = [] + + # 判断是否有人类玩家（人类参与时引擎日志关闭，由 HumanAgent 打印） + has_human = ( + isinstance(agent_a, AgentProtocol) and getattr(agent_a, 'show_team_status', False) + or isinstance(agent_b, AgentProtocol) and getattr(agent_b, 'show_team_status', False) + ) + + if verbose and not has_human: print(f"\n{SEP}") - print(f" {label_a} VS {label_b}") + ab_a = f" [{team_a[0].ability}]" if team_a[0].ability else "" + ab_b = f" [{team_b[0].ability}]" if team_b[0].ability else "" + print(f" {label_a}({agent_a.__class__.__name__}) VS {label_b}({agent_b.__class__.__name__})") skills_a = [s.name for s in team_a[0].skills] skills_b = [s.name for s in team_b[0].skills] - print(f" 先锋: {team_a[0].name}[{', '.join(skills_a)}]") - print(f" vs {team_b[0].name}[{', '.join(skills_b)}]") + print(f" 先锋: {team_a[0].name}{ab_a}[{', '.join(skills_a)}]") + print(f" vs {team_b[0].name}{ab_b}[{', '.join(skills_b)}]") print(SEP) - agent_a = MCTSAgent("a", label_a, iterations=_MCTS_ITERS_BATTLE) - agent_b = MCTSAgent("b", label_b, iterations=_MCTS_ITERS_BATTLE) - history = [] - winner = None for _ in range(BattleEngine.MAX_TURNS): winner = engine.check_winner() if winner: break - if verbose: + if verbose and not has_human: _print_field(state, label_a, label_b) snap = state.deep_copy() action_a = agent_a.choose_action(engine) @@ -147,11 +178,9 @@ def run_battle( if not winner: winner = engine.check_winner() - # 记录经验并保存 - agent_a.experience_db.record_game(history, winner) - agent_b.experience_db.record_game(history, winner) - agent_a.save() - agent_b.save() + # 记录经验 — 通过统一协议调用 + agent_a.on_game_end(history, winner) + agent_b.on_game_end(history, winner) if verbose: tag = f"{label_a} 赢！" if winner == "a" else (f"{label_b} 赢！" if winner == "b" else "平局/超时") @@ -232,7 +261,16 @@ def run_batch( # ============================================================ def _menu_battle() -> None: print(f"\n{SEP}") - print(" 开始对战 — 选择 A 队") + print(" 选择对战模式") + print(f" 1. AI vs AI（MCTS自战）") + print(f" 2. 人 vs AI（你控制A队）") + print(f" 3. 人对人（你选A队，对方键盘输入B队）") + print(f" 4. LLM vs AI（大模型对战MCTS）") + print(f" 5. LLM vs 人类（大模型对战你）") + raw = input(" 选择 [1-5]（默认 2）：").strip() + mode = int(raw) if raw.isdigit() and 1 <= int(raw) <= 5 else 2 + + print(f"\n 开始对战 — 选择 A 队") name_a = _pick_team("A 队序号") if name_a is None: return @@ -245,7 +283,48 @@ def _menu_battle() -> None: team_a = build_team(name_a) team_b = build_team(name_b) - run_battle(team_a, team_b, name_a, name_b, verbose=True) + + # ── 根据模式创建 Agent ─────────────────────── + from sim.human_agent import HumanAgent + + agent_a = None + agent_b = None + + if mode == 1: + # AI vs AI — 默认行为，不传 agent + pass + elif mode == 2: + # 人 vs AI + agent_a = HumanAgent("a", name_a) + elif mode == 3: + # 人对人 + agent_a = HumanAgent("a", name_a) + agent_b = HumanAgent("b", name_b) + elif mode == 4: + # LLM vs AI + try: + from sim.llm_agent import LLMAgent + agent_a = LLMAgent("a", name_a) + except EnvironmentError as e: + print(f"\n [!] {e}") + return + elif mode == 5: + # LLM vs 人类 + try: + from sim.llm_agent import LLMAgent + agent_a = LLMAgent("a", name_a) + except EnvironmentError as e: + print(f"\n [!] {e}") + return + agent_b = HumanAgent("b", name_b) + + run_battle( + team_a, team_b, + label_a=name_a, label_b=name_b, + verbose=True, + agent_a=agent_a, + agent_b=agent_b, + ) # ============================================================ @@ -600,11 +679,35 @@ def _menu_batch() -> None: raw = input(" 模拟场数 N（默认 100）：").strip() n = int(raw) if raw.isdigit() and int(raw) > 0 else 100 - run_batch( - lambda: build_team(name_a), - lambda: build_team(name_b), - name_a, name_b, n, - ) + # 选择模拟模式 + print(f"\n 选择模拟模式：") + print(f" 1. 单线程批量模拟（兼容旧版）") + print(f" 2. 并发批量模拟+经验合并（推荐，更快）") + mode = input(" 选择 [1-2]（默认 2）：").strip() + + if mode == "1": + # 旧版单线程模式 + run_batch( + lambda: build_team(name_a), + lambda: build_team(name_b), + name_a, name_b, n, + ) + else: + # 并发模拟模式 + import multiprocessing as mp + workers = input(" 工作进程数（默认 CPU核心数）：").strip() + if not workers.isdigit() or int(workers) < 1: + workers = mp.cpu_count() + else: + workers = min(int(workers), n) + + from sim.batch_concurrent import run_concurrent_batch_with_experience + run_concurrent_batch_with_experience( + team_a_name=name_a, + team_b_name=name_b, + n=n, + workers=workers, + ) # ============================================================ @@ -777,6 +880,131 @@ def _menu_import_image() -> None: print(f"\n 队伍「{final_name}」{verb}！可在对战菜单中选用。") +# ============================================================ +# 菜单：6. LLM 辅助 +# ============================================================ +def _menu_llm_assist() -> None: + """LLM 辅助功能 — 队伍生成、策略生成等。""" + print(f"\n{SEP}") + print(" LLM 辅助") + print(SEP) + print(" 1. AI 组队（大模型根据精灵库和已有经验设计新阵容）") + print(" 2. 生成策略文件（为现有队伍创建 MCTS 策略配置）") + print(" 3. 查看历史经验（查看 LLM 对战后的经验文档）") + print(SEP) + + try: + choice = input(" 选择 [1-3]（0 取消）：").strip() + except (EOFError, KeyboardInterrupt): + return + + if choice == "0" or not choice: + return + + # ── AI 组队 ─────────────────────── + if choice == "1": + from sim.llm_team_generator import generate_team_with_llm, save_generated_team + + theme = input(" 请输入队伍主题/风格（留空则自由发挥）：").strip() + print(f"\n 正在调用大模型设计阵容...") + result = generate_team_with_llm(theme if theme else None) + if result is None: + print(" [!] LLM 生成失败") + return + + print(f"\n{SEP}") + print(f" AI 设计结果：队伍「{result['team_name']}」") + theme_note = result.get("theme", "") + if theme_note: + print(f" 风格: {theme_note}") + strategy_notes = result.get("strategy_notes", "") + if strategy_notes: + print(f" 战术说明: {strategy_notes}") + for i, m in enumerate(result["members"], 1): + skills_str = ", ".join(s for s in m.get("skills", []) if s) + print(f" {i}. {m['pokemon']:<12} 技能：{skills_str}") + + confirm = input("\n 保存到队伍列表？(Y/n)：").strip().lower() + if confirm != "n": + save_generated_team(result) + + # ── 生成策略文件 ─────────────────────── + elif choice == "2": + from sim.llm_team_generator import generate_strategy_with_llm, save_generated_strategy + + _print_roster("队伍列表") + teams = list_teams() + print(f"\n 选择队伍（0 取消）：", end="") + raw = input().strip() + if raw == "0" or not raw: + return + if not raw.isdigit() or not (1 <= int(raw) <= len(teams)): + print(" [!] 无效序号") + return + + team_name = teams[int(raw) - 1]["name"] + print(f"\n 正在为「{team_name}」生成策略文件...") + strategy = generate_strategy_with_llm(team_name) + if strategy is None: + print(" [!] LLM 生成策略失败") + return + + # 显示生成的策略 + print(f"\n{SEP}") + print(f" 策略配置：") + for key, val in strategy.items(): + if isinstance(val, list): + print(f" {key}: {', '.join(str(v) for v in val)}") + elif isinstance(val, dict): + print(f" {key}:") + for k2, v2 in val.items(): + print(f" {k2}: {v2}") + else: + print(f" {key}: {val}") + + confirm = input("\n 保存策略文件？(Y/n)：").strip().lower() + if confirm != "n": + save_generated_strategy(team_name, strategy) + + # ── 查看历史经验 ─────────────────────── + elif choice == "3": + from sim.llm_agent import _load_experience + import json as _json + + _print_roster("队伍列表") + teams = list_teams() + print(f"\n 选择队伍（0 取消）：", end="") + raw = input().strip() + if raw == "0" or not raw: + return + if not raw.isdigit() or not (1 <= int(raw) <= len(teams)): + print(" [!] 无效序号") + return + + team_name = teams[int(raw) - 1]["name"] + experiences = _load_experience(team_name) + if not experiences: + print(f"\n 「{team_name}」暂无历史经验文档") + return + + print(f"\n{SEP}") + print(f" {team_name} — 历史对战经验（共 {len(experiences)} 条）：") + for i, exp in enumerate(reversed(experiences), 1): + result = exp.get("result", "未知") + summary = exp.get("summary", "")[:200] + timestamp = exp.get("timestamp", "") + lessons = exp.get("lessons", []) + print(f"\n [{i}] {timestamp}") + print(f" 结果: {result} | 回合数: {exp.get('turns', '?')}") + if summary: + print(f" 总结: {summary}") + for lesson in (lessons if isinstance(lessons, list) else [str(lessons)]): + print(f" 教训: {lesson[:200]}") + + else: + print(" 无效选择") + + # ============================================================ # 主菜单 # ============================================================ @@ -796,16 +1024,17 @@ def main() -> None: tag = "[预设]" if t.get("preset") else "[自定]" print(f" {i:2}. {tag} {t['name']}") print(SEP) - print(" 1. 开始对战 （从列表选两支队伍）") + print(" 1. 开始对战 （选择模式：AI/AI、人/AI、人对人、LLM/AI等）") print(" 2. 新建队伍 （交互组队并保存）") print(" 3. 管理队伍 （查看 / 删除 / 重命名）") - print(" 4. 批量模拟 （选两支队伍跑 N 场）") + print(" 4. 批量模拟 （选两支队伍跑 N 场，支持并发加速）") print(" 5. 从图片导入队伍 （识别标准组队分享图）") + print(" 6. LLM 辅助 （AI组队、生成策略、查看经验文档）") print(" 0. 返回") print(SEP) try: - choice = input(" 选择 [0-5]: ").strip() + choice = input(" 选择 [0-6]: ").strip() except (EOFError, KeyboardInterrupt): print("\n 再见！") break @@ -822,8 +1051,10 @@ def main() -> None: _menu_batch() elif choice == "5": _menu_import_image() + elif choice == "6": + _menu_llm_assist() else: - print(" 无效选择，请输入 0-5") + print(" 无效选择，请输入 0-6") continue try: diff --git a/data/llm_config.yaml b/data/llm_config.yaml new file mode 100644 index 0000000..29c1e1b --- /dev/null +++ b/data/llm_config.yaml @@ -0,0 +1,15 @@ +# 大模型配置 — LLM Agent 使用此文件连接外部 API +# +# endpoint: OpenAI 兼容的 API 端点（不含 /v1/chat/completions） +# model: 模型名称 +# api_key: API key +# temperature: 温度参数（越低越确定，0.2~0.7 适合对战决策） +# timeout: HTTP 请求超时秒数（生成队伍/策略时 prompt 较长，需更大值） +# +# 经验文档存储在 data/llm_experience/.json + +endpoint: "http://xxxxx:1234/v1" +model: "qwopus3.6-27b-v1-preview" +api_key: "localapi" +temperature: 0.3 +timeout: 120 diff --git a/db_manage.py b/db_manage.py new file mode 100644 index 0000000..9a967cd --- /dev/null +++ b/db_manage.py @@ -0,0 +1,169 @@ +""" +数据库管理工具 + +提供数据库初始化、迁移、查看统计等命令行功能。 + +用法： + python db_manage.py init # 初始化数据库（创建表） + python db_manage.py migrate # 从JSON迁移数据到数据库 + python db_manage.py stats # 查看数据统计 + python db_manage.py battles N # 查看最近N场对战记录 + python db_match A B # 查询A队对B队的胜率 +""" + +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from sim.data_store import ( + get_config, + init_db, + migrate_experience_from_json, + migrate_pokemon_from_json, + DataStore, +) + + +def cmd_init(): + """初始化数据库""" + config = get_config() + print(f"\n [配置] 数据库: {config['name']}") + print(f" [配置] 地址: {config['host']}:{config['port']}") + + success = init_db(config) + if success: + print(" ✓ 数据库初始化成功！") + else: + print(" ✗ 数据库初始化失败！") + + +def cmd_migrate(): + """从JSON迁移数据到数据库""" + config = get_config() + + print("\n [迁移] 开始从JSON文件迁移数据...") + + # 迁移经验数据 + if os.path.exists(os.path.join(os.path.dirname(__file__), "data", "experience")): + migrate_experience_from_json(config) + else: + print(" [跳过] 经验目录不存在") + + # 迁移精灵数据 + pokemon_path = os.path.join(os.path.dirname(__file__), "data", "sprites.json") + if os.path.exists(pokemon_path): + migrate_pokemon_from_json(config) + else: + print(" [跳过] 精灵JSON文件不存在") + + print("\n ✓ 迁移完成！") + + +def cmd_stats(): + """查看数据统计""" + config = get_config() + store = DataStore(config) + + if not store.use_database(): + print("\n [提示] 当前使用JSON后端，数据库统计不可用") + return + + from sim.data_store import get_engine, ExperienceRecord, BattleRecord, PokemonRecord + from sqlalchemy.orm import sessionmaker + + engine = get_engine(config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + exp_count = session.query(ExperienceRecord).count() + battle_count = session.query(BattleRecord).count() + pokemon_count = session.query(PokemonRecord).count() if PokemonRecord else 0 + + print(f"\n [数据统计]") + print(f" 经验记录: {exp_count}") + print(f" 对战记录: {battle_count}") + if pokemon_count: + print(f" 精灵数据: {pokemon_count}") + except Exception as e: + print(f"\n [!] 统计查询失败: {e}") + finally: + session.close() + + +def cmd_battles(limit=10): + """查看最近的对战记录""" + config = get_config() + store = DataStore(config) + + if not store.use_database(): + print("\n [提示] 当前使用JSON后端，对战统计不可用") + return + + records = store.get_battle_stats(limit=limit) + + if not records: + print("\n [空] 暂无对战记录") + return + + print(f"\n [最近 {len(records)} 场对战]") + print(f" {'A队':<12} {'B队':<12} {'胜者':<6} {'回合':>4} {'耗时':>8}") + print(" " + "-" * 50) + + for r in records: + winner_str = f"{r['team_a']}✓" if r["winner"] == "a" else (f"{r['team_b']}✓" if r["winner"] == "b" else "平局") + print(f" {r['team_a']:<12} {r['team_b']:<12} {winner_str:<6} {r['turns']:>4} {r['elapsed_ms']/1000:.1f}s") + + +def cmd_match(team_a: str, team_b: str): + """查询A队对B队的胜率""" + config = get_config() + store = DataStore(config) + + if not store.use_database(): + print("\n [提示] 当前使用JSON后端，胜率统计不可用") + return + + win_rate = store.get_win_rate(team_a, team_b) + + if win_rate is None: + print(f"\n [空] {team_a} vs {team_b}: 暂无对战数据") + else: + print(f"\n [{team_a} vs {team_b}] 胜率: {win_rate*100:.1f}%") + + +def main(): + if len(sys.argv) < 2: + print("\n 数据库管理工具") + print(" 用法: python db_manage.py <命令> [参数]") + print("\n 命令:") + print(" init - 初始化数据库（创建表）") + print(" migrate - 从JSON迁移数据到数据库") + print(" stats - 查看数据统计") + print(" battles N - 查看最近N场对战记录") + print(" match A B - 查询A队对B队的胜率") + return + + cmd = sys.argv[1].lower() + + if cmd == "init": + cmd_init() + elif cmd == "migrate": + cmd_migrate() + elif cmd == "stats": + cmd_stats() + elif cmd == "battles": + limit = int(sys.argv[2]) if len(sys.argv) > 2 else 10 + cmd_battles(limit) + elif cmd == "match": + if len(sys.argv) < 4: + print("\n [!] 用法: python db_manage.py match A队名 B队名") + return + cmd_match(sys.argv[2], sys.argv[3]) + else: + print(f"\n [!] 未知命令: {cmd}") + + +if __name__ == "__main__": + main() diff --git a/sim/agent_base.py b/sim/agent_base.py new file mode 100644 index 0000000..613d307 --- /dev/null +++ b/sim/agent_base.py @@ -0,0 +1,36 @@ +""" +Agent Protocol — 所有战斗智能体的统一接口 + +定义了 choose_action / on_game_end 两个核心方法， +以及 show_team_status 标记（用于区分人类玩家和 AI）。 +""" + +from typing import List, Optional, Tuple, Protocol + +# Action = (技能索引,) | (-1,)汇合聚能 | (-2, 精灵索引)换人 +Action = Tuple[int, ...] +GameHistory = List[Tuple] # (BattleState深拷贝, Action_a, Action_b) + + +class AgentProtocol(Protocol): + """ + 战斗智能体协议。 + + Attributes + ---------- + team : str + "a" 或 "b"，标识所属队伍。 + show_team_status : bool + True = 人类玩家（自行打印状态）；False = AI（由引擎统一打印）。 + """ + + team: str + show_team_status: bool + + def choose_action(self, engine) -> Action: + """根据当前引擎状态选择本回合动作。""" + ... + + def on_game_end(self, history: GameHistory, winner: Optional[str]) -> None: + """战斗结束回调 — 记录经验或打印结果等。""" + ... diff --git a/sim/batch_concurrent.py b/sim/batch_concurrent.py new file mode 100644 index 0000000..4b199bd --- /dev/null +++ b/sim/batch_concurrent.py @@ -0,0 +1,481 @@ +""" +并发批量战斗模拟器 + +使用多进程并行执行MCTS对战，加速训练过程。 +每个进程独立运行完整的对战流程（加载数据→MCTS搜索→记录经验）， +最后合并结果和经验数据库。 + +架构设计： + - 主进程负责任务分发和进度跟踪 + - 工作进程各自独立运行对战，避免GIL锁争用 + - 经验数据在进程结束后合并到JSON文件 +""" + +import os +import sys +import time +import json +import multiprocessing as mp +from typing import List, Dict, Optional, Callable, Any + +# 确保模块路径正确 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from sim.pokemon_db import load_pokemon_db +from sim.skill_db import load_skills +from sim.team_roster import build_team +from sim.battle_state import BattleState +from sim.battle_engine import BattleEngine +from sim.mcts_agent import MCTSAgent +from sim.experience_db import ExperienceDB, ActionStats + +# 批量模拟MCTS迭代次数（可调整：20快速 / 50标准 / 100强力） +_MCTS_ITERS_BATCH = 20 + +# 最大回合数限制 +_MAX_TURNS = BattleEngine.MAX_TURNS + + +def _run_single_battle(args: tuple) -> dict: + """ + 在子进程中运行单场对战。 + + Parameters + ---------- + args : (team_a_name, team_b_name, mcts_iters) + - team_a_name: A队名称（用于加载队伍和策略） + - team_b_name: B队名称 + - mcts_iters: MCTS迭代次数 + + Returns + ------- + dict : { + "winner": "a" | "b" | None, + "turns": int, + "exp_a": ExperienceDB实例（序列化后）, + "exp_b": ExperienceDB实例 + } + """ + team_a_name, team_b_name, mcts_iters = args + + # 在子进程中加载数据 + load_pokemon_db() + load_skills() + + # 构建队伍 + team_a = build_team(team_a_name) + team_b = build_team(team_b_name) + + # 创建MCTS Agent + from sim.experience_db import ExperienceDB + exp_a = ExperienceDB() # 空经验库 + exp_b = ExperienceDB() + + agent_a = MCTSAgent("a", team_a_name, iterations=mcts_iters, load_exp=False) + agent_b = MCTSAgent("b", team_b_name, iterations=mcts_iters, load_exp=False) + # 替换经验库 — 必须同时替换agent和_search的引用，否则脱节 + agent_a.experience_db = exp_a + agent_a._search.experience_db = exp_a + agent_b.experience_db = exp_b + agent_b._search.experience_db = exp_b + + # 运行对战 + state = BattleState(team_a=team_a, team_b=team_b) + engine = BattleEngine(state, verbose=False) + history = [] + winner = None + + for _ in range(_MAX_TURNS): + winner = engine.check_winner() + if winner: + break + snap = state.deep_copy() + action_a = agent_a.choose_action(engine) + action_b = agent_b.choose_action(engine) + history.append((snap, action_a, action_b)) + engine.execute_turn(action_a, action_b) + + if not winner: + winner = engine.check_winner() + + # 记录经验 + agent_a.experience_db.record_game(history, winner) + agent_b.experience_db.record_game(history, winner) + + return { + "winner": winner, + "turns": state.turn, + "exp_a": agent_a.experience_db._db, + "exp_b": agent_b.experience_db._db, + "total_games_a": agent_a.experience_db.total_games, + "total_games_b": agent_b.experience_db.total_games, + } + + +def _merge_experience_dbs( + exp_data_list: list, # list of (team, exp_dict, total_games) +) -> ExperienceDB: + """ + 合并多个经验数据库。 + + Parameters + ---------- + exp_data_list : [(team, exp_dict, total_games), ...] + - team: "a" or "b" + - exp_dict: _db dict from ExperienceDB (values are ActionStats objects) + - total_games: int + + Returns + ------- + ExperienceDB : 合并后的经验数据库 + """ + merged_db = {"a": {}, "b": {}} + total_games = 0 + + for team, exp_dict, games in exp_data_list: + if team not in merged_db: + continue + total_games += games + state_dict = merged_db[team] + for sk, actions in exp_dict.items(): + if sk not in state_dict: + state_dict[sk] = {} + for ak, stats_obj in actions.items(): + if ak not in state_dict[sk]: + state_dict[sk][ak] = ActionStats() + # 兼容两种格式：ActionStats对象和dict + if isinstance(stats_obj, ActionStats): + merged_stats = state_dict[sk][ak] + merged_stats.wins += stats_obj.wins + merged_stats.total += stats_obj.total + else: + # dict格式 {"w": ..., "n": ...} + state_dict[sk][ak].wins += stats_obj.get("w", 0) + state_dict[sk][ak].total += stats_obj.get("n", 0) + + merged_exp = ExperienceDB() + merged_exp._db = merged_db + merged_exp.total_games = total_games + + return merged_exp + + +def _serialize_exp_db(db_dict): + """将ActionStats对象转为dict以便JSON序列化""" + serialized = {} + for team, states in db_dict.items(): + serialized[team] = {} + for sk, actions in states.items(): + serialized[sk] = {} + for ak, stats in actions.items(): + serialized[sk][ak] = {"w": stats.wins, "n": stats.total} + return serialized + + +def run_concurrent_batch_with_experience( + team_a_name: str, + team_b_name: str, + n: int, + workers: int = None, + mcts_iters: int = _MCTS_ITERS_BATCH, + exp_dir: str = None, +) -> dict: + """ + 并发批量模拟并合并经验数据库。 + + 使用multiprocessing.Pool替代ProcessPoolExecutor， + 避免Python 3.9的"dictionary changed size during iteration" bug。 + + Parameters + ---------- + team_a_name : A队名称 + team_b_name : B队名称 + n : 对战总场数 + workers : 工作进程数 + mcts_iters : MCTS迭代次数 + exp_dir : 经验数据库保存目录 + + Returns + ------- + dict : { + "results": {"a": int, "b": int, "draw": int}, + "total_turns": int, + "elapsed": float, + "exp_a_path": str, # A队经验保存路径 + "exp_b_path": str, # B队经验保存路径 + } + """ + if workers is None: + workers = min(mp.cpu_count(), n) + workers = max(1, workers) + + exp_dir = exp_dir or os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", "data", "experience" + ) + os.makedirs(exp_dir, exist_ok=True) + + print(f"\n [并发模拟+经验] {team_a_name} vs {team_b_name}") + print(f" [配置] 场数={n}, MCTS迭代={mcts_iters}, 工作进程={workers}") + + # 检查是否使用数据库后端 + try: + from sim.data_store import DataStore + store = DataStore() + use_db = store.db_enabled + except ImportError: + use_db = False + + if use_db: + print(f" [存储] MariaDB（增量保存）") + else: + print(f" [存储] JSON文件") + + results = {"a": 0, "b": 0, "draw": 0} + total_turns = 0 + t0 = time.time() + + # 准备任务列表 + tasks = [ + (team_a_name, team_b_name, mcts_iters) + for _ in range(n) + ] + + # 收集经验数据 — 按队伍分组 + all_exp_data_a = [] # list of exp_dict from agent_a + all_exp_data_b = [] # list of exp_dict from agent_b + total_games_sum_a = 0 + total_games_sum_b = 0 + + # 使用multiprocessing.Pool替代ProcessPoolExecutor + # Pool更稳定，不会出现"dictionary changed size during iteration" bug + with mp.Pool(processes=workers) as pool: + # imap_unordered返回结果不保证顺序，但能尽早处理完成的任务 + iterator = pool.imap_unordered(_run_single_battle, tasks) + + completed = 0 + for result in iterator: + winner = result["winner"] + results[winner or "draw"] += 1 + total_turns += result["turns"] + + # 收集经验数据 — 按队伍分别存储 + all_exp_data_a.append(result["exp_a"]) + all_exp_data_b.append(result["exp_b"]) + total_games_sum_a += result["total_games_a"] + total_games_sum_b += result["total_games_b"] + + completed += 1 + elapsed_so_far = time.time() - t0 + rate = elapsed_so_far / completed + eta = rate * (n - completed) + bar_filled = int(20 * completed / n) + bar = "#" * bar_filled + "." * (20 - bar_filled) + + print( + f"\r [{bar}] {completed:4}/{n} " + f"A:{results['a']} B:{results['b']} 平:{results['draw']} " + f"ETA:{eta:.0f}s ", + end="", flush=True, + ) + + elapsed = time.time() - t0 + + # 合并经验数据库 — 按队伍分别合并 + print("\n [合并经验] 正在合并各进程的经验数据...") + + # A队：只合并每个exp_a中team='a'的部分，B队同理 + merged_db_a = {"a": {}, "b": {}} + for exp_dict in all_exp_data_a: + team_data = exp_dict.get("a", {}) + for sk, actions in team_data.items(): + if sk not in merged_db_a["a"]: + merged_db_a["a"][sk] = {} + for ak, stats_obj in actions.items(): + if isinstance(stats_obj, ActionStats): + if ak not in merged_db_a["a"][sk]: + merged_db_a["a"][sk][ak] = ActionStats() + merged_db_a["a"][sk][ak].wins += stats_obj.wins + merged_db_a["a"][sk][ak].total += stats_obj.total + + merged_db_b = {"a": {}, "b": {}} + for exp_dict in all_exp_data_b: + team_data = exp_dict.get("b", {}) + for sk, actions in team_data.items(): + if sk not in merged_db_b["b"]: + merged_db_b["b"][sk] = {} + for ak, stats_obj in actions.items(): + if isinstance(stats_obj, ActionStats): + if ak not in merged_db_b["b"][sk]: + merged_db_b["b"][sk][ak] = ActionStats() + merged_db_b["b"][sk][ak].wins += stats_obj.wins + merged_db_b["b"][sk][ak].total += stats_obj.total + + # 创建合并后的ExperienceDB实例 + merged_a = ExperienceDB() + merged_a._db = merged_db_a + merged_a.total_games = total_games_sum_a + + merged_b = ExperienceDB() + merged_b._db = merged_db_b + merged_b.total_games = total_games_sum_b + + # 保存经验数据库 — 根据配置选择后端 + exp_a_path = os.path.join(exp_dir, f"{team_a_name}.json") + exp_b_path = os.path.join(exp_dir, f"{team_b_name}.json") + + if use_db: + # 保存到数据库（增量保存，只更新被修改过的记录） + print(" [保存] 正在写入MariaDB...") + for team in ("a", "b"): + merged_exp = merged_a if team == "a" else merged_b + team_name = team_a_name if team == "a" else team_b_name + + # 遍历所有经验记录并保存到数据库 + for sk, actions in merged_exp._db.get(team, {}).items(): + for ak, stats_obj in actions.items(): + store.update_experience( + team=team, + state_key=sk, + action_key=ak, + wins=stats_obj.wins, + total=stats_obj.total + ) + + # 记录对战结果到数据库 + store.record_battle( + team_a_name=team_a_name, + team_b_name=team_b_name, + winner="a" if results["a"] > results["b"] else ("b" if results["b"] > results["a"] else None), + turns=total_turns // n, + elapsed_ms=int(elapsed * 1000), + mcts_iters=mcts_iters + ) + + print(f" [保存] A队经验 → MariaDB") + print(f" [保存] B队经验 → MariaDB") + else: + # 保存到JSON文件（原有逻辑） + with open(exp_a_path, "w", encoding="utf-8") as f: + json.dump(_serialize_exp_db(merged_a._db), f, ensure_ascii=False) + + with open(exp_b_path, "w", encoding="utf-8") as f: + json.dump(_serialize_exp_db(merged_b._db), f, ensure_ascii=False) + + print(f" [保存] A队经验 → {exp_a_path}") + print(f" [保存] B队经验 → {exp_b_path}") + + print(f"\n{'=' * 56}") + print(f" 并发批量模拟结果（{n} 场，MCTS×{mcts_iters}）") + print(f" {team_a_name} 胜: {results['a']:4} 场 ({results['a']/n*100:.1f}%)") + print(f" {team_b_name} 胜: {results['b']:4} 场 ({results['b']/n*100:.1f}%)") + print(f" 平局: {results['draw']:4} 场 ({results['draw']/n*100:.1f}%)") + print(f" 平均回合数: {total_turns/n:.1f}") + print(f" 总耗时: {elapsed:.2f}s ({elapsed/n*1000:.1f}ms/场)") + print("=" * 56) + + return { + "results": results, + "total_turns": total_turns, + "elapsed": elapsed, + "exp_a_path": exp_a_path, + "exp_b_path": exp_b_path, + } + + +def run_concurrent_batch( + team_a_name: str, + team_b_name: str, + n: int, + workers: int = None, + mcts_iters: int = _MCTS_ITERS_BATCH, +) -> dict: + """ + 并发批量模拟（不保存经验数据，仅统计结果）。 + + Parameters + ---------- + team_a_name : A队名称 + team_b_name : B队名称 + n : 对战总场数 + workers : 工作进程数 + mcts_iters : MCTS迭代次数 + + Returns + ------- + dict : { + "results": {"a": int, "b": int, "draw": int}, + "total_turns": int, + "elapsed": float, + } + """ + if workers is None: + workers = min(mp.cpu_count(), n) + workers = max(1, workers) + + print(f"\n [并发模拟] {team_a_name} vs {team_b_name}") + print(f" [配置] 场数={n}, MCTS迭代={mcts_iters}, 工作进程={workers}") + + results = {"a": 0, "b": 0, "draw": 0} + total_turns = 0 + t0 = time.time() + + # 准备任务列表 + tasks = [ + (team_a_name, team_b_name, mcts_iters) + for _ in range(n) + ] + + with mp.Pool(processes=workers) as pool: + iterator = pool.imap_unordered(_run_single_battle, tasks) + + completed = 0 + for result in iterator: + winner = result["winner"] + results[winner or "draw"] += 1 + total_turns += result["turns"] + + completed += 1 + elapsed_so_far = time.time() - t0 + rate = elapsed_so_far / completed + eta = rate * (n - completed) + bar_filled = int(20 * completed / n) + bar = "#" * bar_filled + "." * (20 - bar_filled) + + print( + f"\r [{bar}] {completed:4}/{n} " + f"A:{results['a']} B:{results['b']} 平:{results['draw']} " + f"ETA:{eta:.0f}s ", + end="", flush=True, + ) + + elapsed = time.time() - t0 + print(f"\n{'=' * 56}") + print(f" 并发批量模拟结果（{n} 场，MCTS×{mcts_iters}）") + print(f" {team_a_name} 胜: {results['a']:4} 场 ({results['a']/n*100:.1f}%)") + print(f" {team_b_name} 胜: {results['b']:4} 场 ({results['b']/n*100:.1f}%)") + print(f" 平局: {results['draw']:4} 场 ({results['draw']/n*100:.1f}%)") + print(f" 平均回合数: {total_turns/n:.1f}") + print(f" 总耗时: {elapsed:.2f}s ({elapsed/n*1000:.1f}ms/场)") + print("=" * 56) + + return { + "results": results, + "total_turns": total_turns, + "elapsed": elapsed, + } + + +if __name__ == "__main__": + # 测试并发模拟 + load_pokemon_db() + load_skills() + + print("测试并发批量模拟：预设毒队 vs 狼王队（10场）") + run_concurrent_batch_with_experience( + team_a_name="预设毒队", + team_b_name="狼王队", + n=10, + workers=4, # 使用4个进程 + mcts_iters=_MCTS_ITERS_BATCH, + ) diff --git a/sim/battle_engine.py b/sim/battle_engine.py index 51bb5a4..fe429a8 100644 --- a/sim/battle_engine.py +++ b/sim/battle_engine.py @@ -47,6 +47,8 @@ def __init__(self, state: BattleState, verbose: bool = False): self.state = state self.verbose = verbose self.log: List[str] = [] + self.label_a: str = "A队" # A队显示标签 + self.label_b: str = "B队" # B队显示标签 # 为初始出战精灵设置入场回合（仅回合1且未设置时） _ability_hooks.on_battle_start(state, self) @@ -240,9 +242,15 @@ def _execute_action( # ---- 换人 ---- if action[0] == -2: - self._apply_switch(team, action[1]) - new_p = team_list[action[1]] - self._log(f"[{team.upper()}] {current.name} 换人 → {new_p.name} (HP:{new_p.current_hp}/{new_p.hp} 能量:{new_p.energy})") + target_idx = action[1] + current_idx = self.state.get_current_idx(team) + if target_idx == current_idx: + self._log(f"[{team.upper()}] {current.name} 已在前台，无效换人") + return + self._apply_switch(team, target_idx) + new_p = team_list[target_idx] + ab_tag = f" [{new_p.ability}]" if new_p.ability else "" + self._log(f"[{team.upper()}] {current.name} 换人 → {new_p.name}{ab_tag} (HP:{new_p.current_hp}/{new_p.hp} 能量:{new_p.energy})") return # ---- 汇合聚能 ---- @@ -533,8 +541,12 @@ def _apply_self_recovery(self, user: Pokemon, skill: Skill) -> None: user.gain_energy(skill.self_heal_energy) def _apply_switch(self, team: str, target_idx: int) -> None: - """换人（触发换出/换入特性）""" + """换人（触发换出/换入特性，切出时清除非印记负面效果）""" old_idx = self.state.get_current_idx(team) + # 切出精灵：清除非印记状态（中毒、灼烧、冻结、寄生） + outgoing = self.state.get_team(team)[old_idx] + if not outgoing.is_fainted: + outgoing.clear_debuffs() _ability_hooks.on_switch_out(self.state, self, team, old_idx, target_idx) self.state.set_current_idx(team, target_idx) _ability_hooks.on_switch_in(self.state, self, team, target_idx) @@ -597,9 +609,15 @@ def _turn_end_effects(self) -> None: self._log(f" 天气 {self.state.weather.value} 消散了") self.state.weather = Weather.NONE - # --- 2-5. 异常状态处理（所有精灵，包括后备） --- - all_pokemon = self.state.team_a + self.state.team_b - for p in all_pokemon: + # --- 2-5. 异常状态处理（仅场上精灵，非印记效果切出即清除） --- + for team_key in ("team_a", "team_b"): + team_list = getattr(self.state, team_key) + # --- 2-5. 异常状态处理（仅场上精灵，非印记效果切出即清除） --- + for team_key in ("team_a", "team_b"): + team_list = getattr(self.state, team_key) + current_idx = self.state.get_current_idx("a" if team_key == "team_a" else "b") + p = team_list[current_idx] + if p.is_fainted: continue @@ -640,7 +658,6 @@ def _turn_end_effects(self) -> None: # 寄生者在场则回复 parasite_owner = self._find_pokemon_by_name(p.parasited_by) if parasite_owner and not parasite_owner.is_fainted: - # 检查寄生者是否是当前出战精灵 is_on_field = self._is_on_field(parasite_owner) if is_on_field: healed = parasite_owner.heal(actual) @@ -657,8 +674,8 @@ def _turn_end_effects(self) -> None: p.current_hp = 0 p.status = StatusType.FAINTED - # --- 6. 冷却递减 --- - for p in all_pokemon: + # --- 6. 冷却递减（所有精灵） --- + for p in self.state.team_a + self.state.team_b: expired = [] for k, v in p.cooldowns.items(): if v > 0: @@ -667,8 +684,9 @@ def _turn_end_effects(self) -> None: expired.append(k) for k in expired: del p.cooldowns[k] - for k in expired: - del p.cooldowns[k] + # 清理已过期但未被删除的（避免重复删除） + for k in list(expired): + p.cooldowns.pop(k, None) # --- 7. 回合结束特性效果（蚀刻转化 / 特殊清洁场景 / 印记伤害） --- _ability_hooks.on_turn_end(self.state, self) diff --git a/sim/data_store.py b/sim/data_store.py new file mode 100644 index 0000000..8b64511 --- /dev/null +++ b/sim/data_store.py @@ -0,0 +1,799 @@ +""" +数据存储层 - 统一接口，支持JSON文件和MariaDB两种后端 + +架构设计： + - DataStore类提供统一的读写接口 + - 根据.env配置自动选择后端 + - 首次启用数据库时自动从JSON迁移数据 +""" + +import os +import json +from typing import Dict, List, Optional, Tuple, Any +from dataclasses import dataclass, field +from datetime import datetime + +# SQLAlchemy相关 +from sqlalchemy import create_engine, Column, Integer, String, Float, Text, DateTime, Index +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker, Session +from sqlalchemy.pool import StaticPool + +Base = declarative_base() + + +# ============================================================ +# 配置加载 +# ============================================================ + +def _load_env(path: str = None) -> dict: + """从.env文件加载配置""" + config = {} + env_path = path or os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".env") + if not os.path.exists(env_path): + return config + with open(env_path, "r") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + config[key.strip()] = value.strip() + return config + + +def get_config() -> dict: + """获取数据存储配置""" + env = _load_env() + return { + "enabled": env.get("DB_ENABLED", "false").lower() == "true", + "host": env.get("DB_HOST", "127.0.0.1"), + "port": int(env.get("DB_PORT", 3306)), + "user": env.get("DB_USER", "root"), + "password": env.get("DB_PASSWORD", "root"), + "name": env.get("DB_NAME", "rocom_data"), + } + + +# ============================================================ +# 数据库模型 +# ============================================================ + +class ExperienceRecord(Base): + """ + MCTS经验数据表 + + 存储(state_key, action_key) → (wins, total)的映射关系 + """ + __tablename__ = "experience_records" + + id = Column(Integer, primary_key=True, autoincrement=True) + team = Column(String(1), nullable=False) # "a" or "b" + state_key = Column(Text, nullable=False) # 状态指纹（可能较长，用TEXT） + action_key = Column(String(64), nullable=False) # 动作标识 + wins = Column(Float, default=0.0) # 累计胜利值 + total = Column(Integer, default=0) # 总次数 + + __table_args__ = ( + Index("idx_exp_team_state", "team"), + Index("idx_exp_lookup", "team", "action_key"), + ) + + +class BattleRecord(Base): + """ + 对战记录表 + + 存储每场对战的基本信息，用于后续分析和LLM教练系统 + """ + __tablename__ = "battle_records" + + id = Column(Integer, primary_key=True, autoincrement=True) + team_a_name = Column(String(64), nullable=False) # A队名称 + team_b_name = Column(String(64), nullable=False) # B队名称 + winner = Column(String(1)) # "a", "b", or None + turns = Column(Integer, default=0) # 回合数 + elapsed_ms = Column(Integer, default=0) # 耗时（毫秒） + mcts_iters = Column(Integer, default=20) # MCTS迭代次数 + created_at = Column(DateTime, default=datetime.utcnow) + + __table_args__ = ( + Index("idx_battle_time", "created_at"), + ) + + +class PokemonRecord(Base): + """ + 精灵数据表（可选，未来迁移精灵数据时使用） + """ + __tablename__ = "pokemon_records" + + id = Column(Integer, primary_key=True, autoincrement=True) + name = Column(String(64), unique=True, nullable=False) # 精灵名称 + data_json = Column(Text, nullable=False) # 完整JSON数据 + created_at = Column(DateTime, default=datetime.utcnow) + + +# ============================================================ +# 数据库引擎管理 +# ============================================================ + +def get_engine(config: dict): + """创建数据库引擎""" + url = f"mysql+pymysql://{config['user']}:{config['password']}@{config['host']}:{config['port']}/{config['name']}?charset=utf8mb4" + return create_engine(url, pool_pre_ping=True) + + +def init_db(config: dict) -> bool: + """ + 初始化数据库（创建表和必要的数据库）。 + + Returns + ------- + bool : True=成功/已存在，False=失败 + """ + try: + # 先尝试连接MySQL创建数据库 + import pymysql + conn = pymysql.connect( + host=config["host"], + port=config["port"], + user=config["user"], + password=config["password"] + ) + with conn.cursor() as cursor: + cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{config['name']}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci") + conn.close() + + # 创建表 + engine = get_engine(config) + Base.metadata.create_all(engine) + return True + except Exception as e: + print(f" [!] 数据库初始化失败: {e}") + return False + + +def drop_db(config: dict) -> bool: + """ + 删除所有表（危险操作！用于重置）。 + + Returns + ------- + bool : True=成功，False=失败 + """ + try: + engine = get_engine(config) + Base.metadata.drop_all(engine) + return True + except Exception as e: + print(f" [!] 数据库删除失败: {e}") + return False + + +# ============================================================ +# JSON文件路径 +# ============================================================ + +def get_json_dir() -> str: + """获取JSON数据目录""" + return os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", "data", "experience" + ) + + +def get_pokemon_json_path() -> str: + """获取精灵JSON路径""" + return os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", "data", "sprites.json" + ) + + +# ============================================================ +# 数据迁移：JSON → 数据库 +# ============================================================ + +def migrate_experience_from_json(config: dict) -> bool: + """ + 将JSON经验数据迁移到数据库。 + + Returns + ------- + bool : True=成功，False=失败 + """ + json_dir = get_json_dir() + if not os.path.exists(json_dir): + print(" [!] JSON目录不存在，跳过迁移") + return False + + engine = get_engine(config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + total_records = 0 + batch_size = 500 + records_to_add = [] + seen_keys = {} # {(team, state_key, action_key): (wins, total)} for dedup within JSON + + try: + for filename in os.listdir(json_dir): + if not filename.endswith(".json"): + continue + + filepath = os.path.join(json_dir, filename) + with open(filepath, "r", encoding="utf-8") as f: + data = json.load(f) + + # 解析经验数据：{team: {state_key: {action_key: {"w": wins, "n": total}}}} + for team, states in data.items(): + if not isinstance(states, dict): + continue + if team not in ("a", "b"): + continue + for state_key, actions in states.items(): + if not isinstance(actions, dict): + continue + # 检查是否是标准格式（包含w和n键） + first_val = next(iter(actions.values()), None) + if isinstance(first_val, (int, float)): + # 旧格式：{state_key: {"w": wins, "n": total}} + wins = float(actions.get("w", 0)) + total_count = int(actions.get("n", 0)) + action_key = state_key.split("|")[-1] if "|" in state_key else "unknown" + + key = (team, state_key, action_key) + if key not in seen_keys: + seen_keys[key] = (wins, total_count) + records_to_add.append(ExperienceRecord( + team=team, + state_key=state_key, + action_key=action_key, + wins=wins, + total=total_count + )) + else: + sw, st = seen_keys[key] + seen_keys[key] = (sw + wins, st + total_count) + else: + # 新格式：{state_key: {action_key: {"w": wins, "n": total}}} + for action_key, stats in actions.items(): + if not isinstance(stats, dict): + continue + wins = float(stats.get("w", 0)) + total_count = int(stats.get("n", 0)) + + key = (team, state_key, action_key) + if key not in seen_keys: + seen_keys[key] = (wins, total_count) + records_to_add.append(ExperienceRecord( + team=team, + state_key=state_key, + action_key=action_key, + wins=wins, + total=total_count + )) + else: + sw, st = seen_keys[key] + seen_keys[key] = (sw + wins, st + total_count) + + print(f" [迁移] {filename} → 数据库") + + # 批量插入（使用INSERT ... ON DUPLICATE KEY UPDATE） + for i in range(0, len(records_to_add), batch_size): + batch = records_to_add[i:i+batch_size] + session.add_all(batch) + session.commit() + total_records += len(batch) + print(f" [进度] {min(i+len(batch), len(records_to_add))}/{len(records_to_add)}") + + print(f" [迁移完成] 共导入 {total_records} 条经验记录") + return True + except Exception as e: + session.rollback() + print(f" [!] 迁移失败: {e}") + import traceback + traceback.print_exc() + return False + finally: + session.close() + + +def migrate_pokemon_from_json(config: dict) -> bool: + """ + 将精灵JSON数据迁移到数据库。 + + Returns + ------- + bool : True=成功，False=失败 + """ + json_path = get_pokemon_json_path() + if not os.path.exists(json_path): + print(" [!] 精灵JSON文件不存在，跳过迁移") + return False + + engine = get_engine(config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + with open(json_path, "r", encoding="utf-8") as f: + data = json.load(f) + + total_records = 0 + # 精灵数据可能是list格式或dict格式，需要兼容 + if isinstance(data, list): + for item in data: + name = item.get("name", "unknown") + existing = session.query(PokemonRecord).filter_by(name=name).first() + if not existing: + record = PokemonRecord( + name=name, + data_json=json.dumps(item, ensure_ascii=False) + ) + session.add(record) + total_records += 1 + elif isinstance(data, dict): + for name, pdata in data.items(): + existing = session.query(PokemonRecord).filter_by(name=name).first() + if not existing: + record = PokemonRecord( + name=name, + data_json=json.dumps(pdata, ensure_ascii=False) + ) + session.add(record) + total_records += 1 + + session.commit() + print(f" [迁移完成] 精灵数据：{total_records} 条记录") + return True + except Exception as e: + session.rollback() + print(f" [!] 精灵迁移失败: {e}") + import traceback + traceback.print_exc() + return False + finally: + session.close() + + +# ============================================================ +# DataStore - 统一数据访问接口 +# ============================================================ + +class DataStore: + """ + 数据存储的统一接口。 + + 根据配置自动选择后端（数据库或JSON文件）。 + 支持经验数据的读写、对战记录的存储等。 + """ + + def __init__(self, config: dict = None): + self.config = config or get_config() + self._use_db = self.config.get("enabled", False) + + if self._use_db: + # 确保数据库已初始化 + init_db(self.config) + + @property + def db_enabled(self) -> bool: + """是否使用数据库后端（属性访问）""" + return self._use_db + + def use_database(self) -> bool: + """是否使用数据库后端""" + return self._use_db + + # ---------------------------------------------------------- + # 经验数据操作 + # ---------------------------------------------------------- + + def get_experience(self, team: str) -> Dict[str, Dict[str, Tuple[float, int]]]: + """ + 获取某队的全部经验数据。 + + Returns + ------- + dict : {state_key: {action_key: (wins, total)}} + """ + if self._use_db: + return self._get_experience_from_db(team) + else: + return self._get_experience_from_json(team) + + def save_experience( + self, + team: str, + data: Dict[str, Dict[str, Tuple[float, int]]] + ) -> None: + """ + 保存某队的经验数据。 + + Parameters + ---------- + team : "a" or "b" + data : {state_key: {action_key: (wins, total)}} + """ + if self._use_db: + self._save_experience_to_db(team, data) + else: + self._save_experience_to_json(team, data) + + def record_battle( + self, + team_a_name: str, + team_b_name: str, + winner: Optional[str], + turns: int, + elapsed_ms: int, + mcts_iters: int = 20 + ) -> None: + """ + 记录一场对战。 + + Parameters + ---------- + team_a_name : A队名称 + team_b_name : B队名称 + winner : "a", "b", or None + turns : 回合数 + elapsed_ms : 耗时（毫秒） + mcts_iters : MCTS迭代次数 + """ + if self._use_db: + self._record_battle_to_db( + team_a_name, team_b_name, winner, turns, elapsed_ms, mcts_iters + ) + + def get_battle_stats(self, limit: int = 10) -> List[dict]: + """ + 获取最近的对战统计。 + + Returns + ------- + list : [{team_a_name, team_b_name, winner, turns, ...}, ...] + """ + if self._use_db: + return self._get_battle_stats_from_db(limit) + else: + return [] # JSON后端暂不支持对战记录 + + def get_win_rate(self, team_a_name: str, team_b_name: str) -> Optional[float]: + """ + 获取两队之间的胜率（A队对B队的胜率）。 + + Returns + ------- + float : A队胜率 (0.0-1.0)，无数据时返回None + """ + if self._use_db: + return self._get_win_rate_from_db(team_a_name, team_b_name) + else: + return None + + def update_experience( + self, + team: str, + state_key: str, + action_key: str, + wins: float, + total: int + ) -> None: + """ + 更新单条经验记录（UPSERT）。 + + Parameters + ---------- + team : "a" or "b" + state_key : 状态指纹 + action_key : 动作标识 + wins : 累计胜利值 + total : 总次数 + """ + if self._use_db: + self._update_experience_in_db(team, state_key, action_key, wins, total) + + def get_all_experience(self) -> List[ExperienceRecord]: + """ + 获取所有经验记录。 + + Returns + ------- + list : [ExperienceRecord, ...] + """ + if self._use_db: + return self._get_all_experience_from_db() + else: + return [] + + def update_total_games(self, name: str, total: int) -> None: + """ + 更新某队的总局数。 + + Parameters + ---------- + name : 队伍名称（用于标识） + total : 总局数 + """ + # 暂不实现，因为当前数据库没有存储队伍名称对应的总局数的表 + + def get_total_games(self, name: str) -> int: + """ + 获取某队的总局数。 + + Returns + ------- + int : 总局数 + """ + # 暂不实现，返回0 + return 0 + + # ---------------------------------------------------------- + # 数据库后端实现 + # ---------------------------------------------------------- + + def _get_experience_from_db(self, team: str) -> Dict[str, Dict[str, Tuple[float, int]]]: + """从数据库读取经验数据""" + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + records = session.query(ExperienceRecord).filter_by(team=team).all() + result = {} + for r in records: + if r.state_key not in result: + result[r.state_key] = {} + result[r.state_key][r.action_key] = (float(r.wins), int(r.total)) + return result + finally: + session.close() + + def _save_experience_to_db(self, team: str, data: dict) -> None: + """ + 将经验数据保存到数据库。 + + 使用UPSERT逻辑：如果(state_key, action_key)已存在则累加，否则插入新记录。 + """ + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + for state_key, actions in data.items(): + for action_key, (wins, total) in actions.items(): + existing = session.query(ExperienceRecord).filter_by( + team=team, + state_key=state_key, + action_key=action_key + ).first() + + if existing: + existing.wins += float(wins) + existing.total += int(total) + else: + record = ExperienceRecord( + team=team, + state_key=state_key, + action_key=action_key, + wins=float(wins), + total=int(total) + ) + session.add(record) + + session.commit() + except Exception as e: + session.rollback() + print(f" [!] 经验数据保存失败: {e}") + finally: + session.close() + + def _record_battle_to_db(self, team_a_name, team_b_name, winner, turns, elapsed_ms, mcts_iters) -> None: + """记录对战到数据库""" + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + record = BattleRecord( + team_a_name=team_a_name, + team_b_name=team_b_name, + winner=winner, + turns=turns, + elapsed_ms=elapsed_ms, + mcts_iters=mcts_iters + ) + session.add(record) + session.commit() + except Exception as e: + session.rollback() + print(f" [!] 对战记录保存失败: {e}") + finally: + session.close() + + def _get_battle_stats_from_db(self, limit: int) -> List[dict]: + """从数据库获取对战统计""" + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + records = session.query(BattleRecord).order_by( + BattleRecord.created_at.desc() # type: ignore + ).limit(limit).all() + + return [ + { + "team_a": r.team_a_name, + "team_b": r.team_b_name, + "winner": r.winner, + "turns": r.turns, + "elapsed_ms": r.elapsed_ms, + "mcts_iters": r.mcts_iters, + "created_at": str(r.created_at), + } + for r in records + ] + finally: + session.close() + + def _get_win_rate_from_db(self, team_a_name: str, team_b_name: str) -> Optional[float]: + """ + 计算A队对B队的胜率。 + + 考虑两种情况： + - A是team_a时的胜率 + - B是team_b时的胜率（即A是team_b时B输的场次） + """ + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + # A作为team_a的情况 + records_a = session.query(BattleRecord).filter_by( + team_a_name=team_a_name, + team_b_name=team_b_name + ).all() + + # B作为team_a（即A作为team_b）的情况 + records_b = session.query(BattleRecord).filter_by( + team_a_name=team_b_name, + team_b_name=team_a_name + ).all() + + total_games = len(records_a) + len(records_b) + if total_games == 0: + return None + + a_wins = sum(1 for r in records_a if r.winner == "a") + a_wins += sum(1 for r in records_b if r.winner == "b") + + return a_wins / total_games + finally: + session.close() + + def _update_experience_in_db( + self, + team: str, + state_key: str, + action_key: str, + wins: float, + total: int + ) -> None: + """ + 更新单条经验记录（UPSERT）。 + + 如果记录已存在则累加wins和total，否则插入新记录。 + """ + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + existing = session.query(ExperienceRecord).filter_by( + team=team, + state_key=state_key, + action_key=action_key + ).first() + + if existing: + existing.wins += float(wins) + existing.total += int(total) + else: + record = ExperienceRecord( + team=team, + state_key=state_key, + action_key=action_key, + wins=float(wins), + total=int(total) + ) + session.add(record) + + session.commit() + except Exception as e: + session.rollback() + print(f" [!] 经验记录更新失败: {e}") + finally: + session.close() + + def _get_all_experience_from_db(self) -> List[ExperienceRecord]: + """ + 获取所有经验记录。 + + Returns + ------- + list : [ExperienceRecord, ...] + """ + engine = get_engine(self.config) + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + try: + return session.query(ExperienceRecord).all() + finally: + session.close() + + # ---------------------------------------------------------- + # JSON后端实现（保持原有逻辑） + # ---------------------------------------------------------- + + def _get_experience_from_json(self, team: str) -> Dict[str, Dict[str, Tuple[float, int]]]: + """从JSON文件读取经验数据""" + json_dir = get_json_dir() + result = {} + + for filename in os.listdir(json_dir): + if not filename.endswith(".json"): + continue + filepath = os.path.join(json_dir, filename) + with open(filepath, "r", encoding="utf-8") as f: + data = json.load(f) + + if team in data: + for state_key, actions in data[team].items(): + result[state_key] = {} + for action_key, stats in actions.items(): + result[state_key][action_key] = ( + float(stats.get("w", 0)), + int(stats.get("n", 0)) + ) + + return result + + def _save_experience_to_json(self, team: str, data: dict) -> None: + """ + 将经验数据保存到JSON文件。 + + 注意：这里简化处理，直接写入到默认文件。 + 完整实现需要按队伍名分离存储。 + """ + json_dir = get_json_dir() + os.makedirs(json_dir, exist_ok=True) + + # 合并已有数据 + existing_path = os.path.join(json_dir, "combined.json") + if os.path.exists(existing_path): + with open(existing_path, "r", encoding="utf-8") as f: + existing = json.load(f) + else: + existing = {"a": {}, "b": {}} + + # 更新数据 + for state_key, actions in data.items(): + if team not in existing: + existing[team] = {} + if state_key not in existing[team]: + existing[team][state_key] = {} + for action_key, (wins, total) in actions.items(): + if action_key not in existing[team][state_key]: + existing[team][state_key][action_key] = {"w": 0, "n": 0} + existing[team][state_key][action_key]["w"] += float(wins) + existing[team][state_key][action_key]["n"] += int(total) + + # 写入文件 + with open(existing_path, "w", encoding="utf-8") as f: + json.dump(existing, f, ensure_ascii=False) diff --git a/sim/experience_db.py b/sim/experience_db.py index b965cf9..05e57dc 100644 --- a/sim/experience_db.py +++ b/sim/experience_db.py @@ -10,6 +10,10 @@ HP 段：0=低危(≤25%), 1=中低(≤50%), 2=中高(≤75%), 3=满(>75%) 能量段：0=低(0-3), 1=中(4-6), 2=高(7-10) + +数据后端： + - 数据库（MariaDB）— 通过.env配置启用，支持并发写入 + - JSON文件 — 默认方式，向后兼容 """ import json @@ -117,6 +121,8 @@ class ExperienceDB: # 保存 db.save("毒队") + + 数据后端自动根据.env配置选择数据库或JSON文件。 """ def __init__(self): @@ -125,6 +131,18 @@ def __init__(self): "a": {}, "b": {} } self.total_games: int = 0 + # 标记哪些记录被修改过（用于增量保存） + self._dirty_keys: set = set() # {(team, state_key, action_key)} + + @property + def _use_db(self) -> bool: + """是否使用数据库后端""" + try: + from sim.data_store import DataStore + store = DataStore() + return store.db_enabled + except ImportError: + return False # ------------------------------------------------------------------ # 记录一局游戏 @@ -150,6 +168,7 @@ def record_game( ak = _action_key(action, battle_state.get_current(team)) self._get_or_create(team, sk, ak).wins += won self._get_or_create(team, sk, ak).total += 1 + self._dirty_keys.add((team, sk, ak)) def _get_or_create(self, team: str, sk: str, ak: str) -> ActionStats: team_db = self._db[team] @@ -205,9 +224,10 @@ def summary(self, team: str = "a", top_n: int = 8) -> str: return "\n".join(lines) # ------------------------------------------------------------------ - # 持久化 + # 持久化 — JSON后端（向后兼容） # ------------------------------------------------------------------ - def save(self, name: str = "default", directory: str = None) -> str: + def save_json(self, name: str = "default", directory: str = None) -> str: + """保存到JSON文件""" save_dir = directory or _DEFAULT_DIR os.makedirs(save_dir, exist_ok=True) filepath = os.path.join(save_dir, f"{name}.json") @@ -223,7 +243,8 @@ def save(self, name: str = "default", directory: str = None) -> str: json.dump(serialized, f, ensure_ascii=False, separators=(",", ":")) return filepath - def load(self, name: str = "default", directory: str = None) -> bool: + def load_json(self, name: str = "default", directory: str = None) -> bool: + """从JSON文件加载""" load_dir = directory or _DEFAULT_DIR filepath = os.path.join(load_dir, f"{name}.json") if not os.path.exists(filepath): @@ -243,6 +264,98 @@ def load(self, name: str = "default", directory: str = None) -> bool: } return True + # ------------------------------------------------------------------ + # 持久化 — 统一接口（自动选择后端） + # ------------------------------------------------------------------ + def save(self, name: str = "default", directory: str = None) -> str: + """ + 保存经验数据。 + + 如果数据库启用，保存到MariaDB；否则保存到JSON文件。 + """ + if self._use_db: + return self._save_to_db(name) + else: + return self.save_json(name, directory) + + def load(self, name: str = "default", directory: str = None) -> bool: + """ + 加载经验数据。 + + 如果数据库启用，从MariaDB加载；否则从JSON文件加载。 + """ + if self._use_db: + return self._load_from_db(name) + else: + return self.load_json(name, directory) + + # ------------------------------------------------------------------ + # 数据库后端 + # ------------------------------------------------------------------ + def _save_to_db(self, name: str = "default") -> str: + """ + 保存经验数据到MariaDB。 + + 采用增量保存策略：只更新被修改过的记录，减少I/O开销。 + """ + try: + from sim.data_store import DataStore + except ImportError as e: + print(f"[!] DataStore导入失败，回退到JSON: {e}") + return self.save_json(name) + + store = DataStore() + if not store.db_enabled: + return self.save_json(name) + + # 增量保存：只更新被修改过的记录 + for (team, sk, ak) in self._dirty_keys: + stats = self._db.get(team, {}).get(sk, {}).get(ak) + if stats is None: + continue + store.update_experience(team, sk, ak, stats.wins, stats.total) + + # 更新总局数 + store.update_total_games(name, self.total_games) + + self._dirty_keys.clear() + return f"db:{name}" + + def _load_from_db(self, name: str = "default") -> bool: + """ + 从MariaDB加载经验数据。 + + 由于数据库存储的是全局经验（不按队伍名称隔离）， + 所有队伍共享同一份经验库。这符合MCTS学习的设计—— + 历史对战的经验对所有阵容都有参考价值。 + """ + try: + from sim.data_store import DataStore + except ImportError as e: + print(f"[!] DataStore导入失败，回退到JSON: {e}") + return self.load_json(name) + + store = DataStore() + if not store.db_enabled: + return self.load_json(name, None) + + # 从数据库加载所有经验记录 + records = store.get_all_experience() + + self._db = {"a": {}, "b": {}} + for rec in records: + team, sk, ak = rec.team, rec.state_key, rec.action_key + if team not in self._db: + continue + if sk not in self._db[team]: + self._db[team][sk] = {} + self._db[team][sk][ak] = ActionStats(wins=rec.wins, total=rec.total) + + # 加载总局数 + self.total_games = store.get_total_games(name) + + return True + @classmethod def load_or_create(cls, name: str = "default", directory: str = None) -> "ExperienceDB": db = cls() diff --git a/sim/human_agent.py b/sim/human_agent.py new file mode 100644 index 0000000..566e2aa --- /dev/null +++ b/sim/human_agent.py @@ -0,0 +1,155 @@ +""" +Human Agent — 终端交互决策 + +每回合显示战场状态和合法动作列表，读取用户输入后返回 Action。 +实现 AgentProtocol 协议。 +""" + +from typing import List, Optional + +from sim.battle_engine import BattleEngine, Action +from sim.pokemon import Pokemon + + +def _hp_bar(current: int, maximum: int, width: int = 10) -> str: + if maximum <= 0: + return f"[{'?' * width}] ---/---" + pct = max(0.0, min(100.0, current / maximum * 100)) + filled = int(pct / (100 / width)) + return f"[{'#' * filled}{'.' * (width - filled)}] {current:4}/{maximum}" + + +def _status_flags(p: Pokemon) -> str: + parts = [] + if p.burn_stacks: parts.append(f"烧{p.burn_stacks}") + if p.poison_stacks: parts.append(f"毒{p.poison_stacks}") + if p.freeze_stacks: parts.append(f"冻{p.freeze_stacks}") + return " ".join(parts) + + +def _print_field(state, label_you: str, label_opp: str) -> None: + pa = state.get_current("a") + pb = state.get_current("b") + weather_str = f" 天气:{state.weather.value}" if state.weather.value != "none" else "" + lives_a = state.lives_a + lives_b = state.lives_b + print(f"\n{'─' * 56} 回合 {state.turn}{weather_str}") + print( + f" {label_you}: {pa.name:<10} {_hp_bar(pa.current_hp, pa.hp)} " + f"能量:{pa.energy:2} 生命格:{lives_a} {_status_flags(pa)}" + ) + print( + f" {label_opp}: {pb.name:<10} {_hp_bar(pb.current_hp, pb.hp)} " + f"能量:{pb.energy:2} 生命格:{lives_b} {_status_flags(pb)}" + ) + + +def _print_team_summary(label: str, team: list) -> None: + print(f"\n {label}:") + for i, p in enumerate(team): + s = "已倒下" if p.is_fainted else f"HP {p.current_hp}/{p.hp}" + print(f" [{i}] {p.name:<12} {s}") + + +def _print_actions(engine: BattleEngine, team: str) -> dict: + """打印合法动作列表，返回 (编号->Action) 映射""" + actions = engine.get_actions(team) + current = engine.state.get_current(team) + print(f"\n === {current.name} — 可选动作 ===") + + mapping = {} + for idx, action in enumerate(actions, 1): + if len(action) == 1 and action[0] == -1: + label = "汇合聚能（回复5点能量）" + elif len(action) >= 2 and action[0] == -2: + target_idx = action[1] + team_list = engine.state.get_team(team) + target = team_list[target_idx] + label = f"切换 -> {target.name} (HP:{target.current_hp}/{target.hp})" + else: + skill_idx = action[0] + skill = current.skills[skill_idx] + cost = engine._get_effective_energy_cost(skill, team) + category_label = { + "物攻": "物", + "魔攻": "魔", + "防御": "防", + "状态": "状", + }.get(skill.category.value, skill.category.value) + label = f"{skill.name}（{category_label}，威力{skill.power}，能耗{cost}" + if hasattr(engine.state.get_current(team), 'cooldowns'): + cd = engine.state.get_current(team).cooldowns.get(skill_idx, 0) + if cd > 0: + label += f" CD:{cd}" + # 特殊效果提示 + effects = [] + if skill.life_drain: effects.append(f"吸血{int(skill.life_drain*100)}%") + if skill.damage_reduction: effects.append(f"减伤{int(skill.damage_reduction*100)}%") + if skill.self_heal_hp: effects.append(f"回血{int(skill.self_heal_hp*100)}%") + if skill.poison_stacks: effects.append(f"中毒{skill.poison_stacks}") + if skill.burn_stacks: effects.append(f"灼烧{skill.burn_stacks}") + if skill.freeze_stacks: effects.append(f"冻结{skill.freeze_stacks}") + if skill.steal_energy: effects.append(f"偷能{skill.steal_energy}") + if skill.force_switch: effects.append("脱离") + if skill.agility: effects.append("迅捷") + if skill.charge: effects.append("蓄力") + if effects: + label += " [" + ",".join(effects) + "]" + label += ")" + mapping[idx] = action + print(f" {idx}. {label}") + + return mapping + + +def _read_choice(mapping: dict) -> Action: + """读取用户输入，返回对应的 Action""" + while True: + try: + raw = input(" 输入序号: ").strip() + if not raw: + continue + choice = int(raw) + if choice in mapping: + return mapping[choice] + print(f" [!] 无效序号，请输入 1-{len(mapping)}") + except ValueError: + print(" [!] 请输入数字") + + +class HumanAgent: + """ + 人类玩家代理 — 每回合通过终端交互选择动作。 + 实现 AgentProtocol 协议（含 show_team_status = True）。 + + Parameters + ---------- + team : "a" 或 "b" + label : 显示在战场上的队伍名称 + """ + + show_team_status: bool = True + + def __init__(self, team: str, label: str): + self.team = team + self.label = label + + # ------------------------------------------------------------------ + # AgentProtocol — choose_action + # ------------------------------------------------------------------ + + def choose_action(self, engine: BattleEngine) -> Action: + """显示状态，读取用户输入，返回 Action。""" + is_a = self.team == "a" + label_opp = engine.label_b if is_a else engine.label_a + _print_field(engine.state, f"{self.label}(你)", label_opp) + mapping = _print_actions(engine, self.team) + return _read_choice(mapping) + + # ------------------------------------------------------------------ + # AgentProtocol — on_game_end（人类玩家不记录经验） + # ------------------------------------------------------------------ + + def on_game_end(self, history: list, winner: Optional[str]) -> None: + """战斗结束时打印结果。""" + pass # battle.py 会统一打印结果 diff --git a/sim/llm_agent.py b/sim/llm_agent.py new file mode 100644 index 0000000..a72edee --- /dev/null +++ b/sim/llm_agent.py @@ -0,0 +1,473 @@ +""" +LLM Agent — 通过 OpenAI 兼容 API 的大模型战斗智能体 + +实现 AgentProtocol，每回合将战场状态序列化为文本发送给大模型， +解析其 JSON 回复得到 Action。 + +支持： +- 对战决策（逐回合选择动作） +- 战后经验文档生成 +- 战前经验加载作为上下文 +""" + +import json +import os +import time +from typing import Optional, List, Dict, Any + +import requests +import yaml as _yaml + +from sim.battle_engine import BattleEngine, Action +from sim.battle_state import BattleState +from sim.pokemon import Pokemon +from sim.skill import Skill +from sim.types import StatusType + +# ============================================================ +# 配置加载 +# ============================================================ + +_CONFIG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "data", "llm_config.yaml", +) + +_EXPERIENCE_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "data", "llm_experience", +) + + +def _load_llm_config() -> Dict: + """加载 LLM 配置文件。找不到或格式错误时抛出 EnvironmentError。""" + if not os.path.exists(_CONFIG_PATH): + raise EnvironmentError( + f"LLM 配置文件不存在: {_CONFIG_PATH}\n" + f"请在 data/llm_config.yaml 中配置 endpoint、model、api_key" + ) + with open(_CONFIG_PATH, "r", encoding="utf-8") as f: + cfg = _yaml.safe_load(f) + required = ["endpoint", "model", "api_key"] + missing = [k for k in required if not cfg.get(k)] + if missing: + raise EnvironmentError( + f"LLM 配置缺少必要字段: {missing}\n" + f"请在 data/llm_config.yaml 中补充 endpoint、model、api_key" + ) + return cfg + + +# ============================================================ +# HTTP 请求工具 +# ============================================================ + +def _call_llm( + messages: List[Dict[str, str]], + temperature: float = 0.3, + timeout: Optional[int] = None, +) -> str: + """ + 调用 OpenAI 兼容 API，返回 assistant 的回复文本。 + """ + cfg = _load_llm_config() + url = f"{cfg['endpoint']}/chat/completions" + + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {cfg['api_key']}", + } + payload = { + "model": cfg["model"], + "messages": messages, + "temperature": temperature, + } + + # 使用配置的超时值（如果未显式指定） + if timeout is None: + timeout = int(cfg.get("timeout", 120)) + + t0 = time.time() + # 内网/本地端点不经过系统代理 + if cfg["endpoint"].startswith("http://"): + # requests 在某些环境下即使 proxies={} 也会走系统代理， + # 需要临时清除环境变量 + import os as _os + proxy_vars = ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"] + saved = {} + for v in proxy_vars: + if v in _os.environ: + saved[v] = _os.environ[v] + del _os.environ[v] + try: + resp = requests.post(url, json=payload, headers=headers, + timeout=timeout) + finally: + for v, val in saved.items(): + _os.environ[v] = val + else: + resp = requests.post(url, json=payload, headers=headers, timeout=timeout) + elapsed = time.time() - t0 + print(f" [LLM] API调用耗时 {elapsed:.1f}s") + + if resp.status_code != 200: + raise EnvironmentError( + f"LLM API 请求失败: HTTP {resp.status_code}\n" + f"响应: {resp.text[:500]}" + ) + + data = resp.json() + content = data["choices"][0]["message"]["content"] + return content.strip() + + +def _parse_action(content: str) -> Action: + """ + 解析 LLM 返回的 JSON，提取 action 字段。 + 格式: {"action": [skill_idx], "reasoning": "..."} + {"action": [-1], "reasoning": "..."} # 汇合聚能 + {"action": [-2, target_idx], "reasoning": "..."} # 切换精灵 + """ + # 尝试从内容中提取 JSON（处理可能包含 Markdown 代码块的情况） + import re + json_match = re.search(r'\{[^{}]*"action"[^{}]*\}', content) + if json_match: + content = json_match.group(0) + + try: + obj = json.loads(content) + except json.JSONDecodeError: + # 尝试提取最外层的 JSON 对象 + start = content.find("{") + end = content.rfind("}") + 1 + if start >= 0 and end > start: + try: + obj = json.loads(content[start:end]) + except json.JSONDecodeError: + print(f" [LLM] JSON解析失败，原始内容: {content[:200]}") + raise + else: + print(f" [LLM] JSON解析失败，无法找到 JSON 对象: {content[:200]}") + raise + + action_list = obj.get("action") + if action_list is None or not isinstance(action_list, list): + raise ValueError(f"LLM返回缺少 'action' 字段: {obj}") + + # 转换为 tuple + return tuple(int(x) for x in action_list) + + +# ============================================================ +# 战场状态序列化 +# ============================================================ + +def _status_name(s: StatusType) -> str: + mapping = { + StatusType.NORMAL: "正常", + StatusType.POISONED: "中毒", + StatusType.BURNED: "灼烧", + StatusType.PARALYZED: "麻痹", + StatusType.FROZEN: "冻结", + StatusType.SLEEP: "睡眠", + StatusType.CONFUSED: "混乱", + } + return mapping.get(s, s.value) + + +def _serialize_pokemon(p: Pokemon, idx: int) -> Dict: + hp_pct = round(p.current_hp / p.hp * 100, 1) if p.hp > 0 else 0.0 + skills_info = [] + for si, sk in enumerate(p.skills): + cd = p.cooldowns.get(si, 0) + skills_info.append({ + "idx": si, + "name": sk.name, + "power": sk.power, + "cost": sk.energy_cost, + "category": sk.category.value, + "cd": cd if cd > 0 else None, + }) + return { + "idx": idx, + "name": p.name, + "hp_pct": hp_pct, + "hp_raw": f"{p.current_hp}/{p.hp}", + "energy": p.energy, + "is_fainted": p.is_fainted, + "burn_stacks": p.burn_stacks, + "poison_stacks": p.poison_stacks, + "freeze_stacks": p.freeze_stacks, + "skills": skills_info, + } + + +def _serialize_battle_state(engine: BattleEngine, my_team_id: str) -> Dict: + """ + 将当前战场状态序列化为字典（可 JSON 序列化）。 + """ + state = engine.state + enemy_id = "b" if my_team_id == "a" else "a" + + return { + "turn": state.turn, + "weather": state.weather.value, + "weather_turns_left": max(0, state.weather_turns), + "my_lives": state.lives_a if my_team_id == "a" else state.lives_b, + "enemy_lives": state.lives_b if my_team_id == "a" else state.lives_a, + "my_team": [ + _serialize_pokemon(p, i) + for i, p in enumerate(state.get_team(my_team_id)) + ], + "enemy_team": [ + _serialize_pokemon(p, i) + for i, p in enumerate(state.get_team(enemy_id)) + ], + } + + +# ============================================================ +# 经验文档管理 +# ============================================================ + +def _experience_path(team_name: str) -> str: + return os.path.join(_EXPERIENCE_DIR, f"{team_name}.json") + + +def _load_experience(team_name: str) -> Optional[List[Dict]]: + """加载队伍的历史经验文档列表。""" + path = _experience_path(team_name) + if not os.path.exists(path): + return None + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + # 兼容旧格式（单条记录） + if isinstance(data, dict): + return [data] + return data + except (json.JSONDecodeError, IOError): + return None + + +def _save_experience(team_name: str, experience: Dict) -> None: + """ + 追加一条经验文档到队伍的经验文件。 + 保留最近 50 条（防止无限增长）。 + """ + os.makedirs(_EXPERIENCE_DIR, exist_ok=True) + path = _experience_path(team_name) + + existing = [] + if os.path.exists(path): + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, dict): + existing = [data] + elif isinstance(data, list): + existing = data + except (json.JSONDecodeError, IOError): + pass + + # 添加时间戳 + experience["timestamp"] = time.strftime("%Y-%m-%d %H:%M:%S") + existing.append(experience) + + # 只保留最近 50 条 + if len(existing) > 50: + existing = existing[-50:] + + with open(path, "w", encoding="utf-8") as f: + json.dump(existing, f, ensure_ascii=False, indent=2) + + +def _format_experience_for_prompt(experiences: List[Dict]) -> str: + """ + 将经验文档格式化为简短的提示文本（不超过 1000 字符）。 + """ + if not experiences: + return "" + + lines = ["=== 历史对战经验 ==="] + # 只取最近 5 条 + recent = experiences[-5:] + for exp in recent: + result = exp.get("result", "未知") + summary = exp.get("summary", "")[:100] + lesson = exp.get("lessons", [""])[0][:200] if isinstance(exp.get("lessons"), list) else str(exp.get("lessons", ""))[:200] + lines.append(f"- 结果: {result} | 总结: {summary}") + if lesson: + lines.append(f" 教训: {lesson}") + + return "\n".join(lines) + + +# ============================================================ +# LLM Agent +# ============================================================ + +class LLMAgent: + """ + 大模型战斗智能体 — 每回合通过 HTTP API 调用大模型做决策。 + + Parameters + ---------- + team : "a" 或 "b" + team_name : 队伍名称（用于经验文档存储和策略加载） + temperature : LLM 温度参数，默认从配置文件读取 + """ + + show_team_status: bool = False # AI 模式，由引擎打印日志 + + def __init__(self, team: str, team_name: str, temperature: Optional[float] = None): + self.team = team + self.team_name = team_name + self._config = _load_llm_config() + self.temperature = temperature if temperature is not None else float(self._config.get("temperature", 0.3)) + self.timeout = int(self._config.get("timeout", 30)) + + # 加载历史经验文档 + self._experiences = _load_experience(team_name) + if self._experiences: + print(f" [LLM] {team_name}（{team}）加载了 {len(self._experiences)} 条历史经验") + else: + print(f" [LLM] {team_name}（{team}）无历史经验，从头开始") + + # ------------------------------------------------------------------ + # AgentProtocol — choose_action + # ------------------------------------------------------------------ + + def choose_action(self, engine: BattleEngine) -> Action: + """ + 将战场状态发送给 LLM，解析回复得到 Action。 + """ + state = _serialize_battle_state(engine, self.team) + enemy_id = "b" if self.team == "a" else "a" + + # 构建系统提示 + system_prompt = self._build_system_prompt(state) + + # 构建用户消息（战场状态 + 经验） + user_msg = self._build_user_message(state, enemy_id) + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_msg}, + ] + + try: + content = _call_llm(messages, self.temperature, self.timeout) + action = _parse_action(content) + print(f" [LLM] 选择动作: {action}") + return action + except Exception as e: + print(f" [LLM] 决策失败: {e}，回退到汇合聚能") + return (-1,) + + # ------------------------------------------------------------------ + # AgentProtocol — on_game_end + # ------------------------------------------------------------------ + + def on_game_end(self, history: list, winner: Optional[str]) -> None: + """ + 战斗结束后，调用 LLM 生成经验文档并保存。 + """ + result = { + "a": f"{self.team_name} 胜利", + "b": f"对手胜利", + None: "平局/超时", + }.get(winner, "未知") + + # 构建战后分析提示 + state = _serialize_battle_state( + type('FakeEngine', (), {'state': history[-1][0] if history else None})(), + self.team, + ) if history else {} + + try: + analysis = self._generate_post_battle_analysis(result, history) + experience = { + "team_name": self.team_name, + "result": result, + "winner": winner, + "summary": analysis.get("summary", ""), + "lessons": analysis.get("lessons", []), + "turns": len(history), + } + _save_experience(self.team_name, experience) + print(f" [LLM] {self.team_name} 经验文档已保存") + except Exception as e: + print(f" [LLM] 生成经验文档失败: {e}") + + # ------------------------------------------------------------------ + # Prompt 构建 + # ------------------------------------------------------------------ + + def _build_system_prompt(self, state: Dict) -> str: + return f"""你是洛克王国战斗AI。返回严格JSON: {{"action":[技能索引] 或 [-1]汇合聚能 或 [-2,精灵索引]换人}}。 +动作编码: [N]=用第N个技能(从0开始), [-1]=聚能(+5能量), [-2,N]=切到后备精灵索引N。""" + + def _build_user_message(self, state: Dict, enemy_id: str) -> str: + parts = [] + # 我方队伍 + for p in state["my_team"]: + st = "" + if p.get("burn_stacks", 0): st += f"灼烧{p['burn_stacks']} " + if p.get("poison_stacks", 0): st += f"中毒{p['poison_stacks']} " + if p.get("freeze_stacks", 0): st += f"冻结{p['freeze_stacks']} " + sk = ",".join(f"[{s['idx']}]={s['name']}({s['category']},威{s['power']},耗{s['cost']})" + for s in p["skills"]) + parts.append(f"我[{p['idx']}] {p['name']} HP:{p['hp_pct']}% E:{p['energy']} {'|' if st else ''}{st} 技能:[{sk}]") + # 对手 + for p in state["enemy_team"]: + parts.append(f"敌[{p['idx']}] {p['name']} HP:{p['hp_pct']}% E:{p['energy']} {'倒' if p.get('is_fainted') else '活'}") + # 经验 + exp_text = _format_experience_for_prompt(self._experiences or []) + if exp_text: + parts.append(exp_text) + return "\n".join(parts) + + def _generate_post_battle_analysis(self, result: str, history: list) -> Dict: + """ + 调用 LLM 生成战后分析文档。 + """ + system = """你是洛克王国手游的战斗分析师。请根据对战记录生成简短的经验总结。 + +返回严格的 JSON 格式: +- "summary": 一句话概括本局对战的关键点（50字以内） +- "lessons": 2-3条经验教训列表，每条100字以内 + +只返回 JSON，不要包含其他文本。""" + + # 构建对战记录摘要 + if history: + last_state = history[-1][0] if history else None + turn_count = len(history) + summary_text = f"\n本局共 {turn_count} 回合，结果: {result}。" + else: + turn_count = 0 + summary_text = "\n本局对战无有效记录。" + + user_msg = f"请分析以下对战并生成经验总结:{summary_text}" + + messages = [ + {"role": "system", "content": system}, + {"role": "user", "content": user_msg}, + ] + + content = _call_llm(messages, 0.5, self.timeout) + + try: + return json.loads(content) + except json.JSONDecodeError: + print(f" [LLM] 经验分析 JSON 解析失败: {content[:200]}") + return {"summary": "JSON解析失败", "lessons": []} + + # ------------------------------------------------------------------ + # 保存经验（兼容 MCTSAgent 接口） + # ------------------------------------------------------------------ + + def save(self) -> str: + """LLM Agent 不记录 MCTS 经验，此方法为空操作。""" + return "" diff --git a/sim/llm_team_generator.py b/sim/llm_team_generator.py new file mode 100644 index 0000000..e1f218a --- /dev/null +++ b/sim/llm_team_generator.py @@ -0,0 +1,344 @@ +""" +LLM 队伍生成器 — 使用大模型分析经验数据，生成优化后的精灵阵容和策略文件 + +功能： +- 读取 MCTS 历史对战经验（ExperienceDB） +- 调用 LLM 分析胜率数据、属性克制关系 +- 生成新的精灵组合建议 + 技能配置 + 策略 YAML +- 保存到队伍名册中 +""" + +import json +import os +from typing import Optional, List, Dict + +import yaml as _yaml + +from sim.llm_agent import _call_llm, _load_llm_config + +# ============================================================ +# 路径常量 +# ============================================================ + +_STRATEGY_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "data", "strategies", +) + +_EXPERIENCE_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "data", "experience", +) + +_ROSTER_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "data", "teams.json", +) + +_SPRITES_DB = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "data", "sprites.json", +) + +# ============================================================ +# 数据加载辅助 +# ============================================================ + +def _load_sprites_db() -> List[Dict]: + """读取精灵数据库，返回精简信息（名字、属性、六维）。""" + if not os.path.exists(_SPRITES_DB): + return [] + with open(_SPRITES_DB, "r", encoding="utf-8") as f: + data = json.load(f) + + result = [] + for sprite in data: + stats = sprite.get("stats", {}) + attrs = sprite.get("attributes", []) + primary_type = attrs[0] if attrs else "未知" + secondary_type = attrs[1] if len(attrs) > 1 else "" + skills_list = [s.get("name", "") for s in sprite.get("skills", [])] + + info = { + "name": sprite.get("name", ""), + "primary_type": primary_type, + "secondary_type": secondary_type, + "hp": stats.get("hp", 0), + "attack": stats.get("atk", 0), + "defense": stats.get("def", 0), + "sp_attack": stats.get("sp_atk", 0), + "sp_defense": stats.get("sp_def", 0), + "speed": stats.get("spd", 0), + "total": stats.get("total", 0), + "ability": sprite.get("ability", {}).get("name", ""), + "skills": skills_list, + } + result.append(info) + return result + + +def _load_experience_summary() -> Dict: + """ + 读取所有 MCTS 经验文件，汇总胜率统计。 + 返回 {team_name: {total_games, total_wins_a, ...}} + """ + summary = {} + if not os.path.exists(_EXPERIENCE_DIR): + return summary + + for fname in os.listdir(_EXPERIENCE_DIR): + if not fname.endswith(".json"): + continue + path = os.path.join(_EXPERIENCE_DIR, fname) + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + team_name = fname.replace(".json", "") + # 经验格式: {state_key: {action_key: {wins: X, total: Y}, ...}, ...} + summary[team_name] = { + "total_keys": len(data), + "path": path, + } + except (json.JSONDecodeError, IOError): + pass + return summary + + +# ============================================================ +# LLM 队伍生成 +# ============================================================ + +def generate_team_with_llm( + theme: Optional[str] = None, +) -> Optional[Dict]: + """ + 调用 LLM 生成一个新的精灵阵容。 + + Parameters + ---------- + theme : 主题/风格（如 "高速控制队", "重火力输出队", None=自由发挥） + + Returns + ------- + 队伍定义 dict，格式与 teams.json 中的条目一致；失败返回 None。 + """ + sprites = _load_sprites_db() + exp_summary = _load_experience_summary() + + # 构建精灵列表（精简，只保留关键信息） + sprite_lines = [] + # 按总战力排序，取前 80 只 + sorted_sprites = sorted(sprites, key=lambda s: s.get("total", 0), reverse=True)[:80] + for s in sorted_sprites: + primary = s.get("primary_type", "") + secondary = s.get("secondary_type", "") + type_str = primary + ("+" + secondary if secondary else "") + ability = s.get("ability", "") + skills = ", ".join(s.get("skills", [])[:4]) # 只取前4个技能 + sprite_lines.append( + f"{s['name']}: 属性={type_str}, " + f"HP{s['hp']} ATK{s['attack']} DEF{s['defense']} " + f"SPATK{s['sp_attack']} SPDEF{s['sp_defense']} SPD{s['speed']} " + f"总计{s['total']}, 特性={ability}, 技能=[{skills}]" + ) + + # 经验摘要 + exp_lines = [] + for team, info in exp_summary.items(): + exp_lines.append(f"- {team}: {info['total_keys']} 个状态记录") + + system_prompt = """你是洛克王国手游的阵容设计专家。请根据提供的精灵数据库和已有队伍经验， +设计一支新的6人精灵阵容。 + +胜利条件：每方4格生命格，精灵倒下-1格，归零判负。 +核心机制：同时行动制、速度决定先后手、能量系统(初始10, 技能消耗2-8)、属性克制。 + +你必须返回严格的 JSON 格式: +{{ + "team_name": "队伍名称", + "theme": "阵容风格描述", + "members": [ + {{"pokemon": "精灵名", "skills": ["技能1", "技能2", "技能3", "技能4"]}} + ], + "strategy_notes": "简短的战术说明（100字以内）" +}} + +要求： +- 6只精灵，每只4个技能 +- 属性搭配合理（攻防兼备、有控场能力） +- 考虑前后排轮换策略 +- 精灵名必须在提供的列表中 +- 只返回 JSON，不要包含其他文本""" + + theme_hint = f"\n\n用户指定的主题: {theme}" + user_msg = ( + "=== 可用精灵 ===\n" + + "\n".join(sprite_lines[:200]) # 限制长度，避免 prompt 过长 + + f"\n(共 {len(sorted_sprites)} 只精灵，已按总战力排序取前80)" + + (f"\n\n=== 已有队伍经验 ===\n{chr(10).join(exp_lines)}" if exp_lines else "") + + theme_hint + ) + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_msg}, + ] + + try: + content = _call_llm(messages, temperature=0.7) + if not content or len(content.strip()) < 10: + print(f" [LLM] LLM 返回空内容，重试...") + # 第二次尝试，降低温度 + content = _call_llm(messages, temperature=0.3) + result = json.loads(content) + + # 验证返回格式 + if not isinstance(result.get("members"), list) or len(result["members"]) < 4: + print(f" [LLM] 生成的队伍格式不正确: {result.get('team_name', '未知')}") + return None + + # 补充 preset=false + result["preset"] = False + return result + + except json.JSONDecodeError as e: + print(f" [LLM] JSON解析失败: {e}\n原始内容前200字符: {content[:200] if 'content' in dir() else 'N/A'}") + return None + except Exception as e: + print(f" [LLM] 生成队伍失败: {e}") + return None + + +# ============================================================ +# LLM 策略文件生成 +# ============================================================ + +def generate_strategy_with_llm( + team_name: str, +) -> Optional[Dict]: + """ + 调用 LLM 为指定队伍生成策略 YAML 数据。 + + Parameters + ---------- + team_name : 队伍名称（必须在 teams.json 中存在） + + Returns + ------- + 策略 dict（可写入 YAML），失败返回 None。 + """ + # 读取队伍定义 + if not os.path.exists(_ROSTER_PATH): + print(f" [LLM] 找不到队伍名册: {_ROSTER_PATH}") + return None + + with open(_ROSTER_PATH, "r", encoding="utf-8") as f: + roster = json.load(f) + + team_def = None + for t in roster: + if t["name"] == team_name: + team_def = t + break + + if team_def is None: + print(f" [LLM] 找不到队伍「{team_name}」") + return None + + # 读取策略模板 + template_path = os.path.join(_STRATEGY_DIR, "_template.yaml") + template_info = "" + if os.path.exists(template_path): + with open(template_path, "r", encoding="utf-8") as f: + template_info = f.read()[:500] + + # 读取 LLM 经验文档 + from sim.llm_agent import _load_experience + llm_exp = _load_experience(team_name) or [] + exp_text = "" + if llm_exp: + lines = ["=== 历史对战经验 ==="] + for exp in llm_exp[-5:]: + result = exp.get("result", "未知") + summary = exp.get("summary", "")[:100] + lessons = exp.get("lessons", []) + lesson_str = "; ".join(str(l) for l in lessons)[:200] if isinstance(lessons, list) else str(lessons)[:200] + lines.append(f"- 结果: {result} | 总结: {summary}") + if lesson_str: + lines.append(f" 教训: {lesson_str}") + exp_text = "\n".join(lines) + + system_prompt = ( + f"你是洛克王国手游的策略设计师。请为队伍「{team_name}」生成一份策略配置文件。" + f"\n\n该队伍的阵容:\n" + + json.dumps(team_def["members"], ensure_ascii=False, indent=2) + + (f"\n\n{exp_text}" if exp_text else "") + + "\n\n策略文件用于 MCTS AI 的权重调整，格式如下（YAML）：" + "\n- prefer: 明确推荐的动作类型列表（如 ['attack', 'switch']）" + "\n- avoid: 明确排斥的动作类型列表" + "\n- conditions: 条件判断规则（hp_low, type_advantage, energy_low 等）" + "\n- priorities: 优先级设置" + + "\n\n你只需要返回 JSON 格式的策略数据：" + '{"prefer": ["动作1", "动作2"], "avoid": ["动作3"], ' + '"conditions": [{"when": "hp_low", "action": "switch"}], ' + '"notes": "策略说明"}' + + "\n\n只返回 JSON，不要包含其他文本。" + ) + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": f"请为队伍「{team_name}」生成策略配置。"}, + ] + + try: + content = _call_llm(messages, temperature=0.5) + result = json.loads(content) + return result + except Exception as e: + print(f" [LLM] 生成策略失败: {e}") + return None + + +# ============================================================ +# 保存生成的队伍/策略到磁盘 +# ============================================================ + +def save_generated_team(team_def: Dict) -> bool: + """ + 将 LLM 生成的队伍保存到 teams.json。 + """ + if not os.path.exists(_ROSTER_PATH): + print(f" [!] 找不到队伍名册") + return False + + with open(_ROSTER_PATH, "r", encoding="utf-8") as f: + roster = json.load(f) + + # 检查是否已存在同名 + for i, t in enumerate(roster): + if t["name"] == team_def["name"]: + print(f" [!] 队伍「{team_def['name']}」已存在，将覆盖") + roster[i] = team_def + break + else: + roster.append(team_def) + + with open(_ROSTER_PATH, "w", encoding="utf-8") as f: + json.dump(roster, f, ensure_ascii=False, indent=2) + + print(f" [OK] 队伍「{team_def['name']}」已保存") + return True + + +def save_generated_strategy(team_name: str, strategy_data: Dict) -> bool: + """ + 将 LLM 生成的策略保存到 strategies/ 目录。 + """ + os.makedirs(_STRATEGY_DIR, exist_ok=True) + path = os.path.join(_STRATEGY_DIR, f"{team_name}.yaml") + + with open(path, "w", encoding="utf-8") as f: + _yaml.dump(strategy_data, f, allow_unicode=True, default_flow_style=False) + + print(f" [OK] 策略文件已保存: {path}") + return True diff --git a/sim/mcts_agent.py b/sim/mcts_agent.py index 3232365..43ae2be 100644 --- a/sim/mcts_agent.py +++ b/sim/mcts_agent.py @@ -18,6 +18,7 @@ from sim.mcts import MCTSSearch from sim.experience_db import ExperienceDB from sim.strategy import load_strategy, get_starter_idx +from sim.agent_base import AgentProtocol, GameHistory # ============================================================ @@ -37,6 +38,8 @@ class MCTSAgent: load_exp : 是否自动从磁盘加载历史经验 """ + show_team_status: bool = False + def __init__( self, team: str, @@ -74,6 +77,15 @@ def choose_action(self, engine: BattleEngine) -> Action: """根据当前引擎状态，用 MCTS 选择最优动作。""" return self._search.search(engine.state) + # ------------------------------------------------------------------ + # 局后处理 — 记录经验并保存 + # ------------------------------------------------------------------ + + def on_game_end(self, history: GameHistory, winner: Optional[str]) -> None: + """记录本局经验到 ExperienceDB 并持久化。""" + self.experience_db.record_game(history, winner) + self.save() + # ------------------------------------------------------------------ # 保存经验 # ------------------------------------------------------------------