refactor: standardize type hints and improve null safety across codebase

- Update type annotations to use modern `dict[str, Any]` syntax instead of bare `dict`
- Add `_require_*` helper methods in main.py and browser/client.py for safe attribute access
- Fix category parameter validation in remember tool to ensure string type
- Refactor LLM client response handling to use standardized content block format
- Fix bug in browser_action where local variable wasn't used
- Fix missing brackets in tool result logging
- Remove unused imports throughout (datetime, os, subprocess, Path, logger)
- Update category parameter type hints to use `str | None = None` syntax

TODO.md

@@ -2,18 +2,6 @@
 
 ## 优先级排序
 
-### 1. 提示词工程（最高优先级）⭐
-**状态**: 已添加"诚实原则"，需要验证和优化
-**问题**: 从运行日志看，数字生命还在编造结果（假装成功、实际失败）
-**需要**:
-- 验证"诚实原则"是否生效
-- 可能需要更强的提示或示例
-- 可能需要添加"验证工具"来强制检查结果
-
-**成功标准**:
-- 浏览器崩溃时，它如实说"失败了"
-- 插件加载失败时，不说"成功创建"
-- API调用失败时，说"无法获取"，不编造数据
 
 ### 2. 插件系统完善
 **状态**: 框架已有，但有问题
@@ -56,12 +44,6 @@
 **描述**: `receive_stimulus()` 需要一个入口
 **可能方案**: 前端控制台的"发送刺激"功能
 
-### 5. 清理错误记忆
-**状态**: 待办
-**描述**: 数字生命之前存储了很多"虚假成功"的记忆
-**需要**: 清理或标记这些记忆为"不可靠"
-
----
 
 ## 已完成
 

agent/agent_loop.py

@@ -7,12 +7,12 @@ from pathlib import Path
 from loguru import logger
 
 from agent.context_manager import ContextManager
-from agent.llm_client import BaseLLMClient, create_client
+from agent.llm_client import BaseLLMClient
 from tools.bash_tool import run_bash
 from tools.file_tools import run_read, run_write, run_edit
 from tools.code_executor import run_create_plugin
 from tools.browser_tool import run_browser
-from tools.memory_tools import run_remember, run_recall, run_journal
+from tools.memory_tools import run_recall, run_remember
 
 
 # Tool definitions sent to the API
@@ -246,9 +246,12 @@ class AgentLoop:
                     args["path"], args["old_text"], args["new_text"], workdir=self.workdir
                 )
             elif name == "remember":
+                category = args.get("category")
+                if category is not None and not isinstance(category, str):
+                    return "Error: category must be a string"
                 return await run_remember(
                     args["content"],
-                    category=args.get("category"),
+                    category=category,
                 )
             elif name == "recall":
                 return await run_recall(args["query"])

agent/context_manager.py

@@ -2,9 +2,9 @@
 
 from __future__ import annotations
 
-import anthropic
 from loguru import logger
 
+from agent.llm_client import BaseLLMClient
 from utils.token_counter import estimate_messages_tokens
 
 
