Files
commander-ai-lab/coach/services/deck_generator.py
T

640 lines
25 KiB
Python

"""
Commander AI Lab — V3 Deck Generator Service
═════════════════════════════════════════════
Generates Commander decks using Anthropic Claude
structured JSON output, then runs ownership check
and smart substitution.
"""
import json
import logging
import re
import sqlite3
from typing import List, Optional, Tuple
from urllib.request import urlopen, Request
from ..clients.anthropic_client import AnthropicClient
from ..schemas.deck_schema import DECK_LIST_SCHEMA, GeneratedDeckList
from ..schemas.substitution_schema import (
CardAlternative, DeckCardWithStatus, SubstitutionBatchResult,
)
from ..prompts.deck_gen_system import SYSTEM_PROMPT
from ..prompts.deck_gen_user import build_user_prompt
from ..prompts.substitution_prompt import (
SUBSTITUTION_SYSTEM, build_substitution_prompt,
)
from ..config import (
DECK_GEN_MODEL, DECK_GEN_TEMPERATURE, DECK_GEN_MAX_TOKENS,
SUBSTITUTION_MIN_SIMILARITY, SUBSTITUTION_MAX_ALTERNATIVES,
SUBSTITUTION_MODEL, SUBSTITUTION_USE_ANTHROPIC_FALLBACK,
)
logger = logging.getLogger("coach.deckgen")
class DeckGeneratorV3:
"""
V3 Deck Generator using Anthropic Claude structured output.
Pipeline:
1. Resolve commander via Scryfall
2. Build collection summary
3. Call Anthropic Claude with structured JSON schema
4. Validate card names via Scryfall
5. Check ownership against collection DB
6. Run Smart Substitution (embedding + Anthropic Claude fallback)
7. Return finalized deck with substitution data
"""
def __init__(self, anthropic_client: AnthropicClient, db_conn_factory,
embedding_index=None):
"""
Args:
anthropic_client: Configured AnthropicClient instance
db_conn_factory: Callable that returns sqlite3.Connection
embedding_index: Optional MTGEmbeddingIndex for substitution
"""
self.anthropic = anthropic_client
self.db_conn_factory = db_conn_factory
self.embedding_index = embedding_index
# ── Step 1: Resolve Commander ────────────────────────────
def resolve_commander(self, commander_name: str) -> dict:
"""Look up commander on Scryfall to get canonical name, color identity, type."""
try:
encoded = commander_name.replace(" ", "+")
url = f"https://api.scryfall.com/cards/named?fuzzy={encoded}"
req = Request(url)
req.add_header("User-Agent", "CommanderAILab/3.0")
req.add_header("Accept", "application/json")
with urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
return {
"name": data.get("name", commander_name),
"scryfall_id": data.get("id", ""),
"color_identity": data.get("color_identity", []),
"type_line": data.get("type_line", ""),
"mana_cost": data.get("mana_cost", ""),
"image_url": data.get("image_uris", {}).get("normal", ""),
"oracle_text": data.get("oracle_text", ""),
}
except Exception as e:
logger.warning("Scryfall lookup failed for '%s': %s", commander_name, e)
return {
"name": commander_name,
"scryfall_id": "",
"color_identity": [],
"type_line": "",
"mana_cost": "",
"image_url": "",
"oracle_text": "",
}
# ── Step 2: Build Collection Summary ─────────────────────
def build_collection_summary(self, color_identity: List[str],
max_per_group: int = 20) -> str:
"""
Build a text summary of owned cards matching the deck's color identity.
Groups by card type for the LLM prompt.
"""
conn = self.db_conn_factory()
try:
rows = conn.execute(
"SELECT name, type_line, cmc, oracle_text, quantity, tcg_price "
"FROM collection_entries WHERE quantity > 0"
).fetchall()
except Exception as e:
logger.warning("Failed to load collection: %s", e)
return ""
if not rows:
return ""
# Filter by color identity
ci_set = set(color_identity) if color_identity else None
filtered = []
for row in rows:
name = row["name"]
type_line = row["type_line"] or ""
# Basic lands and colorless are always allowed
if ci_set:
# Simple color check from type_line and name
# For proper filtering we'd check mana cost but this is good enough for summary
card_colors = set()
oracle = row["oracle_text"] or ""
mana_cost = ""
try:
mana_cost = row["mana_cost"] or ""
except (IndexError, KeyError):
pass
for sym in ["W", "U", "B", "R", "G"]:
if f"{{{sym}}}" in (mana_cost or oracle):
card_colors.add(sym)
if card_colors and not card_colors.issubset(ci_set):
continue
filtered.append(row)
if not filtered:
return ""
# Group by card type
groups = {}
for row in filtered:
type_line = row["type_line"] or ""
if "Creature" in type_line:
group = "Creatures"
elif "Instant" in type_line:
group = "Instants"
elif "Sorcery" in type_line:
group = "Sorceries"
elif "Artifact" in type_line:
group = "Artifacts"
elif "Enchantment" in type_line:
group = "Enchantments"
elif "Planeswalker" in type_line:
group = "Planeswalkers"
elif "Land" in type_line:
group = "Lands"
else:
group = "Other"
groups.setdefault(group, []).append(row)
# Sort each group by popularity/price and build text
lines = [f"COLLECTION SUMMARY ({len(filtered)} cards in deck colors):"]
for group_name in ["Creatures", "Instants", "Sorceries", "Artifacts",
"Enchantments", "Planeswalkers", "Lands", "Other"]:
cards = groups.get(group_name, [])
if not cards:
continue
# Sort by price descending (higher value = likely better card)
cards.sort(key=lambda r: r["tcg_price"] or 0, reverse=True)
card_strs = []
for c in cards[:max_per_group]:
price = f"${c['tcg_price']:.2f}" if c["tcg_price"] else "$?"
card_strs.append(f"{c['name']} ({price})")
lines.append(f" {group_name} ({len(cards)}): {', '.join(card_strs)}")
return "\n".join(lines)
# ── Step 3: Generate Deck via Anthropic Claude ─────────────────
def generate_deck(
self,
commander_name: str,
strategy: str = "",
target_bracket: int = 3,
budget_usd: float = None,
budget_mode: str = "total",
omit_cards: List[str] = None,
use_collection: bool = True,
model: str = None,
) -> dict:
"""
Generate a complete Commander deck.
Returns dict with:
- commander: dict with Scryfall data
- raw_deck: parsed GeneratedDeckList from Anthropic Claude
- cards: list of DeckCardWithStatus (after ownership check)
- stats: deck statistics
- tokens_used: {prompt, completion}
"""
# Step 1: Resolve commander
cmdr = self.resolve_commander(commander_name)
color_identity = cmdr["color_identity"]
logger.info("Commander: %s, CI: %s", cmdr["name"], color_identity)
# Step 2: Collection summary
collection_block = ""
if use_collection:
collection_block = self.build_collection_summary(color_identity)
logger.info("Collection summary: %d chars",
len(collection_block) if collection_block else 0)
# Step 3: Build prompts
user_prompt = build_user_prompt(
commander=cmdr["name"],
commander_type=cmdr["type_line"],
color_identity=color_identity,
strategy=strategy,
target_bracket=target_bracket,
budget_usd=budget_usd,
budget_mode=budget_mode,
omit_cards=omit_cards,
collection_summary=collection_block,
)
# Step 4: Call Anthropic Claude with structured output
use_model = model or DECK_GEN_MODEL
logger.info("Calling Anthropic Claude (%s) for deck generation...", use_model)
response = self.anthropic.chat_structured(
system_prompt=SYSTEM_PROMPT,
user_prompt=user_prompt,
json_schema=DECK_LIST_SCHEMA,
schema_name="DeckList",
model=use_model,
temperature=DECK_GEN_TEMPERATURE,
max_tokens=DECK_GEN_MAX_TOKENS,
)
if not response.ok:
raise ValueError(
f"Anthropic Claude returned invalid response: {response.content[:300]}"
)
raw_deck = response.parsed_json
card_count = len(raw_deck.get("cards", []))
truncated = card_count < 90 # Commander decks should have ~100 cards
logger.info("Deck generated: %d cards, bracket %d%s",
card_count,
(raw_deck.get("bracket", {}).get("level", 0) if isinstance(raw_deck.get("bracket"), dict) else raw_deck.get("bracket", 0)),
" (TRUNCATED — response may have hit token limit)" if truncated else "")
# Step 5: Check ownership and build enriched card list
cards_with_status = self._check_ownership(raw_deck.get("cards", []))
# Step 5b: Fill basic lands to hit exactly 100 cards
cards_with_status = self._fill_basic_lands(cards_with_status, color_identity)
# Step 5c: Secondary safety-net deduplication
seen = {}
deduped = []
for c in cards_with_status:
key = c.name.lower()
if key in seen:
deduped[seen[key]].count += c.count
logger.warning("Secondary dedupe caught duplicate: '%s'", c.name)
else:
seen[key] = len(deduped)
deduped.append(c)
cards_with_status = deduped
# Step 6: Compute stats
stats = self._compute_stats(cards_with_status)
return {
"commander": cmdr,
"color_identity": color_identity,
"raw_deck": raw_deck,
"cards": [c.model_dump() for c in cards_with_status],
"stats": stats,
"strategy_summary": raw_deck.get("strategy_summary", ""),
"bracket": raw_deck.get("bracket", {}),
"archetype": raw_deck.get("archetype", ""),
"reasoning": raw_deck.get("reasoning", {}),
"estimated_total_usd": raw_deck.get("estimated_total_usd", 0),
"tokens_used": {
"prompt": response.prompt_tokens,
"completion": response.completion_tokens,
},
"model": response.model,
"citations": response.citations,
"truncated": truncated,
}
# ── Step 5: Ownership Check ──────────────────────────────
def _check_ownership(self, cards: List[dict]) -> List[DeckCardWithStatus]:
"""
Cross-reference generated cards against the collection DB.
Sets owned/scryfall_id/owned_qty and real price.
Deduplicates cards by name (case-insensitive) — if the LLM
returns the same card twice, the counts are merged.
"""
conn = self.db_conn_factory()
enriched = []
seen_names: dict[str, int] = {} # name_lower → index in enriched
for card in cards:
name = card.get("name", "")
if not name:
continue
name_lower = name.lower()
if name_lower in seen_names:
idx = seen_names[name_lower]
enriched[idx].count += card.get("count", 1)
logger.warning("Duplicate card '%s' merged (LLM returned it twice)", name)
continue
# Look up in collection
row = conn.execute(
"SELECT name, scryfall_id, quantity, tcg_price, type_line, cmc "
"FROM collection_entries WHERE name = ? COLLATE NOCASE LIMIT 1",
(name,)
).fetchone()
is_owned = bool(row and row["quantity"] and row["quantity"] > 0)
enriched_card = DeckCardWithStatus(
name=row["name"] if row else name,
count=card.get("count", 1),
category=card.get("category", ""),
role_tags=card.get("role_tags", []),
reason=card.get("reason", ""),
estimated_price_usd=(
round(row["tcg_price"], 2) if row and row["tcg_price"]
else card.get("estimated_price_usd", 0)
),
synergy_with=card.get("synergy_with", []),
owned=is_owned,
owned_qty=int(row["quantity"]) if row and row["quantity"] else 0,
scryfall_id=row["scryfall_id"] if row else "",
status="owned" if is_owned else "missing",
)
seen_names[name_lower] = len(enriched)
enriched.append(enriched_card)
return enriched
# ── Step 5b: Basic Land Fill ──────────────────────────────
_COLOR_TO_BASIC = {
'W': 'Plains', 'U': 'Island', 'B': 'Swamp',
'R': 'Mountain', 'G': 'Forest',
}
_BASIC_NAMES = set(_COLOR_TO_BASIC.values()) | {'Wastes'}
def _fill_basic_lands(
self,
cards: List[DeckCardWithStatus],
color_identity: List[str],
target_total: int = 100,
) -> List[DeckCardWithStatus]:
"""
Strip LLM-provided basics, then add the right number evenly
across the commander's colors to reach exactly target_total.
"""
non_basics = [c for c in cards if c.name not in self._BASIC_NAMES]
current_total = sum(c.count for c in non_basics)
basics_needed = max(0, target_total - current_total)
if basics_needed == 0:
return non_basics
ci = [c.upper() for c in color_identity]
basic_names = [self._COLOR_TO_BASIC[c] for c in ci if c in self._COLOR_TO_BASIC]
if not basic_names:
basic_names = ['Wastes']
per_color = basics_needed // len(basic_names)
remainder = basics_needed % len(basic_names)
for i, bname in enumerate(basic_names):
qty = per_color + (1 if i < remainder else 0)
if qty > 0:
non_basics.append(DeckCardWithStatus(
name=bname,
count=qty,
category='Land',
role_tags=[],
reason='Basic land for mana fixing',
estimated_price_usd=0.10,
owned=True,
owned_qty=qty,
status='owned',
))
logger.info("Basic land fill: %d basics added across %d color(s)",
basics_needed, len(basic_names))
return non_basics
# ── Step 6: Smart Substitution ───────────────────────────
def run_substitution(
self,
cards: List[DeckCardWithStatus],
commander: dict,
strategy: str = "",
) -> SubstitutionBatchResult:
"""
Run the Smart Substitution Engine on missing cards.
Pipeline:
a) Identify missing cards
b) For each: try embedding similarity (free, fast)
c) For low-confidence matches: batch Anthropic Claude fallback
d) Auto-select best substitute per card
Returns SubstitutionBatchResult with all cards updated.
"""
color_identity = commander.get("color_identity", [])
missing = [c for c in cards if not c.owned]
owned = [c for c in cards if c.owned]
if not missing:
return SubstitutionBatchResult(
total_cards=len(cards),
owned_count=len(owned),
substituted_count=0,
missing_count=0,
cards=cards,
)
logger.info("Substitution: %d missing / %d owned", len(missing), len(owned))
# Build list of owned card names for substitution pool
conn = self.db_conn_factory()
owned_pool_rows = conn.execute(
"SELECT DISTINCT name FROM collection_entries WHERE quantity > 0"
).fetchall()
owned_pool = [r["name"] for r in owned_pool_rows]
# Exclude cards already in the deck
deck_names = set(c.name.lower() for c in cards)
available_pool = [n for n in owned_pool if n.lower() not in deck_names]
# a) Embedding pass
needs_anthropic_fallback = []
for card in missing:
embedding_alts = self._embedding_substitution(
card, color_identity, available_pool
)
card.alternatives = embedding_alts
# Check if best match meets threshold
if embedding_alts and embedding_alts[0].similarity_score >= SUBSTITUTION_MIN_SIMILARITY:
card.selected_substitute = embedding_alts[0].name
card.status = "substituted"
# Remove selected from pool
if card.selected_substitute.lower() in [n.lower() for n in available_pool]:
available_pool = [n for n in available_pool
if n.lower() != card.selected_substitute.lower()]
else:
needs_anthropic_fallback.append(card)
# b) Anthropic Claude fallback for low-confidence matches
if needs_anthropic_fallback and SUBSTITUTION_USE_ANTHROPIC_FALLBACK:
self._anthropic_substitution(
needs_anthropic_fallback, available_pool,
commander, strategy
)
# Count results
substituted = sum(1 for c in cards if c.status == "substituted")
still_missing = sum(1 for c in cards if c.status == "missing")
return SubstitutionBatchResult(
total_cards=len(cards),
owned_count=len(owned),
substituted_count=substituted,
missing_count=still_missing,
cards=cards,
)
def _embedding_substitution(
self,
card: DeckCardWithStatus,
color_identity: List[str],
available_pool: List[str],
) -> List[CardAlternative]:
"""Find embedding-similar owned cards as substitutes."""
if not self.embedding_index or not self.embedding_index.loaded:
return []
# Use the embeddings search to find similar cards
matches = self.embedding_index.search_similar(
query_card=card.name,
color_filter=color_identity if color_identity else None,
type_filter=card.category if card.category else None,
exclude_cards=[card.name],
top_n=SUBSTITUTION_MAX_ALTERNATIVES * 3, # Over-fetch, then filter to owned
)
# Filter to only cards in the available pool
pool_lower = set(n.lower() for n in available_pool)
alternatives = []
for match in matches:
if match.name.lower() in pool_lower:
# Compute role overlap
match_roles = [] # Would need card's role_tags from DB
role_overlap = list(set(card.role_tags) & set(match_roles)) if match_roles else []
alternatives.append(CardAlternative(
name=match.name,
similarity_score=match.similarity,
reason=f"Similar to {card.name} (cosine sim: {match.similarity:.3f})",
source="embedding",
role_overlap=role_overlap,
cmc_delta=match.mana_value - (card.estimated_price_usd or 0),
))
if len(alternatives) >= SUBSTITUTION_MAX_ALTERNATIVES:
break
return alternatives
def _anthropic_substitution(
self,
missing_cards: List[DeckCardWithStatus],
available_pool: List[str],
commander: dict,
strategy: str,
):
"""
Batch Anthropic Claude fallback for cards with low embedding confidence.
Mutates the cards in-place with substitution data.
"""
if not available_pool:
return
# Build the request
missing_dicts = [
{
"name": c.name,
"role_tags": c.role_tags,
"category": c.category,
"reason": c.reason,
}
for c in missing_cards
]
user_prompt = build_substitution_prompt(
missing_cards=missing_dicts,
allowed_cards=available_pool[:200], # Limit to avoid token overflow
commander=commander.get("name", ""),
color_identity=commander.get("color_identity", []),
strategy=strategy,
)
try:
from ..schemas.substitution_schema import SUBSTITUTION_RESPONSE_SCHEMA
response = self.anthropic.chat_structured(
system_prompt=SUBSTITUTION_SYSTEM,
user_prompt=user_prompt,
json_schema=SUBSTITUTION_RESPONSE_SCHEMA,
schema_name="SubstitutionResponse",
model=SUBSTITUTION_MODEL,
temperature=0.1,
max_tokens=4096,
)
if not response.ok:
logger.warning("Anthropic Claude substitution fallback failed — non-JSON response")
return
subs = response.parsed_json.get("substitutions", [])
logger.info("Anthropic Claude suggested %d substitutions", len(subs))
# Apply substitutions
card_by_name = {c.name.lower(): c for c in missing_cards}
for sub in subs:
orig = sub.get("original", "").lower()
substitute = sub.get("substitute", "")
reason = sub.get("reason", "")
if orig in card_by_name and substitute:
card = card_by_name[orig]
# Add as top alternative
anthropic_alt = CardAlternative(
name=substitute,
similarity_score=0.80, # Nominal score for Anthropic Claude suggestions
reason=reason,
source="Anthropic Claude",
role_overlap=sub.get("role_overlap", []),
)
# Insert at front
card.alternatives.insert(0, anthropic_alt)
card.selected_substitute = substitute
card.status = "substituted"
except Exception as e:
logger.error("Anthropic Claude substitution fallback error: %s", e)
# ── Stats ────────────────────────────────────────────────
@staticmethod
def _compute_stats(cards: List[DeckCardWithStatus]) -> dict:
"""Compute deck statistics from the enriched card list."""
by_category = {}
by_status = {"owned": 0, "substituted": 0, "missing": 0}
total_price = 0.0
owned_price = 0.0
missing_price = 0.0
role_counts = {}
for card in cards:
cat = card.category or "Other"
by_category[cat] = by_category.get(cat, 0) + card.count
by_status[card.status] = by_status.get(card.status, 0) + 1
price = card.estimated_price_usd * card.count
total_price += price
if card.owned:
owned_price += price
else:
missing_price += price
for role in card.role_tags:
role_counts[role] = role_counts.get(role, 0) + 1
land_count = by_category.get("Land", 0)
nonland_count = sum(v for k, v in by_category.items() if k != "Land")
return {
"total_cards": sum(c.count for c in cards),
"land_count": land_count,
"nonland_count": nonland_count,
"by_category": by_category,
"by_status": by_status,
"role_counts": role_counts,
"total_price_usd": round(total_price, 2),
"owned_price_usd": round(owned_price, 2),
"missing_price_usd": round(missing_price, 2),
}