fix: windows path and encoding adaptation

This commit is contained in:
zhayujie
2026-02-06 18:37:05 +08:00
parent bea95d4fae
commit 051ffd78a3
20 changed files with 87 additions and 26 deletions

View File

@@ -11,12 +11,18 @@ from typing import Optional, List
from pathlib import Path
def _default_workspace():
"""Get default workspace path with proper Windows support"""
from common.utils import expand_path
return expand_path("~/cow")
@dataclass
class MemoryConfig:
"""Configuration for memory storage and search"""
# Storage paths (default: ~/cow)
workspace_root: str = field(default_factory=lambda: os.path.expanduser("~/cow"))
workspace_root: str = field(default_factory=_default_workspace)
# Embedding config
embedding_provider: str = "openai" # "openai" | "local"

View File

@@ -304,7 +304,7 @@ class MemoryManager:
):
"""Sync a single file"""
# Compute file hash
content = file_path.read_text()
content = file_path.read_text(encoding='utf-8')
file_hash = MemoryStorage.compute_hash(content)
# Get relative path

View File

@@ -140,7 +140,9 @@ class Agent:
if self.runtime_info.get("model"):
runtime_parts.append(f"模型={self.runtime_info['model']}")
if self.runtime_info.get("workspace"):
runtime_parts.append(f"工作空间={self.runtime_info['workspace']}")
# Replace backslashes with forward slashes for Windows paths
workspace_path = str(self.runtime_info['workspace']).replace('\\', '/')
runtime_parts.append(f"工作空间={workspace_path}")
if self.runtime_info.get("channel") and self.runtime_info.get("channel") != "web":
runtime_parts.append(f"渠道={self.runtime_info['channel']}")

View File

@@ -11,6 +11,7 @@ from typing import Dict, Any
from agent.tools.base_tool import BaseTool, ToolResult
from agent.tools.utils.truncate import truncate_tail, format_size, DEFAULT_MAX_LINES, DEFAULT_MAX_BYTES
from common.log import logger
from common.utils import expand_path
class Bash(BaseTool):
@@ -80,7 +81,7 @@ IMPORTANT SAFETY GUIDELINES:
env = os.environ.copy()
# Load environment variables from ~/.cow/.env if it exists
env_file = os.path.expanduser("~/.cow/.env")
env_file = expand_path("~/.cow/.env")
if os.path.exists(env_file):
try:
from dotenv import dotenv_values

View File

