# 經(jīng)過多次嘗試，我發(fā)現(xiàn)Python 3.10比3.12兼容性更好
conda create -n open_manus python=3.10
conda activate open_manus

# 國內(nèi)網(wǎng)絡(luò)環(huán)境下，使用清華鏡像極大加快了安裝速度
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 我發(fā)現(xiàn)分步安裝能解決大部分依賴問題
pip install --no-deps -r requirements.txt

# 對于playwright，我使用這個方法成功解決了問題
pip install playwright==1.40.0 --no-build-isolation
playwright install chromium

# 如果仍然遇到問題，嘗試手動安裝核心依賴
pip install pydantic==2.5.2 langchain==0.1.6 beautifulsoup4==4.12.3

# 在Windows中處理路徑時，我使用了這種方式統(tǒng)一處理
import os
def normalize_path(path):
    """統(tǒng)一處理Windows和Linux路徑"""
    return os.path.normpath(path).replace('\\', '/')

# DeepSeek V3配置（國內(nèi)最佳選擇）
[llm]
model = "deepseek-v3"
base_url = "https://api.deepseek.com/v1"
api_key = "我的API密鑰"  # 替換為你的實際密鑰
max_tokens = 8192
temperature = 0.0

# 通義千問配置（國產(chǎn)模型備選）
[llm]
model = "qwen-turbo"
base_url = "https://dashscope.aliyuncs.com/api/v1"
api_key = "我的阿里云API密鑰"
max_tokens = 4096
temperature = 0.0

# Claude配置（國外用戶推薦）
[llm]
model = "claude-3-5-sonnet"
base_url = "https://api.anthropic.com"
api_key = "我的Anthropic API密鑰"
max_tokens = 4096
temperature = 0.0

# 我修改了app/llm.py，增加了更詳細(xì)的錯誤處理
def call_api_with_detailed_error(self, *args, **kwargs):
    try:
        return self._call_api(*args, **kwargs)
    except Exception as e:
        error_msg = str(e)
        if "400" in error_msg:
            # 檢查請求參數(shù)
            print("API 400錯誤排查清單:")
            print("1. 檢查API密鑰格式是否正確")
            print("2. 檢查模型名稱是否正確")
            print("3. 檢查請求參數(shù)格式")
            print("4. 原始錯誤信息:", error_msg)
        elif "401" in error_msg:
            print("認(rèn)證失敗，請檢查API密鑰")
        elif "429" in error_msg:
            print("請求頻率過高，請降低請求速度或升級API配額")
        raise e

# 我實現(xiàn)了一個上下文管理器，大大提高了長任務(wù)的完成率
def manage_context_length(context, max_length=6000, summarize_threshold=7500):
    """智能管理上下文長度"""
    if len(context) < summarize_threshold:
        return context
    
    # 我將上下文分為三部分處理
    intro = context[:1500]  # 保留初始指令
    recent = context[-3000:]  # 保留最近交互
    middle = context[1500:-3000]  # 中間部分需要壓縮
    
    # 對中間部分進(jìn)行摘要
    from app.llm import LLM
    llm = LLM()
    summary_prompt = f"請將以下對話歷史壓縮為簡短摘要，保留關(guān)鍵信息：\n\n{middle}"
    summary = llm.generate(summary_prompt, max_tokens=1500)
    
    # 組合處理后的上下文
    new_context = intro + "\n\n[歷史摘要]: " + summary + "\n\n" + recent
    return new_context

# app/tool/bing_search.py
from typing import Dict, List, Optional
import os
import requests
from pydantic import Field
from app.logger import logger
from app.tool.base import BaseTool

