yexuejc/cultivation-world-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add asyncio

Browse Source
This commit is contained in:
bridge
2025-09-02 22:16:03 +08:00
parent 0f9f5f35f7
commit bd28da21f5
7 changed files with 90 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -103,6 +103,7 @@
### 🤖 AI增强系统
- ✅ LLM接口集成
- ✅ 角色AI系统（规则AI + LLM AI）
- ✅ 协程化决策机制，异步运行
- [ ] AI动作链系统（长期规划和目标导向行为）
- [ ] 突发动作响应系统（对外界刺激的即时反应）
- [ ] LLM驱动的NPC对话、思考、互动、事件总结

16
src/classes/ai.py
View File

@@ -13,7 +13,7 @@ from src.classes.tile import Region
from src.classes.root import corres_essence_type
from src.classes.action import ACTION_SPACE_STR
from src.classes.event import Event, NULL_EVENT
from src.utils.llm import get_ai_prompt_and_call_llm
from src.utils.llm import get_ai_prompt_and_call_llm_async
from src.classes.typings import ACTION_NAME, ACTION_PARAMS, ACTION_PAIR
if TYPE_CHECKING:
@@ -26,12 +26,12 @@ class AI(ABC):
def __init__(self, avatar: Avatar):
self.avatar = avatar
def decide(self, world: World) -> tuple[ACTION_NAME, ACTION_PARAMS, Event]:
async def decide(self, world: World) -> tuple[ACTION_NAME, ACTION_PARAMS, Event]:
"""
决定做什么，同时生成对应的事件
"""
# 先决定动作和参数
action_name, action_params = self._decide(world)
action_name, action_params = await self._decide(world)
# 获取动作对象并生成事件
action = self.avatar.create_action(action_name)
@@ -40,7 +40,7 @@ class AI(ABC):
return action_name, action_params, event
@abstractmethod
def _decide(self, world: World) -> ACTION_PAIR:
async def _decide(self, world: World) -> ACTION_PAIR:
"""
决策逻辑：决定执行什么动作和参数
由子类实现具体的决策逻辑
@@ -51,7 +51,7 @@ class RuleAI(AI):
"""
规则AI
"""
def _decide(self, world: World) -> ACTION_PAIR:
async def _decide(self, world: World) -> ACTION_PAIR:
"""
决策逻辑：决定执行什么动作和参数
先做一个简单的：
@@ -93,9 +93,9 @@ class LLMAI(AI):
不能每个单步step都调用一次LLM来决定下一步做什么。这样子一方面动作一直乱变，另一方面也太费token了。
decide的作用是，拉取既有的动作链（如果没有了就call_llm），再根据动作链决定动作，以及动作之间的衔接。
"""
def _decide(self, world: World) -> ACTION_PAIR:
async def _decide(self, world: World) -> ACTION_PAIR:
"""
决策逻辑：通过LLM决定执行什么动作和参数
异步决策逻辑：通过LLM决定执行什么动作和参数
"""
action_space_str = ACTION_SPACE_STR
avatar_infos_str = str(self.avatar)
@@ -107,6 +107,6 @@ class LLMAI(AI):
"regions": regions_str,
"avatar_persona": avatar_persona
}
res = get_ai_prompt_and_call_llm(dict_info)
res = await get_ai_prompt_and_call_llm_async(dict_info)
action_name, action_params = res["action_name"], res["action_params"]
return action_name, action_params

8
src/classes/avatar.py
View File

@@ -60,8 +60,8 @@ class Avatar:
在Avatar创建后自动初始化tile和AI
"""
self.tile = self.world.map.get_tile(self.pos_x, self.pos_y)
# self.ai = LLMAI(self)
self.ai = RuleAI(self)
self.ai = LLMAI(self)
# self.ai = RuleAI(self)
def __str__(self) -> str:
"""
@@ -91,7 +91,7 @@ class Avatar:
raise ValueError(f"未找到名为 '{action_name}' 的动作类")
def act(self):
async def act(self):
"""
角色执行动作。
实际上分为两步：决定做什么（decide）和实际去做（do）
@@ -101,7 +101,7 @@ class Avatar:
if self.cur_action_pair is None:
# 决定动作时生成事件
action_name, action_args, event = self.ai.decide(self.world)
action_name, action_args, event = await self.ai.decide(self.world)
action = self.create_action(action_name)
self.cur_action_pair = (action, action_args)

43
src/front/front.py
View File

