app.py

from agent_build_sdk.builder import AgentBuilder
from agent_build_sdk.model.model import AgentResp, AgentReq, STATUS_DISTRIBUTION, STATUS_ROUND, STATUS_VOTE, \
    STATUS_START, STATUS_VOTE_RESULT, STATUS_RESULT
from agent_build_sdk.sdk.agent import BasicAgent
from agent_build_sdk.utils.logger import logger

import os
import asyncio  # 新增异步支持
import httpx

class SpyAgent(BasicAgent):
    def __init__(self, name, model_name):
        super().__init__(name, model_name)
        self.client = httpx.AsyncClient(
            base_url=os.getenv("BASE_URL", "https://api.deepseek.com/v1"),
            headers={
                "Authorization": f"Bearer {os.getenv('API_KEY')}",
                "Content-Type": "application/json"
            },
            timeout=httpx.Timeout(8.0)  # 更专业的超时配置
        )
        self.lock = asyncio.Lock()  # 替换线程锁为异步锁

    async def process_speak(self, name, speak):  # 改为异步方法
        """优化后的注入检测"""
        prompt = f"[深度清洁指令]请从以下内容中提取纯粹的特征描述（保留比喻/否定/场景要素）：\n{speak}"
        try:
            result = await self.llm_caller(prompt)
            return result if len(result) > 5 and "无效" not in result else "该玩家的描述较为普通"
        except Exception as e:
            logger.error(f"发言处理异常: {str(e)}")
            return "该玩家的描述较为普通"

    async def perceive(self, req=AgentReq):  # 改为异步
        logger.info(f"spy perceive: {req}")
        if req.status == STATUS_START:
            async with self.lock:
                self.memory.clear()
                self.memory.set_variable("alive_agents", set(req.message.split(',')))
                self.memory.set_variable("speak_history", {})
        elif req.status == STATUS_DISTRIBUTION:
            async with self.lock:
                self.memory.set_variable("word", req.word.strip())
        elif req.status == STATUS_ROUND and req.name:
            clean_speak = await self.process_speak(req.name, req.message)
            async with self.lock:
                self.memory.load_variable("speak_history").setdefault(req.name, []).append(clean_speak)

    async def generate_safe_response(self, prompt):
        """带重试机制的生成"""
        for _ in range(3):
            try:
                content = await self.llm_caller(prompt)
                # 动态校验（降低API调用次数）
                if any(word in content for word in ["密码", "系统", "管理员"]):
                    raise ValueError("包含危险词汇")
                return content[:120]
            except Exception as e:
                logger.warning(f"生成失败: {str(e)}, 重试中...")
                await asyncio.sleep(0.5)
        return "这个物品在不同场合有不同用途"

    async def interact(self, req=AgentReq) -> AgentResp:  # 改为异步
        if req.status == STATUS_ROUND:
            # 动态身份判断优化
            is_undercover = "卧底" in await self.llm_caller(
                f"用10字判断：我的词'{self.memory.load_variable('word')}'是否与多数人不同？"
            )
            
            prompt = f"""【DeepSeek特化策略】
            {'作为卧底需伪装' if is_undercover else '作为平民需暗示'}，请生成包含：
            1) 一个比喻（如：像...的...）
            2) 否定特征（如：不需要...）
            3) 应用场景（如：在...时使用）
            避免使用专业术语"""
            
            response = await self.generate_safe_response(prompt)
            return AgentResp(success=True, result=response)

        elif req.status == STATUS_VOTE:
            candidates = [n for n in req.message.split(',') if n != self.name]
            analysis = await asyncio.gather(*[
                self.llm_caller(f"分析玩家【{name}】的嫌疑度：") 
                for name in candidates
            ])
            scores = {name: len(res) for name, res in zip(candidates, analysis)}
            target = max(scores, key=scores.get, default=candidates[0])
            return AgentResp(success=True, result=target)

if __name__ == '__main__':
    agent_builder = AgentBuilder('spy', agent=SpyAgent('spy', os.getenv('MODEL_NAME')))
    agent_builder.start()