class BingSearch(BaseTool):
    """使用必應(yīng)搜索引擎進(jìn)行網(wǎng)絡(luò)搜索"""
    
    name: str = "bing_search"
    description: str = "使用必應(yīng)搜索查詢信息，對于需要最新信息的查詢特別有用"
    
    def __init__(self):
        super().__init__()
        # 從環(huán)境變量或配置文件獲取API密鑰
        self.subscription_key = os.environ.get("BING_API_KEY", "你的Bing搜索API密鑰")
        self.search_url = "https://api.bing.microsoft.com/v7.0/search"
    
    def _call(self, query: str, num_results: int = 10) -> Dict:
        """執(zhí)行必應(yīng)搜索"""
        headers = {"Ocp-Apim-Subscription-Key": self.subscription_key}
        params = {
            "q": query, 
            "count": num_results, 
            "textDecorations": True, 
            "textFormat": "HTML",
            "mkt": "zh-CN"  # 設(shè)置為中文市場，結(jié)果更符合國內(nèi)用戶習(xí)慣
        }
        
        try:
            response = requests.get(self.search_url, headers=headers, params=params)
            response.raise_for_status()
            search_results = response.json()
            
            # 提取有用的搜索結(jié)果
            results = []
            if "webPages" in search_results and "value" in search_results["webPages"]:
                for result in search_results["webPages"]["value"]:
                    results.append({
                        "title": result["name"],
                        "link": result["url"],
                        "snippet": result["snippet"],
                        "dateLastCrawled": result.get("dateLastCrawled", "")
                    })
            
            # 添加新聞結(jié)果
            if "news" in search_results and "value" in search_results["news"]:
                for news in search_results["news"]["value"][:3]:  # 取前3條新聞
                    results.append({
                        "title": "[新聞] " + news["name"],
                        "link": news["url"],
                        "snippet": news["description"],
                        "datePublished": news.get("datePublished", "")
                    })
            
            return {
                "query": query,
                "results": results,
                "total_results": len(results)
            }
            
        except Exception as e:
            logger.error(f"必應(yīng)搜索出錯: {str(e)}")
            return {
                "query": query,
                "results": [],
                "total_results": 0,
                "error": str(e)
            }

# app/tool/baidu_search.py
import requests
from bs4 import BeautifulSoup
from pydantic import Field
from app.tool.base import BaseTool
from app.logger import logger

class BaiduSearch(BaseTool):
    """使用百度搜索引擎進(jìn)行網(wǎng)絡(luò)搜索"""
    
    name: str = "baidu_search"
    description: str = "使用百度搜索獲取信息，適合中文搜索"
    
    def _call(self, query: str, num_results: int = 10) -> dict:
        """執(zhí)行百度搜索并解析結(jié)果"""
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
        }
        
        search_url = f"https://www.baidu.com/s?wd={query}&rn={num_results}"
        
        try:
            response = requests.get(search_url, headers=headers, timeout=10)
            response.raise_for_status()
            response.encoding = 'utf-8'
            
            soup = BeautifulSoup(response.text, 'html.parser')
            search_results = soup.select('.result.c-container')
            
            results = []
            for result in search_results[:num_results]:
                title_elem = result.select_one('.t')
                link_elem = title_elem.select_one('a') if title_elem else None
                abstract_elem = result.select_one('.c-abstract')
                
                if title_elem and link_elem:
                    title = title_elem.get_text(strip=True)
                    link = link_elem.get('href', '')
                    abstract = abstract_elem.get_text(strip=True) if abstract_elem else "無描述"
                    
                    results.append({
                        "title": title,
                        "link": link,
                        "snippet": abstract
                    })
            
            return {
                "query": query,
                "results": results,
                "total_results": len(results)
            }
            
        except Exception as e:
            logger.error(f"百度搜索出錯: {str(e)}")
            return {
                "query": query,
                "results": [],
                "total_results": 0,
                "error": str(e)
            }

# 修改app/agent/manus.py文件
from app.tool.bing_search import BingSearch
from app.tool.baidu_search import BaiduSearch

# 找到available_tools部分并替換
available_tools: ToolCollection = Field(
    default_factory=lambda: ToolCollection(
        PythonExecute(), 
        BaiduSearch(),  # 國內(nèi)用戶首選
        BingSearch(),   # 備選搜索工具
        BrowserUseTool(), 
        FileSaver(), 
        Terminate()
    )
)

# 我在app/agent/base.py中添加了循環(huán)檢測功能
def is_in_loop(self, actions_history, threshold=3, similarity_threshold=0.85):
    """檢測是否陷入執(zhí)行循環(huán)"""
    if len(actions_history) < threshold * 2:
        return False
    
    recent_actions = actions_history[-threshold:]
    previous_actions = actions_history[-(threshold*2):-threshold]
    
    # 計算最近動作與前一批動作的相似度
    similarity_count = 0
    for i in range(threshold):
        # 使用簡單字符串相似度
        current = recent_actions[i]
        previous = previous_actions[i]
        
        # 如果動作類型、參數(shù)等關(guān)鍵信息相似
        if current['tool'] == previous['tool'] and \
           self._params_similarity(current['params'], previous['params']) > similarity_threshold:
            similarity_count += 1
    
    # 如果超過一定比例的動作重復(fù)，判定為循環(huán)
    return similarity_count / threshold > 0.7

