Python 实战：构建文档总结、代码生成与智能检索助手

1.3 核心工具类封装

在开始之前，我们先封装一个统一的 LLM 调用类。这能避免后续重复写初始化逻辑。

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:
    """消息数据结构"""
    role: str  # system / user / assistant
    content: str

class LLMClient:
    """统一的大模型客户端"""
    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:
        """发送聊天请求
        Args:
            messages: 消息列表
            stream: 是否流式输出
            **kwargs: 其他参数（max_tokens 等）
        Returns:
            模型回复内容
        """
        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)
            return full_content
        else:
            return response.choices[0].message.content

    async def chat_with_functions(self, messages: List[Message], functions: List[Dict]) -> Dict:
        """带函数调用的聊天（用于代码执行等场景）"""
        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 写一个快速排序")])
    print(response)

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

二、工具一：智能文档总结器

2.1 功能设计

这个工具的核心在于处理长文本。直接丢给大模型会超出上下文限制，所以我们需要分块处理。

• 输入：PDF、Word、网页、Markdown
• 解析：根据类型调用不同解析库
• 分块：按 Token 数或段落切分
• 并行总结：每个块独立处理，提升速度
• 二次总结：合并块摘要，形成最终报告

2.2 核心代码实现

这里的关键是 TextChunker 和 DocumentSummarizer 的配合。注意 chunk_size 和 overlap 的设置，重叠部分能防止语义断裂。

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:
    """文档摘要结果"""
    title: str
    summary: str
    key_points: List[str]
    reading_time: int  # 预计阅读时间（分钟）
    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:
        """解析网页内容"""
        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):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def chunk(self, text: str) -> List[str]:
        """将文本分成多个块"""
        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())
                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 = ""
        if current_chunk:
            chunks.append(current_chunk.strip())
        return chunks