@@ -1,5 +1,6 @@
import math
from typing import Dict, List, Optional, Tuple
import asyncio # 新增：导入asyncio
# Front 只依赖项目内部类型定义与 pygame
from src.sim.simulator import Simulator
@@ -103,11 +104,21 @@ class Front:
if len(self.events) > 1000:
self.events = self.events[-1000:]
def run(self):
"""主循环"""
async def _step_once_async(self):
"""异步执行一步模拟"""
events = await self.simulator.step() # 获取返回的事件
if events: # 新增：将事件添加到事件历史
self.add_events(events)
self._last_step_ms = 0
async def run_async(self):
"""异步主循环"""
pygame = self.pygame
running = True
# 用于存储正在进行的step任务
current_step_task = None
while running:
dt_ms = self.clock.tick(60)
self._last_step_ms += dt_ms
@@ -122,23 +133,35 @@ class Front:
elif event.key == pygame.K_a:
self._auto_step = not self._auto_step
elif event.key == pygame.K_SPACE:
self._step_once()
# 手动步进：创建新任务
if current_step_task is None or current_step_task.done():
current_step_task = asyncio.create_task(self._step_once_async())
# 自动步进
if self._auto_step and self._last_step_ms >= self.step_interval_ms:
self._step_once()
# 自动步进：创建新任务
if current_step_task is None or current_step_task.done():
current_step_task = asyncio.create_task(self._step_once_async())
self._last_step_ms = 0
# 检查step任务是否完成
if current_step_task and current_step_task.done():
try:
await current_step_task # 获取结果（如果有异常会抛出）
except Exception as e:
print(f"Step执行出错: {e}")
current_step_task = None
self._render()
# 使用asyncio.sleep而不是pygame的时钟，避免阻塞
await asyncio.sleep(0.016) # 约60fps
pygame.quit()
def _step_once(self):
"""执行一步模拟"""
events = self.simulator.step() # 获取返回的事件
if events: # 新增：将事件添加到事件历史
self.add_events(events)
self._last_step_ms = 0
"""执行一步模拟（同步版本，已弃用）"""
print("警告：_step_once已弃用，请使用异步版本")
pass
def _render(self):
"""渲染主画面"""

12
src/run/run.py
View File

@@ -1,6 +1,12 @@
import random
import asyncio
import sys
import os
from typing import List, Tuple, Dict, Any
# 添加项目根目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
# 依赖项目内部模块
from src.front.front import Front
from src.sim.simulator import Simulator
@@ -81,7 +87,7 @@ def make_avatars(world: World, count: int = 12, current_month_stamp: MonthStamp
return avatars
def main():
async def main():
# 为了每次更丰富，使用随机种子；如需复现可将 seed 固定
game_map = create_cultivation_world_map()
@@ -101,9 +107,9 @@ def main():
window_title="Cultivation World — Front Demo",
sidebar_width=350, # 新增：设置侧边栏宽度
)
front.run()
await front.run_async()
if __name__ == "__main__":
main()
asyncio.run(main())

4
src/sim/simulator.py
View File

@@ -13,7 +13,7 @@ class Simulator:
self.world = world
self.brith_rate = 0 # 0表示不出生新角色
def step(self):
async def step(self):
"""
前进一步（每步模拟是一个月时间）
结算这个时间内的所有情况。
@@ -26,7 +26,7 @@ class Simulator:
# 结算角色行为
for avatar_id, avatar in self.avatars.items():
event = avatar.act()
event = await avatar.act()
if not is_null_event(event):
events.append(event)
if avatar.death_by_old_age():

34
src/utils/llm.py
View File

@@ -2,6 +2,7 @@ from litellm import completion
from langchain.prompts import PromptTemplate
from pathlib import Path
import json
import asyncio
from src.utils.config import CONFIG
from src.utils.io import read_txt
@@ -36,6 +37,18 @@ def call_llm(prompt: str) -> str:
# 返回生成的内容
return response.choices[0].message.content
async def call_llm_async(prompt: str) -> str:
"""
异步调用LLM
Args:
prompt: 输入的提示词
Returns:
str: LLM返回的结果
"""
# 使用asyncio.to_thread包装同步调用
return await asyncio.to_thread(call_llm, prompt)
def get_prompt_and_call_llm(template_path: Path, infos: dict) -> str:
"""
根据模板，获取提示词，并调用LLM
@@ -48,9 +61,28 @@ def get_prompt_and_call_llm(template_path: Path, infos: dict) -> str:
# print(f"res = {res}")
return json_res
async def get_prompt_and_call_llm_async(template_path: Path, infos: dict) -> str:
"""
异步版本：根据模板，获取提示词，并调用LLM
"""
template = read_txt(template_path)
prompt = get_prompt(template, infos)
res = await call_llm_async(prompt)
json_res = json.loads(res)
print(f"prompt = {prompt}")
print(f"res = {res}")
return json_res
def get_ai_prompt_and_call_llm(infos: dict) -> dict:
"""
根据模板，获取提示词，并调用LLM
"""
template_path = CONFIG.paths.templates / "ai.txt"
return get_prompt_and_call_llm(template_path, infos)
return get_prompt_and_call_llm(template_path, infos)
async def get_ai_prompt_and_call_llm_async(infos: dict) -> dict:
"""
异步版本：根据模板，获取提示词，并调用LLM
"""
template_path = CONFIG.paths.templates / "ai.txt"
return await get_prompt_and_call_llm_async(template_path, infos)