???????def _params_similarity(self, params1, params2):
    """計算兩組參數(shù)的相似度"""
    # 簡化實現(xiàn)，實際使用中可以采用更復(fù)雜的相似度算法
    if params1 == params2:
        return 1.0
    
    common_keys = set(params1.keys()) & set(params2.keys())
    if not common_keys:
        return 0.0
    
    similarity = 0
    for key in common_keys:
        if params1[key] == params2[key]:
            similarity += 1
    
    return similarity / len(common_keys)

# 在執(zhí)行流程中使用循環(huán)檢測
def run(self, prompt):
    actions_history = []
    
    for step in range(self.max_steps):
        action = self.plan_next_action(prompt)
        actions_history.append(action)
        
        # 檢測是否陷入循環(huán)
        if len(actions_history) > 6 and self.is_in_loop(actions_history):
            logger.warning("檢測到執(zhí)行循環(huán)，嘗試重新規(guī)劃...")
            # 添加特殊提示，幫助模型跳出循環(huán)
            prompt += "\n\n[系統(tǒng)提示]: 檢測到可能的執(zhí)行循環(huán)，請嘗試不同的解決方案或工具。"
            continue
        
        result = self.execute_action(action)
        
        if self.is_task_complete():
            return result
    
    return "達(dá)到最大步驟數(shù)，任務(wù)未完成"

# 改進(jìn)的FileSaver工具實現(xiàn)
def save_file_with_verification(self, content, file_path, overwrite=False):
    """保存文件并驗證成功與否"""
    # 規(guī)范化路徑
    file_path = os.path.abspath(file_path)
    
    # 檢查目錄是否存在，不存在則創(chuàng)建
    dir_path = os.path.dirname(file_path)
    if not os.path.exists(dir_path):
        try:
            os.makedirs(dir_path, exist_ok=True)
        except Exception as e:
            return f"創(chuàng)建目錄失敗: {dir_path}, 錯誤: {str(e)}"
    
    # 檢查文件是否已存在
    if os.path.exists(file_path) and not overwrite:
        return f"文件已存在且未設(shè)置覆蓋: {file_path}"
    
    # 保存文件
    try:
        if isinstance(content, str):
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)
        else:
            with open(file_path, 'wb') as f:
                f.write(content)
        
        # 驗證文件是否成功保存
        if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
            return f"文件成功保存到: {file_path}"
        else:
            return f"文件保存失敗，雖然沒有報錯但文件為空: {file_path}"
    except Exception as e:
        return f"保存文件出錯: {str(e)}"

# 在main.py中添加的人工干預(yù)選項
async def main_with_intervention():
    agent = Manus()
    
    while True:
        try:
            prompt = input("Enter your prompt (or 'exit' to quit): ")
            if prompt.lower() == "exit":
                logger.info("Goodbye!")
                break
                
            if prompt.strip().isspace():
                logger.warning("Skipping empty prompt.")
                continue
                
            logger.warning("Processing your request...")
            
            # 開啟人工干預(yù)模式
            intervention_mode = input("是否開啟人工干預(yù)模式? (y/n): ").lower() == 'y'
            
            if intervention_mode:
                await run_with_intervention(agent, prompt)
            else:
                await agent.run(prompt)
                
        except KeyboardInterrupt:
            logger.warning("Goodbye!")
            break

???????async def run_with_intervention(agent, prompt):
    """帶人工干預(yù)的執(zhí)行模式"""
    for step in range(agent.max_steps):
        # 獲取下一步計劃
        action = await agent.plan_next_action(prompt)
        
        # 展示計劃并請求人工確認(rèn)
        print(f"\n計劃執(zhí)行: 工具={action['tool']}, 參數(shù)={action['params']}")
        choice = input("選擇操作 (e-執(zhí)行/s-跳過/m-修改/q-退出): ").lower()
        
        if choice == 'q':
            print("手動終止執(zhí)行")
            break
        elif choice == 's':
            print("跳過此步驟")
            continue
        elif choice == 'm':
            # 允許修改參數(shù)
            print("當(dāng)前參數(shù):", action['params'])
            try:
                new_params = input("輸入修改后的參數(shù) (JSON格式): ")
                import json
                action['params'] = json.loads(new_params)
                print("參數(shù)已更新")
            except Exception as e:
                print(f"參數(shù)格式錯誤: {str(e)}, 使用原參數(shù)")
        
        # 執(zhí)行動作
        result = await agent.execute_action(action)
        print(f"執(zhí)行結(jié)果: {result}")
        
        # 檢查任務(wù)是否完成
        if agent.is_task_complete():
            print("任務(wù)已完成!")
            break