class DocumentSummarizer:
    """智能文档总结器"""
    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:
        """总结文档"""
        print(f"📖 正在解析文档：{source}")
        
        # 1. 解析文档
        if source_type == "url":
            text = await self.parser.parse_url(source)
            title = await self._extract_title_from_url(text)
        else:
            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)
        print(f"✅ 解析完成，共 {word_count} 字，预计阅读 {reading_time} 分钟")
        print(f"🔪 正在分块...")
        
        # 2. 分块
        chunks = self.chunker.chunk(text)
        print(f"📦 分成 {len(chunks)} 个块")
        
        # 3. 并行总结每个块
        print(f"🤖 正在 AI 总结...")
        chunk_summaries = await self._summarize_chunks(chunks)
        
        # 4. 二次总结
        print(f"🔄 正在整合摘要...")
        final_summary = await self._merge_summaries(chunk_summaries, title)
        
        # 5. 提取关键要点
        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]:
        """并行总结每个文本块"""
        semaphore = asyncio.Semaphore(5)  # 限制并发数
        
        async def summarize_chunk(chunk: str, index: int):
            async with semaphore:
                prompt = f"""请总结以下文本的核心内容，要求：
1. 保留关键信息（数据、结论、人名等）
2. 省略细节和例子
3. 用简洁的语言表达
4. 200 字以内
文本内容：{chunk}
总结："""
                response = await self.llm.chat([
                    Message(role="system", content="你是一个专业的内容总结助手"),
                    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}》的分块摘要，请整合成一篇完整的总结：
{combined}
请按以下格式输出：
# 文档总结
## 核心内容
[200-300 字的完整总结]
## 主要观点
1. [观点 1]
2. [观点 2]
...
整合后的总结："""
        response = await self.llm.chat([
            Message(role="system", content="你是一个专业的内容整合助手"),
            Message(role="user", content=prompt)
        ])
        return response

    async def _extract_key_points(self, summary: str) -> List[str]:
        """提取关键要点"""
        prompt = f"""从以下总结中提取 5-7 个关键要点，每点不超过 20 字：
{summary}
只输出要点列表，每行一个："""
        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:
        """从网页文本中提取标题"""
        prompt = f"""从以下文本中提取文章标题，只返回标题：
{text[:500]}
标题："""
        response = await self.llm.chat([Message(role="user", content=prompt)])
        return response.strip()

# 使用示例
async def main_summarizer():
    llm = LLMClient()
    summarizer = DocumentSummarizer(llm)
    result = await summarizer.summarize(source="research_paper.pdf", source_type="file")
    print("\n" + "="*60)
    print(f"📄 标题：{result.title}")
    print(f"⏱️ 预计阅读时间：{result.reading_time} 分钟")
    print(f"📊 字数：{result.word_count}")
    print("\n🔑 关键要点:")
    for point in result.key_points:
        print(f" • {point}")
    print(f"\n📝 总结:\n{result.summary}")

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

2.3 使用效果对比

三、工具二：AI 代码生成器

3.1 功能架构

这个工具不仅仅是生成代码，还能解释、优化和调试。关键在于意图识别和安全检查。

• 生成新代码：根据自然语言需求编写
• 解释代码：分析现有代码逻辑
• 优化代码：提升性能或可读性
• 修复 Bug：结合错误信息进行调试

3.2 核心实现

注意这里的 _security_check 方法，它会在代码执行前扫描敏感函数，这是生产环境必须做的防护。

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"
    DEBUG = "debug"
    TEST = "test"

@dataclass
class CodeResult:
    """代码生成结果"""
    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
        self.quality_rules = {
            "security": [
                r"eval\s*\(",
                r"exec\s*\(",
                r"pickle\.loads?",
            ],
            "performance": [
                r"for\s+\w+\s+in\s+range\(len\(",
            ]
        }

    async def generate(self, requirement: str, language: str = "python", mode: CodeMode = CodeMode.GENERATE, context: str = "") -> 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,
        }
        prompt_builder = mode_prompts[mode]
        prompt = prompt_builder(requirement, language, context)
        print(f"🤖 正在生成{mode.value}...")
        
        response = await self.llm.chat([
            Message(role="system", content=self._get_system_prompt(language)),
            Message(role="user", content=prompt)
        ])
        
        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:
        return f"""你是一个专业的{language}程序员和教师。
输出代码时：
1. 代码必须可直接运行
2. 添加必要的注释和文档字符串
3. 遵循{language}最佳实践和 PEP8 规范
4. 包含错误处理
5. 代码后附上简洁的使用说明"""

    def _build_generate_prompt(self, requirement: str, language: str, context: str) -> str:
        if context:
            return f"""请根据以下需求生成{language}代码：
需求：{requirement}
上下文代码：
```{context}```
请生成完整的、可直接运行的代码。"""
        return f"""请根据以下需求生成{language}代码：
需求：{requirement}
要求：
1. 代码完整且可运行
2. 包含必要的输入验证和错误处理
3. 添加清晰的注释
4. 如果是算法，注明时间复杂度
请生成代码："""

    def _build_explain_prompt(self, code: str, language: str, context: str) -> str:
        return f"""请详细解释以下{language}代码的功能和工作原理：
```{code}```
请从以下几个方面解释：
1. 整体功能概述
2. 关键代码逻辑
3. 使用的数据结构和算法
4. 时间/空间复杂度
5. 可能的改进点"""

    def _build_optimize_prompt(self, code: str, language: str, context: str) -> str:
        return f"""请优化以下{language}代码：
```{code}```
优化目标：
1. 提升性能
2. 改善可读性
3. 增强健壮性
4. 遵循最佳实践
请给出：
1. 优化后的代码
2. 优化点说明"""

    def _build_debug_prompt(self, code: str, language: str, context: str) -> str:
        return f"""请分析以下{language}代码中的问题并修复：
```{code}```
可能的错误信息：
{context if context else "[无]"}
请给出：
1. 问题分析
2. 修复后的代码
3. 预防建议"""

    def _build_test_prompt(self, code: str, language: str, context: str) -> str:
        return f"""请为以下{language}代码生成完整的测试用例：
```{code}```
测试要求：
- 测试函数名以 test_开头
- 包含正常和异常情况
- 使用合适的测试框架（如 pytest）
只输出测试代码："""

    def _parse_code_response(self, response: str, language: str) -> Tuple[str, str]:
        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:
            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} 使用")
        try:
            ast.parse(code)
        except SyntaxError as e:
            warnings.append(f"⚠️ 语法错误：{e}")
        return warnings

    async def _generate_tests(self, code: str, language: str) -> str:
        prompt = f"""为以下{language}代码编写 pytest 测试：
```{code}```
要求：
- 测试函数名以 test_开头
- 包含正常和异常情况
- 使用 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:
        """安全执行代码"""
        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"执行超时（{timeout}秒）"}
        except Exception as e:
            return {"success": False, "error": str(e)}
        finally:
            import os
            os.unlink(temp_file)

