# 創(chuàng)建虛擬環(huán)境
python -m venv ai-tools-env
source ai-tools-env/bin/activate  # Windows用: ai-tools-env\Scripts\activate
# 安裝依賴
pip install openai pypdf2 requests beautifulsoup4 python-dotenv
pip install aiohttp httpx  # 異步請求支持

# API配置
DEEPSEEK_API_KEY=your_deepseek_api_key
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
# 或使用硅基流動(dòng)（支持多個(gè)模型）
SILICONFLOW_API_KEY=your_siliconflow_key
SILICONFLOW_BASE_URL=https://api.siliconflow.cn/v1
# 搜索API（可選）
BING_SEARCH_API_KEY=your_bing_key

import os
import asyncio
from typing import List, Dict, Optional, AsyncGenerator
from dataclasses import dataclass
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()

@dataclass
class Message:
    """消息數(shù)據(jù)結(jié)構(gòu)"""
    role: str  # system / user / assistant
    content: str

class LLMClient:
    """統(tǒng)一的大模型客戶端"""

    def __init__(
        self,
        api_key: str = None,
        base_url: str = None,
        model: str = "deepseek-chat",
        temperature: float = 0.7
    ):
        self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY")
        self.base_url = base_url or os.getenv("DEEPSEEK_BASE_URL")
        self.model = model
        self.temperature = temperature

        self.client = AsyncOpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )

    async def chat(
        self,
        messages: List[Message],
        stream: bool = False,
        **kwargs
    ) -> str:
        """發(fā)送聊天請求

        Args:
            messages: 消息列表
            stream: 是否流式輸出
            **kwargs: 其他參數(shù)（max_tokens等）

        Returns:
            模型回復(fù)內(nèi)容
        """
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": m.role, "content": m.content} for m in messages],
            temperature=kwargs.get("temperature", self.temperature),
            stream=stream,
            max_tokens=kwargs.get("max_tokens", 4000)
        )

        if stream:
            # 流式輸出處理
            full_content = ""
            async for chunk in response:
                if chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    full_content += content
                    print(content, end="", flush=True)  # 實(shí)時(shí)打印
            return full_content
        else:
            return response.choices[0].message.content

    async def chat_with_functions(
        self,
        messages: List[Message],
        functions: List[Dict]
    ) -> Dict:
        """帶函數(shù)調(diào)用的聊天（用于代碼執(zhí)行等場景）"""
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": m.role, "content": m.content} for m in messages],
            tools=functions,
            tool_choice="auto"
        )
        return response.choices[0].message

# 使用示例
async def test_llm():
    llm = LLMClient()
    response = await llm.chat([
        Message(role="user", content="用Python寫一個(gè)快速排序")
    ])
    print(response)

if __name__ == "__main__":
    asyncio.run(test_llm())

import asyncio
from typing import List, Optional
from pathlib import Path
import PyPDF2
from bs4 import BeautifulSoup
import aiohttp
from dataclasses import dataclass
from datetime import datetime

@dataclass
class DocumentSummary:
    """文檔摘要結(jié)果"""
    title: str
    summary: str
    key_points: List[str]
    reading_time: int  # 預(yù)計(jì)閱讀時(shí)間（分鐘）
    word_count: int
    created_at: str