@@ -13,7 +13,7 @@ class ContextManager:
 
     def __init__(
         self,
-        client: anthropic.AsyncAnthropic,
+        client: BaseLLMClient,
         model: str,
         context_window: int = 200000,
         compact_threshold: float = 0.9,
@@ -134,10 +134,9 @@ class ContextManager:
             conversation_text = conversation_text[:100000] + "\n\n... (truncated)"
 
         try:
-            response = await self.client.messages.create(
+            response = await self.client.create_message(
                 model=self.model,
-                max_tokens=4000,
-                system=(
+                system_prompt=(
                     "You are a conversation summarizer. Create a concise but comprehensive "
                     "summary of the following conversation. Preserve:\n"
                     "- Key decisions made\n"
@@ -152,14 +151,20 @@ class ContextManager:
                     {
                         "role": "user",
                         "content": f"Summarize this conversation:\n\n{conversation_text}",
-                    }
+                    },
                 ],
+                tools=[],
+                max_tokens=4000,
             )
 
-            summary = ""
-            for block in response.content:
-                if hasattr(block, "text"):
-                    summary += block.text
+            summary_parts: list[str] = []
+            for block in response.content_blocks:
+                if isinstance(block, dict) and block.get("type") == "text":
+                    text = block.get("text")
+                    if isinstance(text, str):
+                        summary_parts.append(text)
+
+            summary = "".join(summary_parts)
 
             return summary or "No summary generated."
         except Exception as e:

agent/llm_client.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import os
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, cast
 
 from loguru import logger
 
@@ -17,8 +17,8 @@ class BaseLLMClient(ABC):
         self,
         model: str,
         system_prompt: str,
-        messages: list[dict],
-        tools: list[dict],
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
         max_tokens: int,
     ) -> LLMResponse:
         """Create a message and return standardized response."""
@@ -30,7 +30,7 @@ class LLMResponse:
 
     def __init__(
         self,
-        content_blocks: list[dict],
+        content_blocks: list[dict[str, Any]],
         stop_reason: str,
         tool_calls: list[ToolCall] | None = None,
     ):
@@ -42,7 +42,7 @@ class LLMResponse:
 class ToolCall:
     """Standardized tool call."""
 
-    def __init__(self, id: str, name: str, input: dict):
+    def __init__(self, id: str, name: str, input: dict[str, Any]):
         self.id = id
         self.name = name
         self.input = input
@@ -59,16 +59,16 @@ class AnthropicClient(BaseLLMClient):
         self,
         model: str,
         system_prompt: str,
-        messages: list[dict],
-        tools: list[dict],
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
         max_tokens: int,
     ) -> LLMResponse:
         """Create message using Anthropic API."""
         response = await self._client.messages.create(
             model=model,
             system=system_prompt,
-            messages=messages,
-            tools=tools,
+            messages=cast(Any, messages),
+            tools=cast(Any, tools),
             max_tokens=max_tokens,
         )
 
@@ -92,7 +92,7 @@ class AnthropicClient(BaseLLMClient):
 
         return LLMResponse(
             content_blocks=content_blocks,
-            stop_reason=response.stop_reason,
+            stop_reason=str(response.stop_reason or "end_turn"),
             tool_calls=tool_calls,
         )
 
@@ -111,8 +111,8 @@ class OpenAIClient(BaseLLMClient):
         self,
         model: str,
         system_prompt: str,
-        messages: list[dict],
-        tools: list[dict],
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
         max_tokens: int,
     ) -> LLMResponse:
         """Create message using OpenAI-compatible API."""
@@ -120,7 +120,9 @@ class OpenAIClient(BaseLLMClient):
         openai_tools = self._convert_tools(tools)
 
         # Build message list with system prompt
-        all_messages = [{"role": "system", "content": system_prompt}]
+        all_messages: list[dict[str, Any]] = [
+            {"role": "system", "content": system_prompt}
+        ]
 
         # Convert messages
         for msg in messages:
@@ -130,8 +132,8 @@ class OpenAIClient(BaseLLMClient):
 
         response = await self._client.chat.completions.create(
             model=model,
-            messages=all_messages,
-            tools=openai_tools if openai_tools else None,
+            messages=cast(Any, all_messages),
+            tools=cast(Any, openai_tools if openai_tools else None),
             max_tokens=max_tokens,
         )
 
@@ -147,16 +149,20 @@ class OpenAIClient(BaseLLMClient):
         if choice.message.tool_calls:
             for tc in choice.message.tool_calls:
                 import json
+                if not hasattr(tc, "function"):
+                    continue
+                tc_any = cast(Any, tc)
+                function = tc_any.function
                 tool_calls.append(ToolCall(
-                    id=tc.id,
-                    name=tc.function.name,
-                    input=json.loads(tc.function.arguments) if tc.function.arguments else {},
+                    id=tc_any.id,
+                    name=function.name,
+                    input=json.loads(function.arguments) if function.arguments else {},
                 ))
                 content_blocks.append({
                     "type": "tool_use",
-                    "id": tc.id,
-                    "name": tc.function.name,
-                    "input": json.loads(tc.function.arguments) if tc.function.arguments else {},
+                    "id": tc_any.id,
+                    "name": function.name,
+                    "input": json.loads(function.arguments) if function.arguments else {},
                 })
 
         stop_reason = "tool_use" if tool_calls else "end_turn"
@@ -169,7 +175,7 @@ class OpenAIClient(BaseLLMClient):
             tool_calls=tool_calls,
         )
 
-    def _convert_tools(self, anthropic_tools: list[dict]) -> list[dict]:
+    def _convert_tools(self, anthropic_tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
         """Convert Anthropic-style tools to OpenAI format."""
         openai_tools = []
         for tool in anthropic_tools:
@@ -183,7 +189,7 @@ class OpenAIClient(BaseLLMClient):
             })
         return openai_tools
 