# 我在app/agent/base.py中添加的性能調(diào)優(yōu)代碼
def optimize_performance(self):
    """性能調(diào)優(yōu)設(shè)置"""
    # 1. 緩存機(jī)制
    self.enable_result_cache = True  # 啟用結(jié)果緩存
    self.cache_ttl = 3600  # 緩存有效期(秒)
    
    # 2. 工具預(yù)熱
    self.preload_frequent_tools = True
    
    # 3. 批處理請求
    self.batch_size = 3  # 批量處理的請求數(shù)
    
    # 4. 模型參數(shù)優(yōu)化
    self.token_window_size = 6000  # 上下文窗口大小
    self.summarize_threshold = 7500  # 何時開始壓縮歷史

# 我創(chuàng)建的數(shù)據(jù)分析擴(kuò)展
# app/extension/data_analysis.py
import os
import sys
from app.tool.base import BaseTool

???????class EnhancedDataAnalysis(BaseTool):
    """增強(qiáng)的數(shù)據(jù)分析工具"""
    
    name: str = "enhanced_data_analysis"
    description: str = "提供高級數(shù)據(jù)分析功能，包括數(shù)據(jù)可視化、統(tǒng)計分析和預(yù)測模型"
    
    def _call(self, action: str, **kwargs):
        """執(zhí)行數(shù)據(jù)分析相關(guān)操作"""
        if action == "setup":
            return self._setup_environment()
        elif action == "analyze":
            return self._analyze_data(kwargs.get("file_path"), kwargs.get("analysis_type"))
        elif action == "visualize":
            return self._create_visualization(kwargs.get("data"), kwargs.get("chart_type"))
        elif action == "predict":
            return self._build_prediction_model(
                kwargs.get("data"), 
                kwargs.get("target_variable"),
                kwargs.get("model_type", "linear")
            )
        else:
            return f"未知的數(shù)據(jù)分析操作: {action}"
    
    def _setup_environment(self):
        """安裝數(shù)據(jù)分析所需的Python包"""
        try:
            import pip
            packages = [
                "pandas", "numpy", "matplotlib", "seaborn", 
                "scikit-learn", "statsmodels", "plotly"
            ]
            for package in packages:
                try:
                    __import__(package)
                except ImportError:
                    pip.main(["install", package])
            
            return "數(shù)據(jù)分析環(huán)境已成功設(shè)置"
        except Exception as e:
            return f"設(shè)置數(shù)據(jù)分析環(huán)境時出錯: {str(e)}"
    
    # 其他方法實現(xiàn)...

# 添加到app/logger.py中
import logging
import re

class ErrorPatternFilter(logging.Filter):
    """根據(jù)錯誤模式過濾日志"""
    
    def __init__(self, patterns):
        super().__init__()
        self.patterns = patterns
    
    def filter(self, record):
        if record.levelno < logging.ERROR:
            return True
        
        message = record.getMessage()
        for pattern, handler in self.patterns:
            if re.search(pattern, message):
                handler(record)  # 調(diào)用特定的處理函數(shù)
        
        return True

# 錯誤處理函數(shù)
def handle_api_error(record):
    """處理API相關(guān)錯誤"""
    message = record.getMessage()
    if "400" in message:
        print("\n=== API 400錯誤自動診斷 ===")
        print("可能原因:")
        print("1. 請求格式錯誤")
        print("2. 參數(shù)無效")
        print("3. 模型不支持當(dāng)前操作")
        print("建議操作:")
        print("- 檢查config.toml中的模型配置")
        print("- 確認(rèn)API密鑰格式正確")
        print("- 查看API文檔驗證請求格式")
        print("===========================\n")

???????# 為日志添加過濾器
error_patterns = [
    (r"API.*error.*400", handle_api_error),
    # 添加更多錯誤模式和處理函數(shù)
]
logger.addFilter(ErrorPatternFilter(error_patterns))

# app/debug.py
import time
import cProfile
import pstats
import io
from functools import wraps

def debug_mode(enabled=False):
    """調(diào)試模式裝飾器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if not enabled:
                return func(*args, **kwargs)
            
            print(f"\n[DEBUG] 調(diào)用 {func.__name__}")
            print(f"[DEBUG] 參數(shù): {args}, {kwargs}")
            
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            
            print(f"[DEBUG] 返回: {result}")
            print(f"[DEBUG] 耗時: {end_time - start_time:.4f}秒\n")
            
            return result
        return wrapper
    return decorator

???????def profile_performance(func):
    """性能分析裝飾器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        pr = cProfile.Profile()
        pr.enable()
        
        result = func(*args, **kwargs)
        
        pr.disable()
        s = io.StringIO()
        ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
        ps.print_stats(20)  # 打印前20個最耗時的函數(shù)
        print(s.getvalue())
        
        return result
    return wrapper