class DocumentParser:
    """文檔解析器"""

    @staticmethod
    async def parse_pdf(file_path: str) -> str:
        """解析PDF文件"""
        text = ""
        with open(file_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            for page in pdf_reader.pages:
                text += page.extract_text() + "\n"
        return text

    @staticmethod
    async def parse_text(file_path: str) -> str:
        """解析純文本文件"""
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()

    @staticmethod
    async def parse_url(url: str) -> str:
        """解析網(wǎng)頁內(nèi)容"""
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                html = await response.text()

        soup = BeautifulSoup(html, 'html.parser')
        # 移除腳本和樣式
        for script in soup(['script', 'style']):
            script.decompose()

        return soup.get_text(separator='\n', strip=True)

class TextChunker:
    """文本分塊器"""

    def __init__(self, chunk_size: int = 3000, overlap: int = 200):
        """
        Args:
            chunk_size: 每塊的最大字符數(shù)
            overlap: 塊之間的重疊字符數(shù)
        """
        self.chunk_size = chunk_size
        self.overlap = overlap

    def chunk(self, text: str) -> List[str]:
        """將文本分成多個(gè)塊

        策略：按段落分割，確保每塊不超過chunk_size
        """
        # 按段落分割
        paragraphs = text.split('\n\n')
        chunks = []
        current_chunk = ""

        for para in paragraphs:
            if len(current_chunk) + len(para) <= self.chunk_size:
                current_chunk += para + "\n\n"
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                # 如果單個(gè)段落超過chunk_size，強(qiáng)制分割
                if len(para) > self.chunk_size:
                    for i in range(0, len(para), self.chunk_size - self.overlap):
                        chunks.append(para[i:i + self.chunk_size])
                    current_chunk = ""
                else:
                    current_chunk = para + "\n\n"

        if current_chunk:
            chunks.append(current_chunk.strip())

        return chunks

class DocumentSummarizer:
    """智能文檔總結(jié)器"""

    def __init__(self, llm_client: LLMClient):
        self.llm = llm_client
        self.parser = DocumentParser()
        self.chunker = TextChunker()

    async def summarize(
        self,
        source: str,
        source_type: str = "file",
        output_format: str = "markdown"
    ) -> DocumentSummary:
        """
        總結(jié)文檔

        Args:
            source: 文件路徑或URL
            source_type: "file"(文件) 或 "url"(網(wǎng)頁)
            output_format: "markdown", "json", "mindmap"

        Returns:
            DocumentSummary對象
        """
        print(f"?? 正在解析文檔: {source}")

        # 1. 解析文檔
        if source_type == "url":
            text = await self.parser.parse_url(source)
            title = await self._extract_title_from_url(text)
        else:
            # 根據(jù)擴(kuò)展名判斷
            if source.endswith('.pdf'):
                text = await self.parser.parse_pdf(source)
            else:
                text = await self.parser.parse_text(source)
            title = Path(source).stem

        word_count = len(text)
        reading_time = max(1, word_count // 500)  # 假設(shè)每分鐘讀500字

        print(f"? 解析完成，共 {word_count} 字，預(yù)計(jì)閱讀 {reading_time} 分鐘")
        print(f"?? 正在分塊...")

        # 2. 分塊
        chunks = self.chunker.chunk(text)
        print(f"?? 分成 {len(chunks)} 個(gè)塊")

        # 3. 并行總結(jié)每個(gè)塊
        print(f"?? 正在AI總結(jié)...")
        chunk_summaries = await self._summarize_chunks(chunks)

        # 4. 二次總結(jié)
        print(f"?? 正在整合摘要...")
        final_summary = await self._merge_summaries(chunk_summaries, title)

        # 5. 提取關(guān)鍵要點(diǎn)
        key_points = await self._extract_key_points(final_summary)

        return DocumentSummary(
            title=title,
            summary=final_summary,
            key_points=key_points,
            reading_time=reading_time,
            word_count=word_count,
            created_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        )

    async def _summarize_chunks(self, chunks: List[str]) -> List[str]:
        """并行總結(jié)每個(gè)文本塊"""
        semaphore = asyncio.Semaphore(5)  # 限制并發(fā)數(shù)

        async def summarize_chunk(chunk: str, index: int):
            async with semaphore:
                prompt = f"""請總結(jié)以下文本的核心內(nèi)容，要求：
1. 保留關(guān)鍵信息（數(shù)據(jù)、結(jié)論、人名等）
2. 省略細(xì)節(jié)和例子
3. 用簡潔的語言表達(dá)
4. 200字以內(nèi)

文本內(nèi)容：
{chunk}

總結(jié)："""

                response = await self.llm.chat([
                    Message(role="system", content="你是一個(gè)專業(yè)的內(nèi)容總結(jié)助手"),
                    Message(role="user", content=prompt)
                ])

                print(f"  └─ 塊 {index+1}/{len(chunks)} 完成")
                return response

        tasks = [summarize_chunk(chunk, i) for i, chunk in enumerate(chunks)]
        return await asyncio.gather(*tasks)

    async def _merge_summaries(self, summaries: List[str], title: str) -> str:
        """合并所有摘要"""
        combined = "\n\n".join([f"? {s}" for s in summaries])

        prompt = f"""以下是文檔《{title}》的分塊摘要，請整合成一篇完整的總結(jié)：

{combined}

請按以下格式輸出：
# 文檔總結(jié)

## 核心內(nèi)容
[200-300字的完整總結(jié)]

## 主要觀點(diǎn)
1. [觀點(diǎn)1]
2. [觀點(diǎn)2]
...

整合后的總結(jié)："""

        response = await self.llm.chat([
            Message(role="system", content="你是一個(gè)專業(yè)的內(nèi)容整合助手"),
            Message(role="user", content=prompt)
        ])

        return response

    async def _extract_key_points(self, summary: str) -> List[str]:
        """提取關(guān)鍵要點(diǎn)"""
        prompt = f"""從以下總結(jié)中提取5-7個(gè)關(guān)鍵要點(diǎn)，每點(diǎn)不超過20字：

{summary}

只輸出要點(diǎn)列表，每行一個(gè)："""

        response = await self.llm.chat([
            Message(role="user", content=prompt)
        ])

        return [line.strip() for line in response.split('\n') if line.strip()]

    async def _extract_title_from_url(self, text: str) -> str:
        """從網(wǎng)頁文本中提取標(biāo)題"""
        prompt = f"""從以下文本中提取文章標(biāo)題，只返回標(biāo)題：

{text[:500]}

標(biāo)題："""

        response = await self.llm.chat([
            Message(role="user", content=prompt)
        ])

        return response.strip()

# 使用示例
async def main_summarizer():
    llm = LLMClient()
    summarizer = DocumentSummarizer(llm)

    # 總結(jié)PDF文檔
    result = await summarizer.summarize(
        source="research_paper.pdf",
        source_type="file"
    )

    print("\n" + "="*60)
    print(f"?? 標(biāo)題: {result.title}")
    print(f"??  預(yù)計(jì)閱讀時(shí)間: {result.reading_time} 分鐘")
    print(f"?? 字?jǐn)?shù): {result.word_count}")
    print("\n?? 關(guān)鍵要點(diǎn):")
    for point in result.key_points:
        print(f"  ? {point}")
    print(f"\n?? 總結(jié):\n{result.summary}")

if __name__ == "__main__":
    asyncio.run(main_summarizer())

import re
import subprocess
import tempfile
from typing import Dict, List, Optional, Tuple
from enum import Enum
import ast

class CodeMode(Enum):
    """代碼生成模式"""
    GENERATE = "generate"    # 生成新代碼
    EXPLAIN = "explain"      # 解釋代碼
    OPTIMIZE = "optimize"    # 優(yōu)化代碼
    DEBUG = "debug"          # 調(diào)試代碼
    TEST = "test"            # 生成測試

@dataclass
class CodeResult:
    """代碼生成結(jié)果"""
    code: str
    language: str
    explanation: str
    tests: Optional[str] = None
    warnings: List[str] = None

class CodeGenerator:
    """AI代碼生成器"""

    def __init__(self, llm_client: LLMClient):
        self.llm = llm_client

        # 代碼質(zhì)量檢查規(guī)則
        self.quality_rules = {
            "security": [
                r"eval\s*\(",  # 避免eval
                r"exec\s*\(",  # 避免exec
                r"pickle\.loads?",  # 避免pickle
            ],
            "performance": [
                r"for\s+\w+\s+in\s+range\(len\(",  # 用enumerate
            ]
        }

    async def generate(
        self,
        requirement: str,
        language: str = "python",
        mode: CodeMode = CodeMode.GENERATE,
        context: str = ""
    ) -> CodeResult:
        """
        生成/處理代碼

        Args:
            requirement: 用戶需求
            language: 編程語言
            mode: 生成模式
            context: 上下文代碼（用于續(xù)寫）

        Returns:
            CodeResult對象
        """
        mode_prompts = {
            CodeMode.GENERATE: self._build_generate_prompt,
            CodeMode.EXPLAIN: self._build_explain_prompt,
            CodeMode.OPTIMIZE: self._build_optimize_prompt,
            CodeMode.DEBUG: self._build_debug_prompt,
            CodeMode.TEST: self._build_test_prompt,
        }

        # 構(gòu)建提示詞
        prompt_builder = mode_prompts[mode]
        prompt = prompt_builder(requirement, language, context)

        print(f"?? 正在生成{mode.value}...")

        # 調(diào)用LLM
        response = await self.llm.chat([
            Message(role="system", content=self._get_system_prompt(language)),
            Message(role="user", content=prompt)
        ])

        # 解析響應(yīng)
        code, explanation = self._parse_code_response(response, language)

        # 安全檢查
        warnings = self._security_check(code)

        # 生成測試（如果是生成模式）
        tests = None
        if mode == CodeMode.GENERATE:
            tests = await self._generate_tests(code, language)

        return CodeResult(
            code=code,
            language=language,
            explanation=explanation,
            tests=tests,
            warnings=warnings
        )

    def _get_system_prompt(self, language: str) -> str:
        """獲取系統(tǒng)提示詞"""
        return f"""你是一個(gè)專業(yè)的{language}程序員和教師。

輸出代碼時(shí)：
1. 代碼必須可直接運(yùn)行
2. 添加必要的注釋和文檔字符串
3. 遵循{language}最佳實(shí)踐和PEP8規(guī)范
4. 包含錯(cuò)誤處理
5. 代碼后附上簡潔的使用說明

輸出格式：
```python
# 代碼塊

def _build_generate_prompt(
    self, requirement: str, language: str, context: str
) -> str:
    """構(gòu)建代碼生成提示詞"""
    if context:
        return f"""請根據(jù)以下需求生成{language}代碼：

{context}

    return f"""請根據(jù)以下需求生成{language}代碼：

def _build_explain_prompt(
    self, code: str, language: str, context: str
) -> str:
    """構(gòu)建代碼解釋提示詞"""
    return f"""請?jiān)敿?xì)解釋以下{language}代碼的功能和工作原理：