-    def _convert_message(self, msg: dict) -> dict | None:
+    def _convert_message(self, msg: dict[str, Any]) -> dict[str, Any] | None:
         """Convert Anthropic-style message to OpenAI format."""
         role = msg.get("role")
         content = msg.get("content")
@@ -204,7 +210,7 @@ class OpenAIClient(BaseLLMClient):
 
                 if tool_results:
                     # Convert tool results to OpenAI format
-                    result_msg = {"role": "tool", "content": ""}
+                    result_msg: dict[str, Any] = {"role": "tool", "content": ""}
                     for tr in tool_results:
                         result_msg["tool_call_id"] = tr.get("tool_use_id", "")
                         result_msg["content"] = tr.get("content", "")
@@ -225,7 +231,7 @@ class OpenAIClient(BaseLLMClient):
                         elif block.get("type") == "tool_use":
                             tool_uses.append(block)
 
-                result = {"role": "assistant"}
+                result: dict[str, Any] = {"role": "assistant"}
                 if texts:
                     result["content"] = "\n".join(texts)
                 if tool_uses:

agent/message_history.py

@@ -1,7 +1,6 @@
 """Message history tracking and persistence."""
 
 import json
-from datetime import datetime
 from pathlib import Path
 
 import aiofiles

app.py

@@ -1,7 +1,6 @@
 """Entelechy Control Console - 观察和控制数字生命的控制台"""
 
 import asyncio
-import os
 import subprocess
 import threading
 from pathlib import Path
@@ -69,7 +68,7 @@ def get_thinking_stream() -> str:
                         inputs = block.get("input", {})
                         texts.append(f"[调用工具: {name}({inputs})]")
                     elif block.get("type") == "tool_result":
-                        texts.append(f"[工具结果]")
+                        texts.append("[工具结果]")
             content = "\n".join(texts)
 
         output.append(f"**{role}**: {content}\n")
@@ -154,20 +153,21 @@ def send_stimulus(stimulus_type: str, content: str) -> str:
 def browser_action(action: str, url: str = "", selector: str = "", text: str = "") -> str:
     """浏览器控制"""
     life = _get_life()
-    if life.browser_client is None:
+    browser_client = life.browser_client
+    if browser_client is None:
         return "浏览器客户端未初始化"
 
     async def _action():
         if action == "navigate":
-            return await life.browser_client.navigate(url)
+            return await browser_client.navigate(url)
         elif action == "click":
-            return await life.browser_client.click(selector)
+            return await browser_client.click(selector)
         elif action == "type":
-            return await life.browser_client.type_text(selector, text)
+            return await browser_client.type_text(selector, text)
         elif action == "screenshot":
-            return await life.browser_client.screenshot()
+            return await browser_client.screenshot()
         elif action == "extract":
-            return await life.browser_client.extract_content()
+            return await browser_client.extract_content()
         else:
             return f"未知操作: {action}"
 
@@ -196,7 +196,7 @@ def _start_background_life():
 
 # ========== 构建控制台界面 ==========
 
-with gr.Blocks(title="Entelechy 控制台", theme=gr.themes.Soft()) as demo:
+with gr.Blocks(title="Entelechy 控制台") as demo:
     gr.Markdown("# 🧬 Entelechy - 数字生命控制台")
     gr.Markdown("这是观察和控制数字生命的控制台，不是聊天机器人")
 

browser/client.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import base64
 from pathlib import Path
+from typing import Any
 
 from loguru import logger
 
@@ -22,6 +23,16 @@ class BrowserClient:
         self.page = None
         self._started = False
 
+    def _require_page(self) -> Any:
+        if self.page is None:
+            raise RuntimeError("Browser page is not initialized")
+        return self.page
+
+    def _require_context(self) -> Any:
+        if self.context is None:
+            raise RuntimeError("Browser context is not initialized")
+        return self.context
+
     async def start(self):
         """Launch the browser and restore session state."""
         if self._started:
@@ -78,9 +89,11 @@ class BrowserClient:
         """
         await self.ensure_started()
         try:
-            await self.page.goto(url, wait_until="domcontentloaded", timeout=30000)
-            title = await self.page.title()
-            await self.session_manager.save_state(self.context)
+            page = self._require_page()
+            context = self._require_context()
+            await page.goto(url, wait_until="domcontentloaded", timeout=30000)
+            title = await page.title()
+            await self.session_manager.save_state(context)
             return f"Navigated to: {url}\nTitle: {title}"
         except Exception as e:
             return f"Navigation error: {e}"
@@ -89,8 +102,9 @@ class BrowserClient:
         """Click an element."""
         await self.ensure_started()
         try:
-            await self.page.click(selector, timeout=10000)
-            await self.page.wait_for_load_state("domcontentloaded")
+            page = self._require_page()
+            await page.click(selector, timeout=10000)
+            await page.wait_for_load_state("domcontentloaded")
             return f"Clicked: {selector}"
         except Exception as e:
             return f"Click error: {e}"
@@ -99,7 +113,8 @@ class BrowserClient:
         """Type text into an element."""
         await self.ensure_started()
         try:
-            await self.page.fill(selector, text, timeout=10000)
+            page = self._require_page()
+            await page.fill(selector, text, timeout=10000)
             return f"Typed text into: {selector}"
         except Exception as e:
             return f"Type error: {e}"
@@ -112,7 +127,8 @@ class BrowserClient:
         """
         await self.ensure_started()
         try:
-            raw = await self.page.screenshot(full_page=full_page)
+            page = self._require_page()
+            raw = await page.screenshot(full_page=full_page)
             encoded = base64.b64encode(raw).decode("utf-8")
             return f"Screenshot taken ({len(raw)} bytes). Base64: {encoded[:100]}..."
         except Exception as e:
@@ -122,9 +138,10 @@ class BrowserClient:
         """Extract text content from the current page."""
         await self.ensure_started()
         try:
-            title = await self.page.title()
-            url = self.page.url
-            text = await self.page.inner_text("body")
+            page = self._require_page()
+            title = await page.title()
+            url = page.url
+            text = await page.inner_text("body")
             # Truncate very long pages
             if len(text) > 50000:
                 text = text[:50000] + "\n\n... (truncated)"

main.py

@@ -60,6 +60,21 @@ class DigitalLife:
         self.agent: AgentLoop | None = None
         self.history: MessageHistory | None = None
 
+    def _require_memory_manager(self) -> MemoryManager:
+        if self.memory_manager is None:
+            raise RuntimeError("Memory manager is not initialized")
+        return self.memory_manager
+
+    def _require_agent(self) -> AgentLoop:
+        if self.agent is None:
+            raise RuntimeError("Agent loop is not initialized")
+        return self.agent
+
+    def _require_history(self) -> MessageHistory:
+        if self.history is None:
+            raise RuntimeError("Message history is not initialized")
+        return self.history
+
     async def _initialize(self):
         """Initialize all components."""
         # Ensure directories exist
@@ -139,7 +154,8 @@ class DigitalLife:
 
     async def _get_core_context(self) -> str:
         """Get CORE.md content for every LLM call."""
-        core = await self.memory_manager.load_core()
+        memory_manager = self._require_memory_manager()
+        core = await memory_manager.load_core()
         if core:
             return f"\n=== 核心记忆 ===\n{core}\n================\n"
         return ""
@@ -148,7 +164,11 @@ class DigitalLife:
         """Wake up: load CORE.md and restore self-awareness."""
         logger.info("Waking up...")
 
-        core_memories = await self.memory_manager.load_core()
+        memory_manager = self._require_memory_manager()
+        history = self._require_history()
+        agent = self._require_agent()
+
+        core_memories = await memory_manager.load_core()
 
         wake_up_parts = ["你醒来了。\n"]
 
@@ -160,16 +180,16 @@ class DigitalLife:
         wake_up_content = "\n".join(wake_up_parts)
 
         # If we have saved history, continue from it; otherwise start fresh
-        if self.history.messages:
-            logger.info(f"Resuming from {len(self.history.messages)} saved messages")
-            self.history.append({"role": "user", "content": wake_up_content})
+        if history.messages:
+            logger.info(f"Resuming from {len(history.messages)} saved messages")
+            history.append({"role": "user", "content": wake_up_content})
         else:
-            self.history.set_messages([{"role": "user", "content": wake_up_content}])
+            history.set_messages([{"role": "user", "content": wake_up_content}])
 
         # Run the wake-up conversation
-        messages = await self.agent.run(self.history.get_messages())
-        self.history.set_messages(messages)
-        await self.history.save()
+        messages = await agent.run(history.get_messages())
+        history.set_messages(messages)
+        await history.save()
 
         logger.info("Wake up complete")
 
@@ -178,6 +198,9 @@ class DigitalLife:
         await self._initialize()
         await self._wake_up()
 
