第八章：上下文工程——给智能体提供恰当的信息

class ContextBuilder:
    """
    通过搜索代码库，为任务构建专用上下文。

    组件协作：
    - ServiceMatcher: 从任务描述推断相关服务
    - KeywordExtractor: 从任务描述提取搜索关键词
    - CodeSearcher: 在服务目录中搜索匹配文件
    - FileCategorizer: 将匹配文件分为"需修改"和"需参考"
    - PatternDiscoverer: 从参考文件中提取代码模式
    - GraphitiSearch: 从知识图谱获取历史提示
    """

    def build_context(
        self,
        task: str,
        services: list[str] | None = None,
        keywords: list[str] | None = None,
        include_graph_hints: bool = True,
    ) -> TaskContext:
        # 1. 服务推断
        if not services:
            services = self.service_matcher.suggest_services(task)

        # 2. 关键词提取
        if not keywords:
            keywords = self.keyword_extractor.extract_keywords(task)

        # 3. 代码搜索（遍历每个相关服务）
        all_matches = []
        for service_name in services:
            matches = self.searcher.search_service(
                service_path, service_name, keywords
            )
            all_matches.extend(matches)

        # 4. 文件分类
        files_to_modify, files_to_reference = self.categorizer.categorize_matches(
            all_matches, task
        )

        # 5. 模式发现
        patterns = self.pattern_discoverer.discover_patterns(
            files_to_reference, keywords
        )

        # 6. 历史记忆注入
        graph_hints = []
        if include_graph_hints and is_graphiti_enabled():
            graph_hints = asyncio.run(
                fetch_graph_hints(task, self.project_dir)
            )

        return TaskContext(
            task_description=task,
            scoped_services=services,
            files_to_modify=files_to_modify,
            files_to_reference=files_to_reference,
            patterns_discovered=patterns,
            service_contexts=service_contexts,
            graph_hints=graph_hints,
        )

# 异步/同步边界处理（实际代码中）
try:
    loop = asyncio.get_running_loop()
    # 已在 async 上下文中，不能调用 asyncio.run()
    graph_hints = []  # 降级：跳过 Graphiti
except RuntimeError:
    # 没有运行中的事件循环，可以安全调用
    graph_hints = asyncio.run(fetch_graph_hints(...))

class ServiceMatcher:
    """
    从任务描述推断相关的服务/模块。
    """

    def __init__(self, project_index: dict):
        # project_index 包含项目所有服务的目录和关键词
        self.project_index = project_index

    def suggest_services(self, task: str) -> list[str]:
        """
        根据任务描述推断相关服务。

        示例：
        task = "实现用户登录"
        → 匹配关键词 "user", "login", "auth"
        → 推断服务: ["auth", "user"]
        """
        task_lower = task.lower()
        scored_services = []

        for service_name, service_info in self.project_index.get("services", {}).items():
            score = 0

            # 服务名称匹配
            if service_name.lower() in task_lower:
                score += 10

            # 服务关键词匹配
            for keyword in service_info.get("keywords", []):
                if keyword.lower() in task_lower:
                    score += 1

            if score > 0:
                scored_services.append((score, service_name))

        # 按匹配分数排序
        scored_services.sort(reverse=True)
        return [name for _, name in scored_services[:5]]  # 最多 5 个服务

class KeywordExtractor:
    """从任务描述中提取有意义的搜索关键词。"""

    # 停用词：对搜索无意义的词
    STOP_WORDS = {
        "the", "a", "an", "is", "are", "was", "were",
        "in", "on", "at", "to", "for", "of", "and",
        "implement", "add", "create", "build", "make",
        # 开发领域通用动词，搜索价值低
        "function", "method", "class", "module",
    }

    def extract_keywords(self, task: str) -> list[str]:
        """
        提取关键词策略：
        1. 分词（按空格和标点）
        2. 过滤停用词
        3. 过滤短词（< 3 字符）
        4. 提取驼峰/下划线命名中的词（camelCase → camel + Case）
        5. 返回去重后的关键词列表
        """
        words = re.split(r'[\s,;.!?()[\]{}]+', task.lower())
        keywords = set()

        for word in words:
            if len(word) < 3 or word in self.STOP_WORDS:
                continue

            # 处理驼峰命名：loginUser → login, user
            sub_words = re.sub(r'([A-Z])', r' \1', word).split()
            keywords.update(w.lower() for w in sub_words if len(w) >= 3)

            # 处理下划线：user_email → user, email
            if '_' in word:
                keywords.update(p for p in word.split('_') if len(p) >= 3)

            keywords.add(word)

        return list(keywords)