def _build_optimize_prompt(
    self, code: str, language: str, context: str
) -> str:
    """構(gòu)建代碼優(yōu)化提示詞"""
    return f"""請優(yōu)化以下{language}代碼：

def _build_debug_prompt(
    self, code: str, language: str, context: str
) -> str:
    """構(gòu)建調(diào)試提示詞"""
    return f"""請分析以下{language}代碼中的問題并修復(fù)：

def _build_test_prompt(
    self, code: str, language: str, context: str
) -> str:
    """構(gòu)建測試生成提示詞"""
    return f"""請為以下{language}代碼生成完整的測試用例：

def _parse_code_response(self, response: str, language: str) -> Tuple[str, str]:
    """解析LLM響應(yīng)，提取代碼和說明"""
    # 提取代碼塊
    code_pattern = rf"```{language}\n(.*?)```"
    code_match = re.search(code_pattern, response, re.DOTALL)

    if code_match:
        code = code_match.group(1).strip()
        explanation = response.replace(code_match.group(0), "").strip()
    else:
        # 如果沒有代碼塊標(biāo)記，嘗試提取
        code = response
        explanation = "無額外說明"

    return code, explanation

def _security_check(self, code: str) -> List[str]:
    """代碼安全檢查"""
    warnings = []

    for category, patterns in self.quality_rules.items():
        for pattern in patterns:
            if re.search(pattern, code):
                warnings.append(f"?? 安全警告: 檢測到 {pattern} 使用")

    # Python語法檢查
    try:
        ast.parse(code)
    except SyntaxError as e:
        warnings.append(f"?? 語法錯(cuò)誤: {e}")

    return warnings

async def _generate_tests(self, code: str, language: str) -> str:
    """生成測試代碼"""
    prompt = f"""為以下{language}代碼編寫pytest測試：

    response = await self.llm.chat([
        Message(role="user", content=prompt)
    ])

    return response