@@ -7,6 +7,7 @@ import os
from typing import Dict, Any
from agent.tools.base_tool import BaseTool, ToolResult
from common.utils import expand_path
from agent.tools.utils.diff import (
strip_bom,
detect_line_ending,
@@ -178,7 +179,7 @@ class Edit(BaseTool):
:return: Absolute path
"""
# Expand ~ to user home directory
path = os.path.expanduser(path)
path = expand_path(path)
if os.path.isabs(path):
return path
return os.path.abspath(os.path.join(self.cwd, path))

View File

@@ -9,6 +9,7 @@ from pathlib import Path
from agent.tools.base_tool import BaseTool, ToolResult
from common.log import logger
from common.utils import expand_path
# API Key 知识库:常见的环境变量及其描述
@@ -66,7 +67,7 @@ class EnvConfig(BaseTool):
def __init__(self, config: dict = None):
self.config = config or {}
# Store env config in ~/.cow directory (outside workspace for security)
self.env_dir = os.path.expanduser("~/.cow")
self.env_dir = expand_path("~/.cow")
self.env_path = os.path.join(self.env_dir, '.env')
self.agent_bridge = self.config.get("agent_bridge") # Reference to AgentBridge for hot reload
# Don't create .env file in __init__ to avoid issues during tool discovery

View File

@@ -7,6 +7,7 @@ from typing import Dict, Any
from agent.tools.base_tool import BaseTool, ToolResult
from agent.tools.utils.truncate import truncate_head, format_size, DEFAULT_MAX_BYTES
from common.utils import expand_path
DEFAULT_LIMIT = 500
@@ -51,7 +52,7 @@ class Ls(BaseTool):
absolute_path = self._resolve_path(path)
# Security check: Prevent accessing sensitive config directory
env_config_dir = os.path.expanduser("~/.cow")
env_config_dir = expand_path("~/.cow")
if os.path.abspath(absolute_path) == os.path.abspath(env_config_dir):
return ToolResult.fail(
"Error: Access denied. API keys and credentials must be accessed through the env_config tool only."
@@ -133,7 +134,7 @@ class Ls(BaseTool):
def _resolve_path(self, path: str) -> str:
"""Resolve path to absolute path"""
# Expand ~ to user home directory
path = os.path.expanduser(path)
path = expand_path(path)
if os.path.isabs(path):
return path
return os.path.abspath(os.path.join(self.cwd, path))

View File

@@ -77,7 +77,7 @@ class MemoryGetTool(BaseTool):
if not file_path.exists():
return ToolResult.fail(f"Error: File not found: {path}")
content = file_path.read_text()
content = file_path.read_text(encoding='utf-8')
lines = content.split('\n')
# Handle line range

View File

@@ -9,6 +9,7 @@ from pathlib import Path
from agent.tools.base_tool import BaseTool, ToolResult
from agent.tools.utils.truncate import truncate_head, format_size, DEFAULT_MAX_LINES, DEFAULT_MAX_BYTES
from common.utils import expand_path
class Read(BaseTool):
@@ -77,7 +78,7 @@ class Read(BaseTool):
absolute_path = self._resolve_path(path)
# Security check: Prevent reading sensitive config files
env_config_path = os.path.expanduser("~/.cow/.env")
env_config_path = expand_path("~/.cow/.env")
if os.path.abspath(absolute_path) == os.path.abspath(env_config_path):
return ToolResult.fail(
"Error: Access denied. API keys and credentials must be accessed through the env_config tool only."
@@ -129,7 +130,7 @@ class Read(BaseTool):
:return: Absolute path
"""
# Expand ~ to user home directory
path = os.path.expanduser(path)
path = expand_path(path)
if os.path.isabs(path):
return path
return os.path.abspath(os.path.join(self.cwd, path))

View File

@@ -6,6 +6,7 @@ import os
from typing import Optional
from config import conf
from common.log import logger
from common.utils import expand_path
from bridge.context import Context, ContextType
from bridge.reply import Reply, ReplyType
@@ -31,7 +32,7 @@ def init_scheduler(agent_bridge) -> bool:
from agent.tools.scheduler.scheduler_service import SchedulerService
# Get workspace from config
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
store_path = os.path.join(workspace_root, "scheduler", "tasks.json")
# Create task store

View File

@@ -8,6 +8,7 @@ import threading
from datetime import datetime
from typing import Dict, List, Optional
from pathlib import Path
from common.utils import expand_path
class TaskStore:
@@ -24,7 +25,7 @@ class TaskStore:
"""
if store_path is None:
# Default to ~/cow/scheduler/tasks.json
home = os.path.expanduser("~")
home = expand_path("~")
store_path = os.path.join(home, "cow", "scheduler", "tasks.json")
self.store_path = store_path

View File

@@ -7,6 +7,7 @@ from typing import Dict, Any
from pathlib import Path
from agent.tools.base_tool import BaseTool, ToolResult
from common.utils import expand_path
class Send(BaseTool):
@@ -102,7 +103,7 @@ class Send(BaseTool):
def _resolve_path(self, path: str) -> str:
"""Resolve path to absolute path"""
path = os.path.expanduser(path)
path = expand_path(path)
if os.path.isabs(path):
return path
return os.path.abspath(os.path.join(self.cwd, path))

View File

@@ -8,6 +8,7 @@ from typing import Dict, Any
from pathlib import Path
from agent.tools.base_tool import BaseTool, ToolResult
from common.utils import expand_path
class Write(BaseTool):
@@ -90,7 +91,7 @@ class Write(BaseTool):
:return: Absolute path
"""
# Expand ~ to user home directory
path = os.path.expanduser(path)
path = expand_path(path)
if os.path.isabs(path):
return path
return os.path.abspath(os.path.join(self.cwd, path))

View File

@@ -13,6 +13,7 @@ from bridge.context import Context
from bridge.reply import Reply, ReplyType
from common import const
from common.log import logger
from common.utils import expand_path
from models.openai_compatible_bot import OpenAICompatibleBot
@@ -421,7 +422,7 @@ class AgentBridge:
}
# Use fixed secure location for .env file
env_file = os.path.expanduser("~/.cow/.env")
env_file = expand_path("~/.cow/.env")
# Read existing env vars from .env file
existing_env_vars = {}
@@ -504,7 +505,7 @@ class AgentBridge:
from config import conf
# Reload environment variables from .env file
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
env_file = os.path.join(workspace_root, '.env')
if os.path.exists(env_file):

View File

@@ -11,6 +11,7 @@ from typing import Optional, List
from agent.protocol import Agent
from agent.tools import ToolManager
from common.log import logger
from common.utils import expand_path
class AgentInitializer:
@@ -46,7 +47,7 @@ class AgentInitializer:
from config import conf
# Get workspace from config
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
# Migrate API keys
self._migrate_config_to_env(workspace_root)
@@ -122,7 +123,7 @@ class AgentInitializer:
def _load_env_file(self):
"""Load environment variables from .env file"""
env_file = os.path.expanduser("~/.cow/.env")
env_file = expand_path("~/.cow/.env")
if os.path.exists(env_file):
try:
from dotenv import load_dotenv
@@ -338,7 +339,7 @@ class AgentInitializer:
"linkai_api_key": "LINKAI_API_KEY",
}
env_file = os.path.expanduser("~/.cow/.env")
env_file = expand_path("~/.cow/.env")
# Read existing env vars
existing_env_vars = {}

