This commit is contained in:
bridge
2026-01-03 23:25:38 +08:00
parent 6129e0f898
commit 775d830ec3
12 changed files with 355 additions and 142 deletions

View File

@@ -36,6 +36,7 @@ from .nurture_weapon import NurtureWeapon
from .switch_weapon import SwitchWeapon
from .assassinate import Assassinate
from .move_to_direction import MoveToDirection
from .cast import Cast
# 注册到 ActionRegistry标注是否为实际可执行动作
register_action(actual=False)(Action)
@@ -68,6 +69,7 @@ register_action(actual=True)(NurtureWeapon)
register_action(actual=True)(SwitchWeapon)
register_action(actual=True)(Assassinate)
register_action(actual=True)(MoveToDirection)
register_action(actual=True)(Cast)
# Talk 已移动到 mutual_action 模块,在那里注册
__all__ = [
@@ -103,6 +105,7 @@ __all__ = [
"SwitchWeapon",
"Assassinate",
"MoveToDirection",
"Cast",
# Talk 已移动到 mutual_action 模块
# Occupy 已移动到 mutual_action 模块
]

157
src/classes/action/cast.py Normal file
View File

@@ -0,0 +1,157 @@
from __future__ import annotations
import random
from typing import Optional, TYPE_CHECKING, List
from src.classes.action import TimedAction
from src.classes.cultivation import Realm
from src.classes.event import Event
from src.classes.item import Item
from src.classes.weapon import get_random_weapon_by_realm
from src.classes.auxiliary import get_random_auxiliary_by_realm
from src.classes.single_choice import handle_item_exchange
from src.utils.config import CONFIG
if TYPE_CHECKING:
from src.classes.avatar import Avatar
class Cast(TimedAction):
"""
铸造动作:消耗同阶材料,尝试打造同阶宝物(兵器或辅助装备)。
持续时间3个月
"""
ACTION_NAME = "铸造"
EMOJI = "🔥"
DESC = "消耗材料尝试铸造法宝"
DOABLES_REQUIREMENTS = f"拥有{getattr(CONFIG.action.cast, 'cost', 10)}个同阶材料"
PARAMS = {"target_realm": "目标境界名称('练气''筑基''金丹''元婴'"}
IS_MAJOR = False
duration_months = 3
def __init__(self, avatar: Avatar, world):
super().__init__(avatar, world)
self.target_realm: Optional[Realm] = None
def _get_cost(self) -> int:
# 从配置读取消耗数量默认为10
return getattr(CONFIG.action.cast, "cost", 10)
def _count_materials(self, realm: Realm) -> int:
"""
统计符合条件的材料数量。
注意:仅统计 Item 类的直接实例,不统计 Weapon/Auxiliary 等子类(它们也是 Item但通常不作为铸造原材料
"""
count = 0
for item, qty in self.avatar.items.items():
# 这里使用 type(item) is Item 来严格限制必须是基础材料
# 如果项目里有其他继承自 Item 的材料类,可能需要放宽这个限制
if type(item).__name__ == "Item" and item.realm == realm:
count += qty
return count
def can_start(self, target_realm: str = "") -> tuple[bool, str]:
if not target_realm:
return False, "未指定目标境界"
try:
realm = Realm(target_realm)
except ValueError:
return False, f"无效的境界: {target_realm}"
cost = self._get_cost()
count = self._count_materials(realm)
if count < cost:
return False, f"材料不足,需要 {cost}{target_realm}阶材料,当前拥有 {count}"
return True, ""
def start(self, target_realm: str = "") -> Event:
self.target_realm = Realm(target_realm)
cost = self._get_cost()
# 扣除材料逻辑
to_deduct = cost
items_to_modify = []
# 再次遍历寻找材料进行扣除
for item, qty in self.avatar.items.items():
if to_deduct <= 0:
break
if type(item).__name__ == "Item" and item.realm == self.target_realm:
take = min(qty, to_deduct)
items_to_modify.append((item, take))
to_deduct -= take
for item, take in items_to_modify:
self.avatar.remove_item(item, take)
return Event(
self.world.month_stamp,
f"{self.avatar.name} 开始闭关铸造{target_realm}阶法宝,投入了大量材料。",
related_avatars=[self.avatar.id]
)
def _execute(self) -> None:
# 持续过程中无特殊逻辑
pass
async def finish(self) -> list[Event]:
if self.target_realm is None:
return []
# 1. 计算成功率
base_rate = float(getattr(CONFIG.action.cast, "base_success_rate", 0.3))
extra_rate = float(self.avatar.effects.get("extra_cast_success_rate", 0.0))
success_rate = base_rate + extra_rate
events = []
# 2. 判定结果
if random.random() > success_rate:
# 失败
fail_event = Event(
self.world.month_stamp,
f"{self.avatar.name} 铸造{self.target_realm.value}阶法宝失败,所有材料化为灰烬。",
related_avatars=[self.avatar.id],
is_major=False
)
events.append(fail_event)
return events
# 3. 成功:生成物品
# 50% 兵器50% 辅助装备
is_weapon = random.random() < 0.5
new_item = None
item_type = ""
item_label = ""
if is_weapon:
new_item = get_random_weapon_by_realm(self.target_realm)
item_type = "weapon"
item_label = "兵器"
else:
new_item = get_random_auxiliary_by_realm(self.target_realm)
item_type = "auxiliary"
item_label = "辅助装备"
# 4. 决策:保留还是卖出
base_desc = f"铸造成功!获得了{self.target_realm.value}{item_label}{new_item.name}』。"
_, result_text = await handle_item_exchange(
avatar=self.avatar,
new_item=new_item,
item_type=item_type,
context_intro=base_desc,
can_sell_new=True
)
events.append(Event(
self.world.month_stamp,
result_text,
related_avatars=[self.avatar.id],
is_major=True
))
return events

View File

@@ -194,6 +194,31 @@ EXTRA_FORTUNE_PROBABILITY = "extra_fortune_probability"
- 极高: 0.01 (1%,不少了)
"""
EXTRA_MISFORTUNE_PROBABILITY = "extra_misfortune_probability"
"""
额外霉运概率
类型: float
结算: src/classes/misfortune.py
数值参考:
- 基础概率通常极低 (<0.01)
- 微量: 0.001 (0.1%,有)
- 中量: 0.002~0.005 (高)
- 极高: 0.01 (1%,不少了)
"""
# --- 铸造相关 ---
EXTRA_CAST_SUCCESS_RATE = "extra_cast_success_rate"
"""
额外铸造成功率
类型: float
结算: src/classes/action/cast.py
说明: 铸造Cast动作的成功率加成。
数值参考:
- 微量: 0.05 (+5%)
- 中量: 0.1 (+10%)
- 大量: 0.2+ (+20%)
"""
# --- 兵器相关 ---
EXTRA_WEAPON_PROFICIENCY_GAIN = "extra_weapon_proficiency_gain"
"""
@@ -391,6 +416,9 @@ ALL_EFFECTS = [
"extra_fortune_probability", # float - 额外奇遇概率
"extra_misfortune_probability", # float - 额外霉运概率
# 铸造相关
"extra_cast_success_rate", # float - 额外铸造成功率
# 兵器相关
"extra_weapon_proficiency_gain", # float - 额外兵器熟练度增长倍率
"extra_weapon_upgrade_chance", # float - 额外兵器升华概率

View File

@@ -394,49 +394,8 @@ async def try_trigger_fortune(avatar: Avatar) -> list[Event]:
actors_for_story = [avatar] # 用于生成故事的角色列表
# 导入单选决策模块
from src.classes.single_choice import make_decision, format_swap_choice_desc
async def _resolve_choice(
new_obj: Any,
old_obj: Optional[Any],
type_label: str,
extra_context: str = ""
) -> tuple[bool, str]:
"""
通用决策辅助函数
Returns: (should_replace, result_text)
"""
new_name = new_obj.name
new_grade_val = getattr(new_obj, "grade", getattr(new_obj, "realm", None)).value
if old_obj is None:
return True, f"{avatar.name} 获得{new_grade_val}{type_label}{new_name}"
old_name = old_obj.name
swap_desc = format_swap_choice_desc(new_obj, old_obj, type_label)
options = [
{
"key": "A",
"desc": f"保留原{type_label}{old_name}』,放弃新{type_label}{new_name}』。"
},
{
"key": "B",
"desc": f"卖掉原{type_label}{old_name}』换取灵石,接受新{type_label}{new_name}』。\n{swap_desc}"
}
]
base_context = f"你在奇遇中发现了{new_grade_val}{type_label}{new_name}』,但你手中已有『{old_name}』。"
context = f"{base_context} {extra_context}".strip()
choice = await make_decision(avatar, context, options)
if choice == "A":
return False, f"{avatar.name} 放弃了{new_grade_val}{type_label}{new_name}』,保留了『{old_name}"
else:
return True, f"{avatar.name} 获得了{new_grade_val}{type_label}{new_name}』,卖掉了『{old_name}"
# 导入通用决策模块
from src.classes.single_choice import handle_item_exchange
if kind == FortuneKind.WEAPON:
weapon = _get_weapon_for_avatar(avatar)
@@ -445,14 +404,17 @@ async def try_trigger_fortune(avatar: Avatar) -> list[Event]:
kind = FortuneKind.TECHNIQUE
theme = _pick_theme(kind)
else:
should_equip, res_text = await _resolve_choice(
weapon, avatar.weapon, "兵器"
intro = f"你在奇遇中发现了{weapon.realm.value}兵器『{weapon.name}』。"
if avatar.weapon:
intro += f" 但你手中已有『{avatar.weapon.name}』。"
_, res_text = await handle_item_exchange(
avatar=avatar,
new_item=weapon,
item_type="weapon",
context_intro=intro,
can_sell_new=False
)
if should_equip:
# 自动卖掉旧武器
if avatar.weapon is not None:
avatar.sell_weapon(avatar.weapon)
avatar.change_weapon(weapon)
if kind == FortuneKind.AUXILIARY:
auxiliary = _get_auxiliary_for_avatar(avatar)
@@ -461,29 +423,34 @@ async def try_trigger_fortune(avatar: Avatar) -> list[Event]:
kind = FortuneKind.TECHNIQUE
theme = _pick_theme(kind)
else:
should_equip, res_text = await _resolve_choice(
auxiliary, avatar.auxiliary, "辅助装备"
intro = f"你在奇遇中发现了{auxiliary.realm.value}辅助装备『{auxiliary.name}』。"
if avatar.auxiliary:
intro += f" 但你手中已有『{avatar.auxiliary.name}』。"
_, res_text = await handle_item_exchange(
avatar=avatar,
new_item=auxiliary,
item_type="auxiliary",
context_intro=intro,
can_sell_new=False
)
if should_equip:
# 自动卖掉旧辅助装备
if avatar.auxiliary is not None:
avatar.sell_auxiliary(avatar.auxiliary)
avatar.change_auxiliary(auxiliary)
if kind == FortuneKind.TECHNIQUE:
tech = _get_fortune_technique_for_avatar(avatar)
if tech is None:
return []
should_learn, res_text = await _resolve_choice(
tech, avatar.technique, "功法",
extra_context=f"这与你当前主修的『{avatar.technique.name if avatar.technique else ''}』冲突。"
intro = f"你在奇遇中领悟了上品功法『{tech.name}』。"
if avatar.technique:
intro += f" 这与你当前主修的『{avatar.technique.name}』冲突。"
_, res_text = await handle_item_exchange(
avatar=avatar,
new_item=tech,
item_type="technique",
context_intro=intro,
can_sell_new=False
)
if should_learn:
avatar.technique = tech
elif kind == FortuneKind.FIND_MASTER:
master = _find_potential_master(avatar)

View File

@@ -2,7 +2,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import random
from src.classes.single_choice import make_decision, format_swap_choice_desc
from src.classes.single_choice import handle_item_exchange
if TYPE_CHECKING:
from src.classes.avatar import Avatar
@@ -27,7 +27,7 @@ async def kill_and_grab(winner: Avatar, loser: Avatar) -> str:
# 检查辅助装备
if loser.auxiliary:
loot_candidates.append(("auxiliary", loser.auxiliary))
loot_candidates.append(("auxiliary", loser.auxiliary))
if not loot_candidates:
return ""
@@ -39,50 +39,24 @@ async def kill_and_grab(winner: Avatar, loser: Avatar) -> str:
best_candidates = [c for c in loot_candidates if c[1].realm == best_realm]
loot_type, loot_item = random.choice(best_candidates)
should_loot = False
# 判定是否夺取
# 1. 如果winner当前部位为空直接夺取
winner_current = getattr(winner, loot_type)
if winner_current is None :
should_loot = True
else:
# 其他情况下都让 AI 决策
# 构建详细描述,包含效果
item_label = '兵器' if loot_type == 'weapon' else '辅助装备'
context = f"战斗胜利,{loser.name} 身死道消,留下了一件{loot_item.realm.value}{item_label}{loot_item.name}』。"
swap_desc = format_swap_choice_desc(loot_item, winner_current, item_label)
options = [
{
"key": "A",
"desc": f"夺取『{loot_item.name}』,卖掉身上的『{winner_current.name}』换取灵石。\n{swap_desc}"
},
{
"key": "B",
"desc": f"放弃『{loot_item.name}』,保留身上的『{winner_current.name}』。"
}
]
choice = await make_decision(winner, context, options)
if choice == "A":
should_loot = True
item_label = '兵器' if loot_type == 'weapon' else '辅助装备'
context = f"战斗胜利,{loser.name} 身死道消,留下了一件{loot_item.realm.value}{item_label}{loot_item.name}』。"
if should_loot:
swapped, log_text = await handle_item_exchange(
avatar=winner,
new_item=loot_item,
item_type=loot_type,
context_intro=context,
can_sell_new=False
)
if swapped:
if loot_type == "weapon":
# 自动卖掉旧武器
if winner.weapon is not None:
winner.sell_weapon(winner.weapon)
winner.change_weapon(loot_item)
loser.change_weapon(None)
else:
# 自动卖掉旧辅助装备
if winner.auxiliary is not None:
winner.sell_auxiliary(winner.auxiliary)
winner.change_auxiliary(loot_item)
loser.change_auxiliary(None)
return f"{winner.name}夺取了对方的{loot_item.realm.value}{loot_item.name}』!"
return log_text
return ""

View File

@@ -1,4 +1,4 @@
from typing import Any, Dict, List, TYPE_CHECKING
from typing import Any, Dict, List, TYPE_CHECKING, Tuple, Optional, Callable
from src.utils.llm import call_llm_with_task_name
from src.utils.config import CONFIG
import json
@@ -13,15 +13,6 @@ async def make_decision(
) -> str:
"""
让角色在多个选项中做出单选决策。
Args:
avatar: 做出决策的角色
context_desc: 决策背景描述
options: 选项列表,每个选项是一个字典,必须包含 'key''desc' 字段
例如: [{'key': 'A', 'desc': '...'}, {'key': 'B', 'desc': '...'}]
Returns:
str: AI 选择的选项 Key (如 'A''B')
"""
# 1. 获取角色信息 (详细模式)
avatar_infos = str(avatar.get_info(detailed=True))
@@ -60,15 +51,13 @@ async def make_decision(
data = json.loads(json_str)
choice = data.get("choice", "").strip()
except (json.JSONDecodeError, ValueError):
# 如果 JSON 解析失败,直接看字符串内容是否就是选项 key
choice = clean_result
# 有时候 llm 会输出 "choice: A",这里做个兼容
else:
choice = clean_result
# 验证 choice 是否在 options key 中
valid_keys = {opt["key"] for opt in options}
# 简单的容错:如果返回的是 "A." 或者 "A "
# 简单的容错
if choice not in valid_keys:
for k in valid_keys:
if k in choice:
@@ -76,29 +65,117 @@ async def make_decision(
break
if choice not in valid_keys:
# 兜底:默认选第一个
choice = options[0]["key"]
return choice
def format_swap_choice_desc(new_item: Any, old_item: Any | None, item_type_name: str) -> str:
def _get_item_ops(avatar: "Avatar", item_type: str) -> dict:
"""根据物品类型返回对应的操作函数和标签"""
if item_type == "weapon":
return {
"label": "兵器",
"get_current": lambda: avatar.weapon,
"equip": avatar.change_weapon,
"sell": avatar.sell_weapon
}
elif item_type == "auxiliary":
return {
"label": "辅助装备",
"get_current": lambda: avatar.auxiliary,
"equip": avatar.change_auxiliary,
"sell": avatar.sell_auxiliary
}
elif item_type == "technique":
return {
"label": "功法",
"get_current": lambda: avatar.technique,
"equip": lambda x: setattr(avatar, 'technique', x),
"sell": None # 功法通常不能卖
}
else:
raise ValueError(f"Unsupported item type: {item_type}")
async def handle_item_exchange(
avatar: "Avatar",
new_item: Any,
item_type: str, # "weapon", "auxiliary", "technique"
context_intro: str,
can_sell_new: bool = False
) -> Tuple[bool, str]:
"""
生成替换装备/功法时的决策描述文本
通用处理物品(装备/功法)的获取、替换与决策逻辑
Args:
new_item: 新获得的物品对象(必须实现 get_info(detailed=True)
old_item: 当前持有的物品对象(可能为 None
item_type_name: 物品类型名称(如"兵器""功法"
avatar: 角色对象
new_item: 新获得的物品
item_type: 物品类型键值 ("weapon", "auxiliary", "technique")
context_intro: 决策背景描述
can_sell_new: 如果拒绝装备,是否允许卖掉新物品换灵石
Returns:
(swapped, result_text)
"""
ops = _get_item_ops(avatar, item_type)
label = ops["label"]
current_item = ops["get_current"]()
new_name = new_item.name
new_grade = getattr(new_item, "realm", getattr(new_item, "grade", None)).value
# 1. 自动装备:当前无装备且不强制考虑卖新
if current_item is None and not can_sell_new:
ops["equip"](new_item)
return True, f"{avatar.name} 获得了{new_grade}{label}{new_name}』并装备。"
# 2. 需要决策:准备描述
old_name = current_item.name if current_item else ""
new_info = new_item.get_info(detailed=True)
if old_item:
old_info = old_item.get_info(detailed=True)
return (
f"现有{item_type_name}{old_info}\n"
f"{item_type_name}{new_info}\n"
f"(选择替换将卖出旧{item_type_name}"
)
swap_desc = f"{label}{new_info}"
if current_item:
old_info = current_item.get_info(detailed=True)
swap_desc = f"现有{label}{old_info}\n{swap_desc}"
if ops["sell"]:
swap_desc += f"\n(选择替换将卖出旧{label}"
# 3. 构建选项
# Option A: 装备新物品
opt_a_text = f"装备新{label}{new_name}"
if current_item and ops["sell"]:
opt_a_text += f",卖掉旧{label}{old_name}"
elif current_item:
opt_a_text += f",替换旧{label}{old_name}"
# Option B: 拒绝新物品
if can_sell_new and ops["sell"]:
opt_b_text = f"卖掉新{label}{new_name}』换取灵石,保留现状"
else:
return f"{item_type_name}{new_info}"
opt_b_text = f"放弃『{new_name}"
if current_item:
opt_b_text += f",保留身上的『{old_name}"
options = [
{"key": "A", "desc": opt_a_text},
{"key": "B", "desc": opt_b_text}
]
full_context = f"{context_intro}\n{swap_desc}"
choice = await make_decision(avatar, full_context, options)
# 4. 执行决策
if choice == "A":
# 卖旧(如果有且能卖)
if current_item and ops["sell"]:
ops["sell"](current_item)
# 装新
ops["equip"](new_item)
return True, f"{avatar.name} 换上了{new_grade}{label}{new_name}』。"
else:
# 卖新(如果被要求且能卖)
if can_sell_new and ops["sell"]:
sold_price = ops["sell"](new_item)
return False, f"{avatar.name} 卖掉了新获得的{new_name},获利 {sold_price} 灵石。"
else:
return False, f"{avatar.name} 放弃了{new_name}"