async def execute_code(
    self, code: str, language: str = "python",
    timeout: int = 10
) -> Dict:
    """安全執(zhí)行代碼

    Returns:
        {
            "success": bool,
            "output": str,
            "error": str
        }
    """
    with tempfile.NamedTemporaryFile(
        mode='w',
        suffix=f'.{language}',
        delete=False
    ) as f:
        f.write(code)
        temp_file = f.name

    try:
        result = subprocess.run(
            ['python', temp_file],
            capture_output=True,
            text=True,
            timeout=timeout
        )

        return {
            "success": result.returncode == 0,
            "output": result.stdout,
            "error": result.stderr
        }
    except subprocess.TimeoutExpired:
        return {
            "success": False,
            "error": f"執(zhí)行超時(shí)（{timeout}秒）"
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }
    finally:
        import os
        os.unlink(temp_file)

def __init__(self, llm_client: LLMClient):
    self.generator = CodeGenerator(llm_client)
    self.history: List[Dict] = []

async def chat(self, user_input: str) -> str:
    """對話式代碼助手"""
    # 檢測意圖
    intent = await self._detect_intent(user_input)

    if intent == "generate":
        result = await self.generator.generate(
            requirement=user_input,
            mode=CodeMode.GENERATE
        )

        output = f"```python\n{result.code}\n```\n\n"
        output += f"**說明：**\n{result.explanation}\n\n"

        if result.warnings:
            output += "**安全警告：**\n" + "\n".join(result.warnings) + "\n\n"

        if result.tests:
            output += f"**測試代碼：**\n```python\n{result.tests}\n```"

        return output

    elif intent == "explain":
        # 提取代碼
        code = self._extract_code_from_input(user_input)
        result = await self.generator.generate(
            requirement=code,
            mode=CodeMode.EXPLAIN
        )
        return result.explanation

