28.2 核心组件详解

核心组件概述#

Claude Code 的核心组件是其功能实现的基础，包括意图识别、任务规划、工具调度等关键模块。深入理解这些组件对于掌握Claude Code的工作原理至关重要。

意图识别器#

1. 意图识别原理#

markdown
复制
# 示例：意图识别器

用户请求：
"实现一个意图识别器"

Claude Code 生成的代码：

```python
from typing import Dict, List, Any, Optional
from datetime import datetime
import logging
import re

logger = logging.getLogger(__name__)

class Intent:
    """意图"""

    def __init__(self, name: str, description: str, confidence: float = 0.0):
        self.name = name
        self.description = description
        self.confidence = confidence
        self.entities: Dict[str, Any] = {}
        self.metadata: Dict[str, Any] = {}

class IntentRecognizer:
    """意图识别器注册意图"""
        self.intents[name] = {
            'name': name,
            'description': description,
            'patterns': patterns or []
        }

        if patterns:
            self.patterns[name] = patterns

        logger.info(f"Intent registered: {name}")

    def recognize(self, text: str) -> Intent:
        """识别意图"""
        scores = {}

        # 基于模式匹配
        for intent_name, patterns in self.patterns.items():
            score = self._match_patterns(text, patterns)
            scores[intent_name] = score

        # 基于关键词匹配
        for intent_name, intent_info in self.intents.items():
            score = self._match_keywords(text, intent_info)
            scores[intent_name] = max(scores.get(intent_name, 0), score)

        # 基于模型预测
        if self.models:
            model_scores = self._predict_with_model(text)
            for intent_name, score in model_scores.items():
                scores[intent_name] = max(scores.get(intent_name, 0), score)

        # 选择最高分的意图
        if not scores:
            return Intent("unknown", "Unknown intent", 0.0)

        best_intent_name = max(scores, key=scores.get)
        confidence = scores[best_intent_name]

        # 提取实体
        entities = self._extract_entities(text, best_intent_name)

        intent = Intent(
            name=best_intent_name,
            description=self.intents[best_intent_name]['description'],
            confidence=confidence
        )
        intent.entities = entities

        return intent

    def _match_patterns(self, text: str, patterns: List[str]) -> float:
        """匹配模式"""
        max_score = 0.0

        for pattern in patterns:
            try:
                if re.search(pattern, text, re.IGNORECASE):
                    max_score = max(max_score, 0.9)
            except re.error:
                continue

        return max_score

    def _match_keywords(self, text: str, intent_info: Dict[str, Any]) -> float:
        """匹配关键词"""
        keywords = intent_info.get('keywords', [])
        if not keywords:
            return 0.0

        text_lower = text.lower()
        matched = sum(1 for keyword in keywords if keyword.lower() in text_lower)

        return matched / len(keywords) if keywords else 0.0

    def _predict_with_model(self, text: str) -> Dict[str, float]:
        """使用模型预测"""
        # 这里可以使用机器学习模型进行预测
        # 简化版本：返回空字典
        return {}

    def _extract_entities(self, text: str, intent_name: str) -> Dict[str, Any]:
        """提取实体"""
        entities = {}

        # 根据意图类型提取实体使用示例
recognizer = IntentRecognizer()

# 注册意图识别意图
text = "Please read the file config.txt"
intent = recognizer.recognize(text)

print(f"Intent: {intent.name}")

print(f"Confidence: {intent.confidence}") print(f"Entities: {intent.entities}")

bash
复制
> **意图识别器特点**:
> - 支持多种识别方法
> - 模式匹配
> - 关键词匹配
> - 实体提取

## 任务规划器

### 1. 任务规划原理

```python
# 示例：任务规划器
用户请求：
"实现一个任务规划器"
Claude Code 生成的代码：
```python
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

class Task:
    """任务任务规划器添加任务"""
        self.tasks[task.id] = task
        self.task_graph[task.id] = task.dependencies
        logger.info(f"Task added: {task.id}")

    def create_plan(self, goal: str, context: Dict[str, Any]) -> List[Task]:
        """创建执行计划"""
        # 根据目标和上下文生成任务
        tasks = self._generate_tasks(goal, context)

        # 添加任务
        for task in tasks:
            self.add_task(task)

        # 生成执行顺序
        self.execution_plan = self._topological_sort()

        # 返回按顺序排列的任务
        return [self.tasks[task_id] for task_id in self.execution_plan]

    def _generate_tasks(self, goal: str, context: Dict[str, Any]) -> List[Task]:
        """生成任务"""
        tasks = []

        # 根据目标类型生成任务
        if 'read' in goal.lower() and 'file' in goal.lower():
            tasks = self._generate_file_reading_tasks(goal, context)
        elif 'write' in goal.lower() and 'file' in goal.lower():
            tasks = self._generate_file_writing_tasks(goal, context)
        elif 'execute' in goal.lower() and 'code' in goal.lower():
            tasks = self._generate_code_execution_tasks(goal, context)
        else:
            tasks = self._generate_default_tasks(goal, context)

        return tasks

    def _generate_file_reading_tasks(self, goal: str, context: Dict[str, Any]) -> List[Task]:
        """生成文件读取任务"""
        file_path = context.get('file_path', 'default.txt')

        tasks = [
            Task(
                task_id="validate_file_path",
                name="Validate File Path",
                description=f"Validate file path: {file_path}",
                action="validate_file_path"
            ),
            Task(
                task_id="read_file",
                name="Read File",
                description=f"Read file: {file_path}",
                action="read_file",
                dependencies=["validate_file_path"]
            ),
            Task(
                task_id="parse_content",
                name="Parse Content",
                description="Parse file content",
                action="parse_content",
                dependencies=["read_file"]
            )
        ]

        return tasks

    def _generate_file_writing_tasks(self, goal: str, context: Dict[str, Any]) -> List[Task]:
        """生成文件写入任务"""
        file_path = context.get('file_path', 'output.txt')
        content = context.get('content', '')

        tasks = [
            Task(
                task_id="validate_file_path",
                name="Validate File Path",
                description=f"Validate file path: {file_path}",
                action="validate_file_path"
            ),
            Task(
                task_id="prepare_content",
                name="Prepare Content",
                description="Prepare content for writing",
                action="prepare_content",
                dependencies=["validate_file_path"]
            ),
            Task(
                task_id="write_file",
                name="Write File",
                description=f"Write content to file: {file_path}",
                action="write_file",
                dependencies=["prepare_content"]
            ),
            Task(
                task_id="verify_write",
                name="Verify Write",
                description="Verify file was written successfully",
                action="verify_write",
                dependencies=["write_file"]
            )
        ]

        return tasks

    def _generate_code_execution_tasks(self, goal: str, context: Dict[str, Any]) -> List[Task]:
        """生成代码执行任务"""
        code = context.get('code', '')
        language = context.get('language', 'python')

        tasks = [
            Task(
                task_id="validate_code",
                name="Validate Code",
                description="Validate code syntax",
                action="validate_code"
            ),
            Task(
                task_id="setup_environment",
                name="Setup Environment",
                description=f"Setup {language} execution environment",
                action="setup_environment",
                dependencies=["validate_code"]
            ),
            Task(
                task_id="execute_code",
                name="Execute Code",
                description="Execute the code",
                action="execute_code",
                dependencies=["setup_environment"]
            ),
            Task(
                task_id="capture_output",
                name="Capture Output",
                description="Capture execution output",
                action="capture_output",
                dependencies=["execute_code"]
            )
        ]

        return tasks

    def _generate_default_tasks(self, goal: str, context: Dict[str, Any]) -> List[Task]:
        """生成默认任务"""
        tasks = [
            Task(
                task_id="analyze_goal",
                name="Analyze Goal",
                description=f"Analyze goal: {goal}",
                action="analyze_goal"
            ),
            Task(
                task_id="execute_goal",
                name="Execute Goal",
                description=f"Execute goal: {goal}",
                action="execute_goal",
                dependencies=["analyze_goal"]
            )
        ]

        return tasks

    def _topological_sort(self) -> List[str]:
        """拓扑排序"""
        visited = set()
        result = []

        def visit(task_id: str):
            if task_id in visited:
                return

            visited.add(task_id)

            # 先访问依赖
            for dep_id in self.task_graph.get(task_id, []):
                visit(dep_id)

            result.append(task_id)

        for task_id in self.tasks:
            visit(task_id)

        return result

    def update_task_status(self, task_id: str, status: str, result: Any = None, error: str = None):
        """更新任务状态"""
        task = self.tasks.get(task_id)
        if not task:
            logger.warning(f"Task not found: {task_id}")
            return

        task.status = status

        if status == "in_progress":
            task.started_at = datetime.utcnow()
        elif status == "completed":
            task.completed_at = datetime.utcnow()
            task.result = result
        elif status == "failed":
            task.completed_at = datetime.utcnow()
            task.error = error

        logger.info(f"Task {task_id} status updated to {status}")

    def get_plan_status(self) -> Dict[str, Any]:
        """获取计划状态"""
        total_tasks = len(self.tasks)
        completed_tasks = len([t for t in self.tasks.values() if t.status == "completed"])
        failed_tasks = len([t for t in self.tasks.values() if t.status == "failed"])
        in_progress_tasks = len([t for t in self.tasks.values() if t.status == "in_progress"])
        pending_tasks = len([t for t in self.tasks.values() if t.status == "pending"])

        return {
            'total_tasks': total_tasks,
            'completed_tasks': completed_tasks,
            'failed_tasks': failed_tasks,
            'in_progress_tasks': in_progress_tasks,
            'pending_tasks': pending_tasks,
            'progress': completed_tasks / total_tasks if total_tasks > 0 else 0
        }

# 使用示例
planner = TaskPlanner()

# 创建计划
goal = "Read the file config.txt"
context = {'file_path': 'config.txt'}
plan = planner.create_plan(goal, context)

print("Execution plan:")
for task in plan:
    print(f"  - {task.name}: {task.description}")

# 更新任务状态
planner.update_task_status("validate_file_path", "completed")
planner.update_task_status("read_file", "in_progress")

# 获取计划状态
status = planner.get_plan_status()
print(f"\nPlan status: {status}")

python
复制
# 示例：工具调度器用户请求：
"实现一个工具调度器"Claude Code 生成的代码：
```python
from typing import Dict, List, Any, Optional
from datetime import datetime
import logging
import asyncio