class CodeSearcher:
    """在代码文件中搜索相关内容。"""

    def search_service(
        self,
        service_path: Path,
        service_name: str,
        keywords: list[str],
    ) -> list[FileMatch]:
        matches = []

        for file_path in self._iter_code_files(service_path):
            try:
                content = file_path.read_text(encoding="utf-8", errors="ignore")
                content_lower = content.lower()

                score = 0
                matching_keywords = []
                matching_lines = []

                for keyword in keywords:
                    if keyword in content_lower:
                        count = content_lower.count(keyword)
                        score += min(count, 10)  # 单个关键词最多贡献 10 分
                        matching_keywords.append(keyword)

                        # 记录匹配行（每个关键词最多 3 行）
                        lines = content.split("\n")
                        found = 0
                        for i, line in enumerate(lines, 1):
                            if keyword in line.lower() and found < 3:
                                matching_lines.append((i, line.strip()[:100]))
                                found += 1

                if score > 0:
                    rel_path = str(file_path.relative_to(self.project_dir))
                    matches.append(FileMatch(
                        path=rel_path,
                        service=service_name,
                        reason=f"Contains: {', '.join(matching_keywords)}",
                        relevance_score=score,
                        matching_lines=matching_lines[:5],  # 最多 5 个匹配行
                    ))

            except (OSError, UnicodeDecodeError):
                continue  # 跳过无法读取的文件

        # 按相关度排序
        return sorted(matches, key=lambda m: m.relevance_score, reverse=True)

class FileCategorizer:
    """
    将搜索结果分类为：
    - files_to_modify：智能体可能需要修改的文件
    - files_to_reference：智能体需要了解但不修改的文件
    """

    def categorize_matches(
        self,
        matches: list[FileMatch],
        task: str,
    ) -> tuple[list[dict], list[dict]]:
        """
        分类规则：

        files_to_modify（高概率修改）：
        - 文件名/路径包含任务关键词
        - 文件本身是功能实现文件（handlers.py, services.py, views.py）
        - 相关度分数最高的几个文件

        files_to_reference（参考资料）：
        - 模型定义（models.py, schemas.py）
        - 配置文件（config.py, settings.py）
        - 工具函数（utils.py, helpers.py）
        - 测试文件（test_*.py）
        """
        to_modify = []
        to_reference = []

        # 识别实现文件的模式
        IMPL_PATTERNS = ["handler", "service", "controller", "view", "route", "api"]
        REF_PATTERNS = ["model", "schema", "config", "setting", "util", "helper", "test"]

        for match in matches:
            file_lower = match.path.lower()

            if any(p in file_lower for p in IMPL_PATTERNS):
                to_modify.append(self._format_match(match))
            elif any(p in file_lower for p in REF_PATTERNS):
                to_reference.append(self._format_match(match))
            elif match.relevance_score > MODIFY_THRESHOLD:
                # 高相关度但无法确定角色，归为待修改
                to_modify.append(self._format_match(match))
            else:
                to_reference.append(self._format_match(match))

        # 限制数量（防止上下文溢出）
        return to_modify[:10], to_reference[:15]

class PatternDiscoverer:
    """从代码文件中发现并提取代码模式。"""

    # 需要发现的模式类型
    PATTERN_EXTRACTORS = {
        "decorator_patterns": r"@(\w+)\(",
        "error_patterns": r"raise\s+(\w+Exception|HTTPException)",
        "import_patterns": r"^from\s+(\S+)\s+import",
        "db_patterns": r"(db\.\w+|session\.\w+)",
        "auth_patterns": r"(@require_auth|@login_required|@jwt_required)",
    }

    def discover_patterns(
        self,
        reference_files: list[dict],
        keywords: list[str],
    ) -> dict[str, str]:
        """
        扫描参考文件，提取频繁出现的代码模式。

        返回格式：
        {
            "decorator_usage": "@require_auth is used in 8/10 handler files",
            "error_handling": "Use raise HTTPException(status_code=...) for API errors",
            "database": "Database operations use db.session context manager",
        }
        """
        pattern_counts = defaultdict(Counter)

        for file_info in reference_files:
            file_path = self.project_dir / file_info["path"]
            try:
                content = file_path.read_text(encoding="utf-8", errors="ignore")
            except OSError:
                continue

            for pattern_type, regex in self.PATTERN_EXTRACTORS.items():
                matches = re.findall(regex, content, re.MULTILINE)
                for match in matches:
                    pattern_counts[pattern_type][match] += 1

        # 将统计结果转化为自然语言描述
        return self._format_patterns(pattern_counts, len(reference_files))

