28.2 核心组件详解
约 1499 字大约 5 分钟
核心组件概述
Claude Code 的核心组件是其功能实现的基础，包括意图识别、任务规划、工具调度等关键模块。深入理解这些组件对于掌握Claude Code的工作原理至关重要。
意图识别器
1. 意图识别原理
# 示例：意图识别器
用户请求：
"实现一个意图识别器"
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:
"""意图识别器"""
def __init__(self):
self.intents: Dict[str, Dict[str, Any]] = {}
self.patterns: Dict[str, List[str]] = {}
self.models: Dict[str, Any] = {}
def register_intent(self, name: str, description: str, patterns: List[str] = None):
"""注册意图"""
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 = {}
# 根据意图类型提取实体
if intent_name == 'read_file':
file_match = re.search(r'file\s*["\']?([^"\']+)["\']?', text, re.IGNORECASE)
if file_match:
entities['file_path'] = file_match.group(1)
elif intent_name == 'write_file':
file_match = re.search(r'file\s*["\']?([^"\']+)["\']?', text, re.IGNORECASE)
content_match = re.search(r'content\s*["\']?([^"\']+)["\']?', text, re.IGNORECASE)
if file_match:
entities['file_path'] = file_match.group(1)
if content_match:
entities['content'] = content_match.group(1)
elif intent_name == 'execute_code':
code_match = re.search(r'code\s*["\']?([^"\']+)["\']?', text, re.IGNORECASE)
language_match = re.search(r'language\s*["\']?([^"\']+)["\']?', text, re.IGNORECASE)
if code_match:
entities['code'] = code_match.group(1)
if language_match:
entities['language'] = language_match.group(1)
return entities
# 使用示例
recognizer = IntentRecognizer()
# 注册意图
recognizer.register_intent(
name='read_file',
description='Read a file',
patterns=[r'read\s+file', r'open\s+file', r'cat\s+file'],
keywords=['read', 'file', 'open', 'cat']
)
recognizer.register_intent(
name='write_file',
description='Write to a file',
patterns=[r'write\s+file', r'save\s+file', r'create\s+file'],
keywords=['write', 'file', 'save', 'create']
)
recognizer.register_intent(
name='execute_code',
description='Execute code',
patterns=[r'execute\s+code', r'run\s+code', r'eval\s+code'],
keywords=['execute', 'code', 'run', 'eval']
)
# 识别意图
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}")
任务规划器特点:
- 根据目标生成任务
- 处理任务依赖
- 拓扑排序
- 状态追踪
工具调度器#
1. 工具调度原理#
# 示例：工具调度器
用户请求：
"实现一个工具调度器"
Claude Code 生成的代码：
```python
from typing import Dict, List, Any, Optional
from datetime import datetime
import logging
import asyncio
logger = logging.getLogger(__name__)
class Tool:
"""工具"""
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
class ToolScheduler:
"""工具调度器"""
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
def register_tool(self, tool: Tool):
"""注册工具"""
self.tools[tool.id] = tool
logger.info(f"Tool registered: {tool.id}")
def schedule_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
async def start(self):
"""启动调度器"""
self.running = True
logger.info("Tool scheduler started")
# 启动工作线程
for i in range(self.max_concurrent):
asyncio.create_task(self._worker(f"worker-{i}"))
async def stop(self):
"""停止调度器"""
self.running = False
# 等待所有任务完成
await asyncio.gather(*self.running_tasks.values(), return_exceptions=True)
logger.info("Tool scheduler stopped")
async def _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)
if not tool:
logger.warning(f"Tool not found: {tool_id}")
continue
# 执行任务
await self._execute_task(worker_name, task, tool)
except asyncio.TimeoutError:
continue
except Exception as e:
logger.error(f"{worker_name} error: {e}")
logger.info(f"{worker_name} stopped")
async def _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"
async def _run_tool_function(self, tool: Tool, parameters: Dict[str, Any]) -> Any:
"""运行工具函数"""
# 检查函数是否是协程函数
if asyncio.iscoroutinefunction(tool.function):
return await tool.function(**parameters)
else:
# 在线程池中运行同步函数
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, tool.function, **parameters)
def get_tool_status(self, tool_id: str) -> Optional[Dict[str, Any]]:
"""获取工具状态"""
tool = self.tools.get(tool_id)
if not tool:
return None
return {
'id': tool.id,
'name': tool.name,
'status': tool.status,
'last_used': tool.last_used.isoformat() if tool.last_used else None,
'usage_count': tool.usage_count
}
def get_all_tool_status(self) -> Dict[str, Dict[str, Any]]:
"""获取所有工具状态"""
return {
tool_id: self.get_tool_status(tool_id)
for tool_id in self.tools
}
# 使用示例
async def main():
"""主函数"""
scheduler = ToolScheduler(max_concurrent=2)
# 注册工具
def read_file_tool(file_path: str) -> str:
"""读取文件工具"""
with open(file_path, 'r') as f:
return f.read()
def write_file_tool(file_path: str, content: str) -> str:
"""写入文件工具"""
with open(file_path, 'w') as f:
f.write(content)
return f"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())工具调度器特点:
- 异步任务调度
- 并发控制
- 工具状态管理
- 使用统计
总结
核心组件详解包括：
- 意图识别器: 意图识别原理、识别方法实现
- 任务规划器: 任务规划原理、执行计划生成
- 工具调度器: 工具调度原理、异步执行
通过理解这些核心组件，可以更好地掌握Claude Code的工作原理。
在下一节中，我们将探讨数据流与工作流程。