logger = logging.getLogger(__name__)
classTool:"""工具"""
def__init__(self, tool_id:str, name:str, description:str, function:callable):        self.id= tool_id
        self.name = name
        self.description = description
        self.function = function
        self.status ="idle"        self.last_used: Optional[datetime]=None        self.usage_count =0
classToolScheduler:"""工具调度器"""
def__init__(self, max_concurrent:int=3):        self.tools: Dict[str, Tool]={}        self.task_queue: asyncio.Queue = asyncio.Queue()        self.running_tasks: Dict[str, asyncio.Task]={}        self.max_concurrent = max_concurrent
        self.running =False
defregister_tool(self, tool: Tool):"""注册工具"""        self.tools[tool.id]= tool
        logger.info(f"Tool registered: {tool.id}")
defschedule_task(self, tool_id:str, parameters: Dict[str, Any])->str:"""调度任务"""        task_id =f"task_{datetime.utcnow().timestamp()}"
        task ={'task_id': task_id,'tool_id': tool_id,'parameters': parameters,'status':'pending','created_at': datetime.utcnow()}
        asyncio.create_task(self.task_queue.put(task))        logger.info(f"Task scheduled: {task_id}")
return task_id

asyncdefstart(self):"""启动调度器"""        self.running =True        logger.info("Tool scheduler started")
# 启动工作线程for i inrange(self.max_concurrent):            asyncio.create_task(self._worker(f"worker-{i}"))
asyncdefstop(self):"""停止调度器"""        self.running =False
# 等待所有任务完成await asyncio.gather(*self.running_tasks.values(), return_exceptions=True)
        logger.info("Tool scheduler stopped")