View File

@@ -21,6 +21,7 @@ from dingtalk_stream.card_replier import CardReplier
from bridge.context import Context, ContextType
from bridge.reply import Reply, ReplyType
from channel.chat_channel import ChatChannel
from common.utils import expand_path
from channel.dingtalk.dingtalk_message import DingTalkMessage
from common.expired_dict import ExpiredDict
from common.log import logger
@@ -276,7 +277,7 @@ class DingTalkChanel(ChatChannel, dingtalk_stream.ChatbotHandler):
# 保存到临时文件
file_name = os.path.basename(file_path) or f"media_{uuid.uuid4()}"
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
tmp_dir = os.path.join(workspace_root, "tmp")
os.makedirs(tmp_dir, exist_ok=True)
temp_file = os.path.join(tmp_dir, file_name)

View File

@@ -9,6 +9,7 @@ from channel.chat_message import ChatMessage
# -*- coding=utf-8 -*-
from common.log import logger
from common.tmp_dir import TmpDir
from common.utils import expand_path
from config import conf
@@ -49,7 +50,7 @@ class DingTalkMessage(ChatMessage):
download_url = image_download_handler.get_image_download_url(download_code)
# 下载到工作空间 tmp 目录
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
tmp_dir = os.path.join(workspace_root, "tmp")
os.makedirs(tmp_dir, exist_ok=True)
@@ -67,7 +68,7 @@ class DingTalkMessage(ChatMessage):
self.ctype = ContextType.TEXT
# 下载到工作空间 tmp 目录
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
tmp_dir = os.path.join(workspace_root, "tmp")
os.makedirs(tmp_dir, exist_ok=True)

View File

@@ -6,6 +6,7 @@ import requests
from common.log import logger
from common.tmp_dir import TmpDir
from common import utils
from common.utils import expand_path
from config import conf
@@ -31,7 +32,7 @@ class FeishuMessage(ChatMessage):
image_key = content.get("image_key")
# 下载图片到工作空间临时目录
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
tmp_dir = os.path.join(workspace_root, "tmp")
os.makedirs(tmp_dir, exist_ok=True)
image_path = os.path.join(tmp_dir, f"{image_key}.png")
@@ -97,7 +98,7 @@ class FeishuMessage(ChatMessage):
if image_keys:
# 如果包含图片,下载并在文本中引用本地路径
workspace_root = os.path.expanduser(conf().get("agent_workspace", "~/cow"))
workspace_root = expand_path(conf().get("agent_workspace", "~/cow"))
tmp_dir = os.path.join(workspace_root, "tmp")
os.makedirs(tmp_dir, exist_ok=True)

View File

@@ -76,3 +76,42 @@ def remove_markdown_symbol(text: str):
if not text:
return text
return re.sub(r'\*\*(.*?)\*\*', r'\1', text)
def expand_path(path: str) -> str:
"""
Expand user path with proper Windows support.
On Windows, os.path.expanduser('~') may not work properly in some shells (like PowerShell).
This function provides a more robust path expansion.
Args:
path: Path string that may contain ~
Returns:
Expanded absolute path
"""
if not path:
return path
# Try standard expansion first
expanded = os.path.expanduser(path)
# If expansion didn't work (path still starts with ~), use HOME or USERPROFILE
if expanded.startswith('~'):
import platform
if platform.system() == 'Windows':
# On Windows, try USERPROFILE first, then HOME
home = os.environ.get('USERPROFILE') or os.environ.get('HOME')
else:
# On Unix-like systems, use HOME
home = os.environ.get('HOME')
if home:
# Replace ~ with home directory
if path == '~':
expanded = home
elif path.startswith('~/') or path.startswith('~\\'):
expanded = os.path.join(home, path[2:])
return expanded

View File

@@ -82,7 +82,7 @@ Cow项目从简单的聊天机器人全面升级为超级智能助理 **CowAgent
#### 3.2 搜索和图像识别
- **搜索技能:** 系统内置实现了 `bocha-search`(博查搜索)的Skill依赖环境变量 `BOCHA_SEARCH_API_KEY`,可在[控制台]()进行创建并发送给Agent完成配置
- **搜索技能:** 系统内置实现了 `bocha-search`(博查搜索)的Skill依赖环境变量 `BOCHA_SEARCH_API_KEY`,可在[控制台](https://open.bochaai.com/)进行创建并发送给Agent完成配置
- **图像识别技能:** 实现了 `openai-image-vision` 插件,可使用 gpt-4.1-mini、gpt-4.1 等图像识别模型。依赖秘钥 `OPENAI_API_KEY`可通过config.json或env_config工具进行维护。
<img width="800" src="https://cdn.link-ai.tech/doc/20260202213219.png">