async def _detect_intent(self, user_input: str) -> str:
    """檢測用戶意圖"""
    prompt = f"""判斷用戶意圖，只返回：generate / explain / optimize / debug

    response = await self.generator.llm.chat([
        Message(role="user", content=prompt)
    ])

    intent = response.strip().lower()
    return intent if intent in ["generate", "explain", "optimize", "debug"] else "generate"

def _extract_code_from_input(self, user_input: str) -> str:
    """從輸入中提取代碼"""
    # 提取```代碼塊```
    match = re.search(r'```(?:python)?\n(.*?)```', user_input, re.DOTALL)
    if match:
        return match.group(1).strip()

    # 如果沒有代碼塊，返回原文
    return user_input

async def main_code_generator():
llm = LLMClient()
assistant = InteractiveCodeAssistant(llm)
# 示例1：生成代碼
print("="*60)
print("示例1：生成快速排序代碼")
print("="*60)
result = await assistant.chat("用Python實(shí)現(xiàn)一個(gè)快速排序，要求有詳細(xì)注釋")
print(result)
# 示例2：解釋代碼
print("\n" + "="*60)
print("示例2：解釋代碼")
print("="*60)
code = """

def quicksort(arr):
if len(arr) <= 1:
return arr
pivot = arr[len(arr) // 2]
left = [x for x in arr if x < pivot]
middle = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
return quicksort(left) + middle + quicksort(right)
“”"
explanation = await assistant.chat (f"解釋這段代碼在做什么：\n\n[code]\n")
print(explanation)
if name == “main”:
asyncio.run(main_code_generator())

import aiohttp
from typing import List, Dict, Optional
from dataclasses import dataclass
import re
from urllib.parse import quote, urljoin
import json

@dataclass
class SearchResult:
    """搜索結(jié)果"""
    title: str
    url: str
    snippet: str
    source: str  # google / bing / docs / stackoverflow
    relevance: float = 0.0

@dataclass
class ResearchResult:
    """研究結(jié)果"""
    answer: str
    sources: List[SearchResult]
    related_questions: List[str]
    confidence: float

class SearchEngine:
    """搜索引擎封裝"""

    def __init__(self, bing_api_key: str = None):
        self.bing_api_key = bing_api_key or os.getenv("BING_SEARCH_API_KEY")
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        }

    async def search_bing(
        self,
        query: str,
        count: int = 10
    ) -> List[SearchResult]:
        """使用Bing搜索API"""
        if not self.bing_api_key:
            return await self._search_duckduckgo(query, count)

        url = "https://api.bing.microsoft.com/v7.0/search"
        params = {
            "q": query,
            "count": count,
            "responseFilter": "webpages"
        }

        async with aiohttp.ClientSession() as session:
            async with session.get(
                url,
                params=params,
                headers={"Ocp-Apim-Subscription-Key": self.bing_api_key}
            ) as response:
                data = await response.json()

        results = []
        for item in data.get("webPages", {}).get("value", []):
            results.append(SearchResult(
                title=item["name"],
                url=item["url"],
                snippet=item["snippet"],
                source="bing"
            ))

        return results

    async def _search_duckduckgo(
        self,
        query: str,
        count: int = 10
    ) -> List[SearchResult]:
        """使用免費(fèi)的DuckDuckGo搜索"""
        # 使用DuckDuckGo的HTML版本
        url = f"https://html.duckduckgo.com/html/?q={quote(query)}"

        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=self.headers) as response:
                html = await response.text()

        # 解析結(jié)果
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, 'html.parser')

        results = []
        for result in soup.select('.result')[:count]:
            title_elem = result.select_one('.result__a')
            snippet_elem = result.select_one('.result__snippet')
            url_elem = result.select_one('.result__url')

            if title_elem and url_elem:
                results.append(SearchResult(
                    title=title_elem.get_text(),
                    url=url_elem.get('href', ''),
                    snippet=snippet_elem.get_text() if snippet_elem else '',
                    source="duckduckgo"
                ))

        return results

    async def search_stackoverflow(
        self,
        query: str,
        count: int = 5
    ) -> List[SearchResult]:
        """搜索StackOverflow"""
        search_query = f"site:stackoverflow.com {query}"
        results = await self._search_duckduckgo(search_query, count)

        # 標(biāo)記來源
        for r in results:
            r.source = "stackoverflow"

        return results

    async def search_docs(
        self,
        query: str,
        docs_domain: str,
        count: int = 5
    ) -> List[SearchResult]:
        """搜索特定文檔站（如Python文檔）"""
        search_query = f"site:{docs_domain} {query}"
        results = await self._search_duckduckgo(search_query, count)

        for r in results:
            r.source = "docs"

        return results