asyncdef_worker(self, worker_name:str):"""工作线程"""        logger.info(f"{worker_name} started")
while self.running:try:# 获取任务                task =await asyncio.wait_for(                    self.task_queue.get(),                    timeout=1.0)
                tool_id = task['tool_id']                tool = self.tools.get(tool_id)
ifnot tool:                    logger.warning(f"Tool not found: {tool_id}")continue
# 执行任务await self._execute_task(worker_name, task, tool)
except asyncio.TimeoutError:continueexcept Exception as e:                logger.error(f"{worker_name} error: {e}")
        logger.info(f"{worker_name} stopped")
asyncdef_execute_task(self, worker_name:str, task: Dict[str, Any], tool: Tool):"""执行任务"""        task_id = task['task_id']
# 更新工具状态        tool.status ="busy"
try:# 执行工具函数            result =await self._run_tool_function(tool, task['parameters'])
# 更新任务状态            task['status']='completed'            task['result']= result
            task['completed_at']= datetime.utcnow()
# 更新工具统计            tool.last_used = datetime.utcnow()            tool.usage_count +=1
            logger.info(f"{worker_name} completed task {task_id}")
except Exception as e:# 更新任务状态            task['status']='failed'            task['error']=str(e)            task['completed_at']= datetime.utcnow()
            logger.error(f"{worker_name} failed task {task_id}: {e}")