@dataclass
class TaskContext:
    """任务的完整上下文。"""

    task_description: str          # 任务描述（原文）
    scoped_services: list[str]     # 推断的相关服务列表
    files_to_modify: list[dict]    # 可能需要修改的文件
    files_to_reference: list[dict] # 需要参考的文件
    patterns_discovered: dict[str, str]  # 发现的代码模式
    service_contexts: dict[str, dict]    # 每个服务的概要信息
    graph_hints: list[dict]        # 来自知识图谱的历史提示

## Task Context

**Task:** 实现用户登录 API，支持邮箱密码验证

**Relevant Services:** auth, user

**Files You Will Likely Modify:**
- src/auth/handlers.py — Contains: login, user, auth (score: 23)
- src/auth/routes.py — Contains: auth, password (score: 15)

**Reference Files (Do Not Modify):**
- src/models/user.py — Contains: user, email (score: 18)
- src/auth/utils.py — Contains: password, hash (score: 12)

**Discovered Patterns:**
- decorator_usage: @require_auth is used in 8/10 handler files
- error_handling: Use raise HTTPException(status_code=...) for API errors

**Historical Context (from memory):**
- [gotcha] JWT library requires utc=True, otherwise timezone issues in production
- [pattern] Password hashing uses bcrypt with cost factor 12

class CoderAgent:
    """
    Coder 智能体，包含文件路径模糊匹配能力。
    """

    def _build_file_index(self, worktree_path: Path) -> dict[str, Path]:
        """构建文件名到完整路径的索引。"""
        index = {}
        for file_path in worktree_path.rglob("*"):
            if file_path.is_file():
                # 多种索引键：完整路径、文件名、相对路径
                index[str(file_path)] = file_path
                index[file_path.name] = file_path
                index[str(file_path.relative_to(worktree_path))] = file_path
        return index

    def _score_and_select(
        self,
        target: str,
        file_index: dict,
    ) -> Optional[Path]:
        """
        模糊匹配：找到最接近的文件路径。

        评分维度：
        - 精确匹配：+100
        - 文件名匹配：+50
        - 路径包含目标：+30
        - 编辑距离相近：+20
        """
        best_score = 0
        best_path = None

        target_lower = target.lower()
        target_name = os.path.basename(target_lower)

        for key, path in file_index.items():
            key_lower = key.lower()
            score = 0

            if key_lower == target_lower:
                return path  # 精确匹配，直接返回

            if os.path.basename(key_lower) == target_name:
                score += 50

            if target_lower in key_lower or key_lower in target_lower:
                score += 30

            # 简单编辑距离估算
            common_chars = len(set(key_lower) & set(target_lower))
            score += common_chars * 2

            if score > best_score:
                best_score = score
                best_path = path

        # 只有分数足够高才返回（避免误匹配）
        return best_path if best_score >= 20 else None

# lab_08_context.py

import re
import os
from pathlib import Path
from dataclasses import dataclass, field
from collections import Counter, defaultdict
from typing import Optional

@dataclass
class FileMatch:
    path: str
    score: float
    matching_keywords: list[str]
    sample_lines: list[str]  # 最多 3 行匹配样本

class KeywordExtractor:
    STOP_WORDS = {
        "the", "a", "an", "is", "are", "in", "on", "for",
        "implement", "add", "create", "build", "make", "use",
    }

    def extract(self, task: str) -> list[str]:
        """
        TODO Part A: 从任务描述提取搜索关键词

        要求：
        1. 按空格和标点分词
        2. 过滤停用词和短词（< 3字符）
        3. 处理驼峰和下划线命名（loginUser → login, user）
        4. 去重，返回关键词列表

        示例：
        "implement user loginAPI with emailValidation"
        → ["user", "login", "api", "email", "validation", "loginapi", "emailvalidation"]
        """
        pass


class CodeSearcher:
    # 跳过的目录
    SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "dist", "build"}
    # 代码文件扩展名
    CODE_EXTENSIONS = {".py", ".ts", ".js", ".tsx", ".jsx", ".go", ".java"}

    def __init__(self, project_dir: Path):
        self.project_dir = project_dir

    def search(
        self,
        keywords: list[str],
        max_results: int = 20,
    ) -> list[FileMatch]:
        """
        TODO Part A: 在项目目录中搜索包含关键词的代码文件

        要求：
        1. 递归遍历项目目录（跳过 SKIP_DIRS）
        2. 只搜索 CODE_EXTENSIONS 中的文件类型
        3. 对每个文件：
           a. 统计每个关键词出现次数
           b. 计算综合分数（每个词贡献 min(count, 10)）
           c. 记录匹配行样本（每词最多 2 行，行内容截断到 100 字符）
        4. 按分数排序，返回 max_results 个最相关文件

        注意：使用 errors="ignore" 打开文件，避免编码错误
        """
        pass