class IntelligentResearcher:
    """智能研究助手"""

    def __init__(self, llm_client: LLMClient, search_engine: SearchEngine):
        self.llm = llm_client
        self.search = search_engine

    async def research(
        self,
        question: str,
        depth: int = 1,
        sources: List[str] = None
    ) -> ResearchResult:
        """
        研究問題

        Args:
            question: 研究問題
            depth: 研究深度（1-3）
            sources: 指定搜索源 ["google", "docs", "stackoverflow"]

        Returns:
            ResearchResult對象
        """
        print(f"?? 正在研究: {question}")

        # 1. 并行搜索多個(gè)源
        search_tasks = []

        if not sources or "google" in sources:
            search_tasks.append(self.search.search_bing(question))

        if not sources or "stackoverflow" in sources:
            search_tasks.append(self.search.search_stackoverflow(question))

        # 如果是技術(shù)問題，搜索官方文檔
        if self._is_technical_question(question):
            # 檢測可能的技術(shù)棧
            tech = await self._detect_tech_stack(question)
            if tech:
                docs_url = self._get_docs_url(tech)
                search_tasks.append(
                    self.search.search_docs(question, docs_url)
                )

        # 執(zhí)行所有搜索
        search_results_list = await asyncio.gather(*search_tasks)

        # 合并結(jié)果
        all_results = []
        for results in search_results_list:
            all_results.extend(results)

        print(f"?? 找到 {len(all_results)} 條相關(guān)結(jié)果")

        # 2. 提取頁面內(nèi)容（深度研究）
        if depth > 1:
            all_results = await self._fetch_page_contents(all_results[:5])

        # 3. AI分析并整合答案
        answer = await self._synthesize_answer(question, all_results)

        # 4. 生成相關(guān)問題
        related = await self._generate_related_questions(question, answer)

        # 5. 計(jì)算置信度
        confidence = self._calculate_confidence(all_results)

        return ResearchResult(
            answer=answer,
            sources=all_results[:5],  # 返回最相關(guān)的5條
            related_questions=related,
            confidence=confidence
        )

    def _is_technical_question(self, question: str) -> bool:
        """判斷是否是技術(shù)問題"""
        tech_keywords = [
            "python", "javascript", "java", "api", "函數(shù)",
            "如何使用", "怎么用", "documentation", "example"
        ]
        return any(kw in question.lower() for kw in tech_keywords)

    async def _detect_tech_stack(self, question: str) -> Optional[str]:
        """檢測技術(shù)棧"""
        prompt = f"""從以下問題中檢測涉及的技術(shù)棧，只返回技術(shù)名稱：

問題：{question}

技術(shù)棧（如python、react、docker等）："""

        response = await self.llm.chat([
            Message(role="user", content=prompt)
        ])

        tech = response.strip().lower()
        tech_docs = {
            "python": "docs.python.org",
            "javascript": "developer.mozilla.org",
            "react": "react.dev",
            "vue": "vuejs.org",
            "docker": "docs.docker.com",
            "kubernetes": "kubernetes.io",
        }

        return tech_docs.get(tech)

    def _get_docs_url(self, tech: str) -> str:
        """獲取文檔站點(diǎn)URL"""
        tech_docs = {
            "python": "docs.python.org",
            "javascript": "developer.mozilla.org",
            "react": "react.dev",
            "vue": "vuejs.org",
            "docker": "docs.docker.com",
        }
        return tech_docs.get(tech, "docs.python.org")

    async def _fetch_page_contents(
        self,
        results: List[SearchResult]
    ) -> List[SearchResult]:
        """獲取頁面完整內(nèi)容"""
        async def fetch_content(result: SearchResult):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(
                        result.url,
                        headers=self.search.headers,
                        timeout=aiohttp.ClientTimeout(total=10)
                    ) as response:
                        html = await response.text()

                from bs4 import BeautifulSoup
                soup = BeautifulSoup(html, 'html.parser')

                # 提取主要內(nèi)容
                for script in soup(['script', 'style', 'nav', 'footer']):
                    script.decompose()

                text = soup.get_text(separator='\n', strip=True)
                # 取前2000字符
                result.snippet = text[:2000] + "..."
                result.relevance = 1.0  # 已獲取完整內(nèi)容，相關(guān)性高

            except Exception as e:
                print(f"  ?? 獲取失敗 {result.url}: {e}")

        tasks = [fetch_content(r) for r in results]
        await asyncio.gather(*tasks)

        return results

    async def _synthesize_answer(
        self,
        question: str,
        results: List[SearchResult]
    ) -> str:
        """綜合搜索結(jié)果生成答案"""
        # 構(gòu)建上下文
        context = "\n\n".join([
            f"來源{i+1}: {r.title}\n{r.snippet}\n鏈接: {r.url}"
            for i, r in enumerate(results[:5])
        ])

        prompt = f"""基于以下搜索結(jié)果回答問題，要求：

1. 準(zhǔn)確引用信息來源
2. 綜合多個(gè)來源的信息
3. 如果信息沖突，說明不同觀點(diǎn)
4. 給出清晰的結(jié)構(gòu)化答案
5. 標(biāo)注信息來源（如[來源1]）

問題：{question}

搜索結(jié)果：
{context}

請給出詳細(xì)答案："""

        answer = await self.llm.chat([
            Message(role="system", content="你是一個(gè)專業(yè)的研究助手，擅長綜合多源信息給出準(zhǔn)確答案"),
            Message(role="user", content=prompt)
        ])

        return answer

    async def _generate_related_questions(
        self,
        question: str,
        answer: str
    ) -> List[str]:
        """生成相關(guān)問題"""
        prompt = f"""基于以下問答，生成3-5個(gè)相關(guān)的深入研究問題：

問題：{question}
答案：{answer[:500]}...

請生成相關(guān)問題，每行一個(gè)："""

        response = await self.llm.chat([
            Message(role="user", content=prompt)
        ])

        return [
            line.strip()
            for line in response.split('\n')
            if line.strip() and not line.startswith('-')
        ][:5]

    def _calculate_confidence(self, results: List[SearchResult]) -> float:
        """計(jì)算答案置信度"""
        if not results:
            return 0.0

        # 基于結(jié)果數(shù)量和相關(guān)性計(jì)算
        base_confidence = min(1.0, len(results) / 10)

        # 如果有官方文檔，提高置信度
        has_docs = any(r.source == "docs" for r in results)
        if has_docs:
            base_confidence = min(1.0, base_confidence + 0.2)

        return round(base_confidence, 2)