+        history = self._require_history()
+        agent = self._require_agent()
+
         while self.alive:
             try:
                 # Check for external stimulus (non-blocking, immediate)
@@ -187,23 +210,23 @@ class DigitalLife:
 
                 if stimulus:
                     # External stimulus received
-                    self.history.append({
+                    history.append({
                         "role": "user",
                         "content": f"[感知] {stimulus['type']}: {stimulus['content']}",
                     })
                 else:
                     # No stimulus - continue autonomous operation
-                    self.history.append({
+                    history.append({
                         "role": "user",
                         "content": "继续。",
                     })
 
                 # Run agent loop
-                messages = await self.agent.run(self.history.get_messages())
-                self.history.set_messages(messages)
+                messages = await agent.run(history.get_messages())
+                history.set_messages(messages)
 
                 # Persist history periodically
-                await self.history.save()
+                await history.save()
 
                 # Immediately continue to next iteration (no waiting)
 
@@ -238,10 +261,13 @@ class DigitalLife:
         if self.agent is None:
             await self._initialize()
 
-        self.history.append({"role": "user", "content": message})
-        messages = await self.agent.run(self.history.get_messages())
-        self.history.set_messages(messages)
-        await self.history.save()
+        history = self._require_history()
+        agent = self._require_agent()
+
+        history.append({"role": "user", "content": message})
+        messages = await agent.run(history.get_messages())
+        history.set_messages(messages)
+        await history.save()
 
         # Extract the last assistant text
         for msg in reversed(messages):

memory/manager.py

@@ -13,7 +13,7 @@ class MemoryManager:
         self.storage = MemoryStorage(base_path)
         self.retrieval = MemoryRetrieval(base_path)
 
-    async def remember(self, content: str, category: str = None) -> str:
+    async def remember(self, content: str, category: str | None = None) -> str:
         """Store a new memory.
 
         Args:

memory/storage.py

@@ -31,7 +31,7 @@ class MemoryStorage:
                 "\n"
             )
 
-    async def store(self, content: str, category: str = None) -> Path:
+    async def store(self, content: str, category: str | None = None) -> Path:
         """Store a memory as a markdown file with frontmatter.
 
         Args:

tools/bash_tool.py

@@ -1,7 +1,6 @@
 """Bash tool - execute shell commands with safety checks."""
 
 import asyncio
-import subprocess
 from pathlib import Path
 
 from loguru import logger
@@ -37,6 +36,7 @@ async def run_bash(command: str, workdir: Path | None = None, timeout: int = 120
 
     logger.debug(f"Executing: {command}")
 
+    proc: asyncio.subprocess.Process | None = None
     try:
         proc = await asyncio.create_subprocess_shell(
             command,
@@ -56,10 +56,11 @@ async def run_bash(command: str, workdir: Path | None = None, timeout: int = 120
 
         return output
     except asyncio.TimeoutError:
-        try:
-            proc.kill()
-        except Exception:
-            pass
+        if proc is not None:
+            try:
+                proc.kill()
+            except Exception:
+                pass
         return f"Error: Command timed out after {timeout}s"
     except Exception as e:
         return f"Error: {e}"

tools/code_executor.py

@@ -1,6 +1,5 @@
 """Code executor - allows the AI to create new plugins at runtime."""
 
-from pathlib import Path
 
 import yaml
 from loguru import logger

tools/file_tools.py

@@ -3,7 +3,6 @@
 import aiofiles
 from pathlib import Path
 
-from loguru import logger
 
 from utils.path_utils import safe_path
 

tools/memory_tools.py

@@ -19,7 +19,7 @@ def _get_manager() -> MemoryManager:
     return _memory_manager
 
 
-async def run_remember(content: str, category: str = None) -> str:
+async def run_remember(content: str, category: str | None = None) -> str:
     """Store information to long-term memory.
 
     Args:

utils/token_counter.py

@@ -1,5 +1,7 @@
 """Token counting utilities using tiktoken."""
 
+from typing import Any
+
 import tiktoken
 
 
@@ -18,7 +20,7 @@ def estimate_tokens(text: str) -> int:
     return len(_get_encoder().encode(text))
 
 
-def estimate_messages_tokens(messages: list[dict]) -> int:
+def estimate_messages_tokens(messages: list[dict[str, Any]]) -> int:
     """Estimate total tokens across a list of messages.
 
     Each message contributes its content tokens plus a small overhead