finally:# 更新工具状态            tool.status ="idle"
asyncdef_run_tool_function(self, tool: Tool, parameters: Dict[str, Any])-> Any:"""运行工具函数"""# 检查函数是否是协程函数if asyncio.iscoroutinefunction(tool.function):returnawait tool.function(**parameters)else:# 在线程池中运行同步函数            loop = asyncio.get_event_loop()returnawait loop.run_in_executor(None, tool.function,**parameters)
defget_tool_status(self, tool_id:str)-> Optional[Dict[str, Any]]:"""获取工具状态"""        tool = self.tools.get(tool_id)ifnot tool:returnNone
return{'id': tool.id,'name': tool.name,'status': tool.status,'last_used': tool.last_used.isoformat()if tool.last_used elseNone,'usage_count': tool.usage_count
}
defget_all_tool_status(self)-> Dict[str, Dict[str, Any]]:"""获取所有工具状态"""return{            tool_id: self.get_tool_status(tool_id)for tool_id in self.tools
}
# 使用示例asyncdefmain():"""主函数"""    scheduler = ToolScheduler(max_concurrent=2)
# 注册工具defread_file_tool(file_path:str)->str:"""读取文件工具"""withopen(file_path,'r')as f:return f.read()
defwrite_file_tool(file_path:str, content:str)->str:"""写入文件工具"""withopen(file_path,'w')as f:            f.write(content)returnf"Written to {file_path}"
    scheduler.register_tool(Tool("read_file","Read File","Read file content", read_file_tool))    scheduler.register_tool(Tool("write_file","Write File","Write content to file", write_file_tool))
# 启动调度器await scheduler.start()
# 调度任务    task1 = scheduler.schedule_task("read_file",{'file_path':'test.txt'})    task2 = scheduler.schedule_task("write_file",{'file_path':'output.txt','content':'Hello'})    task3 = scheduler.schedule_task("read_file",{'file_path':'test.txt'})
# 等待任务完成await asyncio.sleep(2)
# 获取工具状态    status = scheduler.get_all_tool_status()print(f"Tool status: {status}")
# 停止调度器await scheduler.stop()
if __name__ =='__main__':    asyncio.run(main())

工具调度器特点:

• 异步任务调度
• 并发控制
• 工具状态管理
• 使用统计