# 使用示例
async def main_researcher():
    llm = LLMClient()
    search = SearchEngine()
    researcher = IntelligentResearcher(llm, search)

    # 研究問題
    result = await researcher.research(
        question="Python中asyncio和multiprocessing的區(qū)別是什么？",
        depth=2
    )

    print("\n" + "="*60)
    print("?? 研究結(jié)果")
    print("="*60)
    print(f"\n置信度: {result.confidence*100}%\n")
    print(f"答案:\n{result.answer}\n")

    print("?? 參考來源:")
    for i, source in enumerate(result.sources, 1):
        print(f"{i}. {source.title}")
        print(f"   {source.url}")
        print(f"   來源: {source.source}\n")

    print("? 相關(guān)問題:")
    for q in result.related_questions:
        print(f"  ? {q}")

if __name__ == "__main__":
    asyncio.run(main_researcher())

import argparse
import asyncio
from pathlib import Path
import json

class AIToolsCLI:
    """AI工具命令行界面"""

    def __init__(self):
        self.llm = LLMClient()
        self.summarizer = DocumentSummarizer(self.llm)
        self.code_assistant = InteractiveCodeAssistant(self.llm)
        self.researcher = IntelligentResearcher(
            self.llm,
            SearchEngine()
        )

    async def run(self):
        """運(yùn)行CLI"""
        parser = argparse.ArgumentParser(
            description="AI工具集 - 你的智能助手",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="""
示例:
  # 總結(jié)文檔
  python ai_tools.py summarize paper.pdf

  # 生成代碼
  python ai_tools.py code "用Python寫一個(gè)爬蟲"

  # 研究問題
  python ai_tools.py research "量子計(jì)算的原理"
            """
        )

        subparsers = parser.add_subparsers(dest='command', help='可用命令')

        # summarize命令
        sum_parser = subparsers.add_parser('summarize', help='總結(jié)文檔')
        sum_parser.add_argument('file', help='文件路徑或URL')
        sum_parser.add_argument('-t', '--type', default='file',
                               choices=['file', 'url'],
                               help='輸入類型')
        sum_parser.add_argument('-o', '--output', help='輸出文件路徑')

        # code命令
        code_parser = subparsers.add_parser('code', help='生成/處理代碼')
        code_parser.add_argument('prompt', help='需求或代碼')
        code_parser.add_argument('-m', '--mode',
                                choices=['generate', 'explain', 'optimize', 'debug'],
                                default='generate',
                                help='處理模式')
        code_parser.add_argument('-l', '--language', default='python',
                                help='編程語言')
        code_parser.add_argument('-x', '--execute', action='store_true',
                                help='執(zhí)行生成的代碼')

        # research命令
        res_parser = subparsers.add_parser('research', help='研究問題')
        res_parser.add_argument('question', help='研究問題')
        res_parser.add_argument('-d', '--depth', type=int, default=1,
                               choices=[1, 2, 3],
                               help='研究深度')
        res_parser.add_argument('-s', '--sources', nargs='+',
                               choices=['google', 'docs', 'stackoverflow'],
                               help='指定搜索源')

        args = parser.parse_args()

        if not args.command:
            parser.print_help()
            return

        # 執(zhí)行對應(yīng)命令
        if args.command == 'summarize':
            await self._cmd_summarize(args)
        elif args.command == 'code':
            await self._cmd_code(args)
        elif args.command == 'research':
            await self._cmd_research(args)

    async def _cmd_summarize(self, args):
        """處理summarize命令"""
        print(f"?? 正在總結(jié): {args.file}")

        result = await self.summarizer.summarize(
            source=args.file,
            source_type=args.type
        )

        # 輸出
        output = f"""# {result.title}

**?? 統(tǒng)計(jì)信息**
- 字?jǐn)?shù): {result.word_count}
- 預(yù)計(jì)閱讀時(shí)間: {result.reading_time} 分鐘
- 生成時(shí)間: {result.created_at}

**?? 關(guān)鍵要點(diǎn)**
{chr(10).join(f'{i+1}. {p}' for i, p in enumerate(result.key_points))}

**?? 總結(jié)**
{result.summary}
"""

        if args.output:
            with open(args.output, 'w', encoding='utf-8') as f:
                f.write(output)
            print(f"? 已保存到: {args.output}")
        else:
            print(output)

    async def _cmd_code(self, args):
        """處理code命令"""
        print(f"?? 正在處理: {args.prompt[:50]}...")

        result = await self.code_assistant.generator.generate(
            requirement=args.prompt,
            language=args.language,
            mode=CodeMode(args.mode)
        )

        # 輸出代碼
        print(f"\n```{args.language}")
        print(result.code)
        print("```\n")

        # 輸出說明
        print(f"**說明**\n{result.explanation}\n")

        # 輸出警告
        if result.warnings:
            print("**警告**")
            for w in result.warnings:
                print(f"  {w}")
            print()

        # 輸出測試
        if result.tests:
            print(f"**測試代碼**\n```{args.language}")
            print(result.tests)
            print("```\n")

        # 執(zhí)行代碼
        if args.execute:
            print("? 正在執(zhí)行代碼...")
            exec_result = await self.code_assistant.generator.execute_code(
                result.code,
                args.language
            )

            if exec_result['success']:
                print(f"? 執(zhí)行成功\n輸出:\n{exec_result['output']}")
            else:
                print(f"? 執(zhí)行失敗\n錯(cuò)誤:\n{exec_result['error']}")

    async def _cmd_research(self, args):
        """處理research命令"""
        print(f"?? 正在研究: {args.question}")

        result = await self.researcher.research(
            question=args.question,
            depth=args.depth,
            sources=args.sources
        )

        # 輸出結(jié)果
        print(f"""
# 研究結(jié)果

**?? 置信度**: {result.confidence*100}%

## 答案
{result.answer}

## 參考來源
""")

        for i, source in enumerate(result.sources, 1):
            print(f"{i}. **{source.title}**")
            print(f"   鏈接: {source.url}")
            print(f"   來源: {source.source}\n")

        if result.related_questions:
            print("## 相關(guān)問題")
            for q in result.related_questions:
                print(f"- {q}")

async def main():
    cli = AIToolsCLI()
    await cli.run()

if __name__ == "__main__":
    asyncio.run(main())

# 總結(jié)論文
python ai_tools.py summarize research_paper.pdf -o summary.md

# 生成代碼并執(zhí)行
python ai_tools.py code "用Python寫一個(gè)二分查找" -x

# 解釋代碼
python ai_tools.py code "explain this code: `def foo(): return 1`" -m explain

# 深度研究
python ai_tools.py research "RAG和Fine-tuning的區(qū)別" -d 2

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ ./src/
COPY cli.py .
COPY config.py .
ENV PYTHONPATH=/app
CMD ["python", "cli.py", "--help"]

# docker-compose.yml
version: '3.8'
services:
  ai-tools:
    build: .
    env_file:
      - .env
    volumes:
      - ./data:/app/data
    ports:
      - "8000:8000"