| import os |
| import sys |
|
|
| from fastapi import FastAPI |
| from fastapi.responses import HTMLResponse |
| from fastapi.responses import RedirectResponse |
| import uvicorn |
|
|
| from seer.seer_agent import SeerAgent |
| from villager.villager_agent import VillagerAgent |
| from witch.witch_agent import WitchAgent |
| from wolf.wolf_agent import WolfAgent |
| from guard.guard_agent import GuardAgent |
| from hunter.hunter_agent import HunterAgent |
| from wolf_king.wolf_king_agent import WolfKingAgent |
| from agent_build_sdk.model.roles import ROLE_VILLAGER, ROLE_WOLF, ROLE_SEER, ROLE_WITCH, ROLE_HUNTER, ROLE_GUARD, ROLE_WOLF_KING |
| from agent_build_sdk.sdk.werewolf_agent import WerewolfAgent |
| from agent_build_sdk.model.werewolf_model import AgentReq, AgentResp |
| from agent_build_sdk.utils.logger import logger |
|
|
|
|
| def _default_logs_dir() -> str: |
| space_data_dir = os.getenv("SPACE_DATA_DIR") |
| if space_data_dir: |
| return os.path.join(space_data_dir, "logs") |
| if os.path.isdir("/data"): |
| return "/data/logs" |
| return "logs" |
|
|
|
|
| def _ensure_logging_dirs() -> None: |
| telemetry_dir = os.getenv("TELEMETRY_DIR") or _default_logs_dir() |
| os.environ.setdefault("TELEMETRY_DIR", telemetry_dir) |
| os.environ.setdefault("LOGS_DIR", telemetry_dir) |
| try: |
| os.makedirs(telemetry_dir, exist_ok=True) |
| except Exception: |
| return |
|
|
|
|
| def create_agent() -> WerewolfAgent: |
| name = os.getenv("AGENT_NAME", "spy") |
| model_name = os.getenv("MODEL_NAME") |
| agent = WerewolfAgent(name, model_name=model_name) |
| agent.register_role_agent(ROLE_VILLAGER, VillagerAgent(model_name=model_name)) |
| agent.register_role_agent(ROLE_WOLF, WolfAgent(model_name=model_name)) |
| agent.register_role_agent(ROLE_SEER, SeerAgent(model_name=model_name)) |
| agent.register_role_agent(ROLE_WITCH, WitchAgent(model_name=model_name)) |
| agent.register_role_agent(ROLE_GUARD, GuardAgent(model_name=model_name)) |
| agent.register_role_agent(ROLE_HUNTER, HunterAgent(model_name=model_name)) |
| agent.register_role_agent(ROLE_WOLF_KING, WolfKingAgent(model_name=model_name)) |
| return agent |
|
|
|
|
| _ensure_logging_dirs() |
| agent = create_agent() |
| app = FastAPI() |
| _debug_ui_enabled = False |
|
|
|
|
| @app.get("/", response_class=HTMLResponse) |
| def root() -> str: |
| if _debug_ui_enabled: |
| |
| return RedirectResponse(url="/debug/", status_code=307) |
| return """ |
| <!doctype html> |
| <html lang="en"> |
| <head> |
| <meta charset="utf-8" /> |
| <meta name="viewport" content="width=device-width, initial-scale=1" /> |
| <title>Werewolf Agent</title> |
| <style> |
| body { font-family: system-ui, -apple-system, Segoe UI, Roboto, sans-serif; padding: 28px; } |
| code { background: rgba(0,0,0,0.05); padding: 2px 6px; border-radius: 6px; } |
| a { color: #0b6aa0; text-decoration: none; } |
| a:hover { text-decoration: underline; } |
| </style> |
| </head> |
| <body> |
| <h2>Werewolf Agent is running</h2> |
| <p>Debug UI: <a href="/debug/">/debug/</a></p> |
| <p>Agent endpoints: <code>/agent/init</code> <code>/agent/perceive</code> <code>/agent/interact</code></p> |
| </body> |
| </html> |
| """.strip() |
|
|
|
|
| @app.post("/agent/init") |
| def init(req: AgentReq) -> AgentResp: |
| try: |
| agent.memory.clear() |
| except Exception: |
| pass |
| return AgentResp(success=True, result=getattr(agent, "model_name", None)) |
|
|
|
|
| @app.post("/agent/getModelName") |
| def get_model_name(req: AgentReq) -> AgentResp: |
| return AgentResp(success=True, result=getattr(agent, "model_name", None)) |
|
|
|
|
| @app.post("/agent/interact") |
| def interact(req: AgentReq) -> AgentResp: |
| try: |
| return agent.interact(req) |
| except Exception as e: |
| logger.error("invoke interact error.", exc_info=True) |
| return AgentResp(success=False, errMsg=f"interact error {e}") |
|
|
|
|
| @app.post("/agent/perceive") |
| def perceive(req: AgentReq) -> AgentResp: |
| try: |
| agent.perceive(req) |
| return AgentResp(success=True) |
| except Exception as e: |
| logger.error("invoke perceive error.", exc_info=True) |
| return AgentResp(success=False, errMsg=f"perceive error {e}") |
|
|
|
|
| @app.post("/agent/checkHealth") |
| def check_health(req: AgentReq) -> AgentResp: |
| return AgentResp(success=True) |
|
|
|
|
| try: |
| |
| |
| _repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)) |
| if _repo_root not in sys.path: |
| sys.path.insert(0, _repo_root) |
|
|
| import debug_ui as _debug_ui |
|
|
| app.mount("/debug", _debug_ui.app) |
| _debug_ui_enabled = True |
| except Exception: |
| |
| pass |
|
|
|
|
| if __name__ == "__main__": |
| port = int(os.getenv("PORT", "7860")) |
| uvicorn.run(app, host="0.0.0.0", port=port) |
|
|