四、工具三：智能资料助手

4.1 系统架构

这个工具结合了搜索引擎和大模型的推理能力。它能区分事实查询、API 文档和 StackOverflow 等不同来源。

• 多源并行搜索：同时查询 Bing、官方文档、SO
• 内容清洗：去除广告和无关 HTML
• 相关性排序：基于关键词匹配
• AI 整合：综合多源信息生成答案

4.2 核心代码

这里使用了 aiohttp 进行异步请求，确保搜索过程不会阻塞主线程。对于技术文档，我们自动检测技术栈并定向搜索。

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

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

@dataclass
class ResearchResult:
    """研究结果"""
    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]:
        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]:
        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()
                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]:
        search_query = f"site:stackoverflow.com {query}"
        results = await self._search_duckduckgo(search_query, count)
        for r in results:
            r.source = "stackoverflow"
        return results

    async def search_docs(self, query: str, docs_domain: str, count: int = 5) -> List[SearchResult]:
        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:
        print(f"🔍 正在研究：{question}")
        
        # 1. 并行搜索多个源
        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))
        
        if self._is_technical_question(question):
            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))
        
        search_results_list = await asyncio.gather(*search_tasks)
        all_results = []
        for results in search_results_list:
            all_results.extend(results)
        
        print(f"📊 找到 {len(all_results)} 条相关结果")
        
        # 2. 提取页面内容（深度研究）
        if depth > 1:
            all_results = await self._fetch_page_contents(all_results[:5])
        
        # 3. AI 分析并整合答案
        answer = await self._synthesize_answer(question, all_results)
        
        # 4. 生成相关问题
        related = await self._generate_related_questions(question, answer)
        
        # 5. 计算置信度
        confidence = self._calculate_confidence(all_results)
        
        return ResearchResult(answer=answer, sources=all_results[:5], related_questions=related, confidence=confidence)

    def _is_technical_question(self, question: str) -> bool:
        tech_keywords = ["python", "javascript", "java", "api", "函数", "如何使用", "怎么用", "documentation", "example"]
        return any(kw in question.lower() for kw in tech_keywords)

    async def _detect_tech_stack(self, question: str) -> Optional[str]:
        prompt = f"""从以下问题中检测涉及的技术栈，只返回技术名称：
问题：{question}
技术栈（如 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:
        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]:
        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')
                        for script in soup(['script', 'style', 'nav', 'footer']):
                            script.decompose()
                        text = soup.get_text(separator='\n', strip=True)
                        result.snippet = text[:2000] + "..."
                        result.relevance = 1.0
            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:
        context = "\n\n".join([f"来源{i+1}: {r.title}\n{r.snippet}\n链接：{r.url}" for i, r in enumerate(results[:5])])
        prompt = f"""基于以下搜索结果回答问题，要求：
1. 准确引用信息来源
2. 综合多个来源的信息
3. 如果信息冲突，说明不同观点
4. 给出清晰的结构化答案
5. 标注信息来源（如 [来源 1]）
问题：{question}
搜索结果：
{context}
请给出详细答案："""
        answer = await self.llm.chat([
            Message(role="system", content="你是一个专业的研究助手，擅长综合多源信息给出准确答案"),
            Message(role="user", content=prompt)
        ])
        return answer

    async def _generate_related_questions(self, question: str, answer: str) -> List[str]:
        prompt = f"""基于以下问答，生成 3-5 个相关的深入研究问题：