class FileCategorizer:
    # 实现文件的路径关键词
    IMPL_KEYWORDS = ["handler", "service", "controller", "view", "route", "api", "endpoint"]
    # 参考文件的路径关键词
    REF_KEYWORDS = ["model", "schema", "config", "setting", "util", "helper", "test", "type"]

    def categorize(
        self,
        matches: list[FileMatch],
        top_n_modify: int = 5,
        top_n_reference: int = 10,
    ) -> tuple[list[FileMatch], list[FileMatch]]:
        """
        TODO Part B: 将搜索结果分类为"待修改"和"参考"两组

        分类规则：
        1. 路径中包含 IMPL_KEYWORDS → 待修改
        2. 路径中包含 REF_KEYWORDS → 参考
        3. 分数前 1/3 的文件 → 待修改
        4. 其余 → 参考

        限制结果数量：
        - 待修改最多 top_n_modify 个
        - 参考最多 top_n_reference 个
        """
        pass


class PatternDiscoverer:
    """从参考文件中发现代码约定和模式。"""

    PATTERNS = {
        "decorators": r"@(\w+)\(",
        "exceptions": r"raise\s+(\w+Error|\w+Exception)",
        "imports": r"^(?:from|import)\s+(\S+)",
    }

    def discover(self, reference_files: list[FileMatch], project_dir: Path) -> dict[str, list[str]]:
        """
        TODO Part B: 从参考文件中提取频繁出现的代码模式

        要求：
        1. 对每个模式类型，统计在参考文件中出现的实例
        2. 只返回出现次数 >= 2 的模式（过滤噪音）
        3. 每种模式类型最多返回 5 个实例
        4. 返回格式：{"decorators": ["@require_auth", "@cached"], ...}
        """
        pass

@dataclass
class TaskContext:
    task: str
    keywords: list[str]
    files_to_modify: list[FileMatch]
    files_to_reference: list[FileMatch]
    patterns: dict[str, list[str]]

    def to_prompt_text(self, max_chars: int = 3000) -> str:
        """
        TODO Part C: 将 TaskContext 序列化为提示词文本

        格式：
        ## Task Context

        **Keywords Identified:** login, user, email, password

        **Files to Modify:**
        - src/auth/handlers.py (score: 23.0)
          Matches: login, user, auth
          Sample: `def login_user(request):...`

        **Reference Files:**
        - src/models/user.py (score: 18.0)
          ...

        **Code Patterns:**
        - decorators: @require_auth, @cached
        - exceptions: AuthenticationError, ValidationError

        要求：
        1. 严格控制总字符数不超过 max_chars
        2. 如果内容太多，优先保留 files_to_modify，其次 reference，最后 patterns
        3. 截断时在文本末尾加上 "...[truncated]" 标记
        """
        pass


def build_context(task: str, project_dir: Path) -> TaskContext:
    """
    主入口：为任务构建完整上下文。
    """
    extractor = KeywordExtractor()
    searcher = CodeSearcher(project_dir)
    categorizer = FileCategorizer()
    discoverer = PatternDiscoverer()

    keywords = extractor.extract(task)
    matches = searcher.search(keywords)
    to_modify, to_reference = categorizer.categorize(matches)
    patterns = discoverer.discover(to_reference, project_dir)

    return TaskContext(
        task=task,
        keywords=keywords,
        files_to_modify=to_modify,
        files_to_reference=to_reference,
        patterns=patterns,
    )


def test_context_builder():
    """使用 Auto-Claude 自己的代码库测试上下文构建。"""
    import tempfile, shutil

    # 使用当前项目目录
    project_dir = Path(__file__).parent.parent  # 调整为实际项目路径

    task = "implement user authentication with JWT and session management"
    context = build_context(task, project_dir)

    print(f"关键词: {context.keywords}")
    print(f"待修改文件: {len(context.files_to_modify)} 个")
    print(f"参考文件: {len(context.files_to_reference)} 个")
    print(f"发现模式: {list(context.patterns.keys())}")
    print("\n=== 提示词文本 ===")
    prompt = context.to_prompt_text(max_chars=2000)
    print(prompt)
    print(f"\n字符数: {len(prompt)}")

    # 验证
    assert len(context.keywords) > 0, "应提取到关键词"
    assert len(context.files_to_modify) <= 5, "待修改文件不超过 5 个"
    assert len(context.files_to_reference) <= 10, "参考文件不超过 10 个"
    assert len(prompt) <= 2000, "提示词文本不超过 2000 字符"
    print("\n✓ 所有验证通过")


if __name__ == "__main__":
    test_context_builder()