问题：{question}
答案：{answer[:500]}...
请生成相关问题，每行一个："""
        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:
        if not results:
            return 0.0
        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 的区别是什么？", depth=2)
    print("\n" + "="*60)
    print("📚 研究结果")
    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("❓ 相关问题:")
    for q in result.related_questions:
        print(f" • {q}")

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

4.3 搜索效率对比

五、整合三大利器：打造超级 AI 助手

5.1 统一 CLI 工具

为了方便日常使用，我们将三个工具集成到一个命令行界面（CLI）中。这样你可以直接在终端调用它们。

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):
        parser = argparse.ArgumentParser(description="AI 工具集 - 你的智能助手", formatter_class=argparse.RawDescriptionHelpFormatter,
                                         epilog=""" 示例:
# 总结文档
python ai_tools.py summarize paper.pdf
# 生成代码
python ai_tools.py code "用 Python 写一个爬虫"
# 研究问题
python ai_tools.py research "量子计算的原理" """)
        subparsers = parser.add_subparsers(dest='command', help='可用命令')
        
        # summarize 命令
        sum_parser = subparsers.add_parser('summarize', help='总结文档')
        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='执行生成的代码')
        
        # 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
        
        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):
        print(f"📖 正在总结：{args.file}")
        result = await self.summarizer.summarize(source=args.file, source_type=args.type)
        output = f"""# {result.title}
**📊 统计信息**
- 字数：{result.word_count}
- 预计阅读时间：{result.reading_time} 分钟
- 生成时间：{result.created_at}
**🔑 关键要点**
{chr(10).join(f'{i+1}. {p}' for i, p in enumerate(result.key_points))}
**📝 总结**
{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):
        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")
        if args.execute:
            print("⚡ 正在执行代码...")
            exec_result = await self.code_assistant.generator.execute_code(result.code, args.language)
            if exec_result['success']:
                print(f"✅ 执行成功\n输出:\n{exec_result['output']}")
            else:
                print(f"❌ 执行失败\n错误:\n{exec_result['error']}")

    async def _cmd_research(self, args):
        print(f"🔍 正在研究：{args.question}")
        result = await self.researcher.research(question=args.question, depth=args.depth, sources=args.sources)
        print(f""" # 研究结果
**📊 置信度**: {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("## 相关问题")
            for q in result.related_questions:
                print(f"- {q}")

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

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

5.2 使用示例

# 总结论文
python ai_tools.py summarize research_paper.pdf -o summary.md
# 生成代码并执行
python ai_tools.py code "用 Python 写一个二分查找" -x
# 解释代码
python ai_tools.py code "explain this code: `def foo(): return 1`" -m explain
# 深度研究
python ai_tools.py research "RAG 和 Fine-tuning 的区别" -d 2

5.3 成本分析

六、完整源码与部署指南

6.1 项目结构

ai-tools/
├── src/
│   ├── __init__.py
│   ├── llm.py # LLM 客户端
│   ├── summarizer.py # 文档总结器
│   ├── code_generator.py # 代码生成器
│   └── researcher.py # 研究助手
├── cli.py # 命令行入口
├── config.py # 配置管理
├── requirements.txt # 依赖列表
├── .env.example # 环境变量示例
├── README.md # 使用文档
└── examples/
    ├── example_summarize.py
    ├── example_code.py
    └── example_research.py

6.2 部署到云端

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"

6.3 进阶功能扩展

总结

通过这篇文章，我们用 Python 打造了三个强大的 AI 工具：

关键收获

1. LLM 调用很简单 - 用好 OpenAI SDK，30 行代码就能连接大模型
2. 提示词是关键 - 好的 prompt 能让效果翻倍
3. 异步处理很重要 - 并行请求能大幅提升速度
4. 安全意识不能少 - 代码执行要隔离，API 调用要限流

下一步学习

• 📖 深入学习 LangChain 框架
• 🔬 研究 Agent 与 RAG 技术
• 🚀 打造专属 AI 应用

参考资源

• OpenAI Python SDK 文档
• DeepSeek API 文档
• 硅基流动平台
• Python 异步编程指南