้กน็ฎ่ทฏๅพ:
claude-code-python/็ถๆ: ๐ง ่ฟ่กไธญ๏ผ็ผบๅฐ memory/hook ๆจกๅ๏ผ ๅฏนๅบๆบ็ : instructkr/claude-code
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ main.py โ
โ (CLI ๅ
ฅๅฃ ยท typer) โ
โโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ QueryEngine โ
โ (ๆ ธๅฟ Agent Loop) โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ while not done: โ โ
โ โ response = call_llm(context) โ โ
โ โ parse tool_calls from response โ โ
โ โ execute tools via registry โ โ
โ โ add results to context โ โ
โ โ repeat โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โโโโโโโโโโโโโโโโโผโโโโโโโโโโโโโโโโ
โผ โผ โผ
โโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโ
โ AgentContext โ โ ToolRegistry โ โ LLM Client โ
โ (ไธไธๆ็ฎก็) โ โ (ๅทฅๅ
ทๆณจๅ) โ โ (API ่ฐ็จ) โ
โโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโ
โ Tool (ๆฝ่ฑกๅบ็ฑป) โ
โ โโ FileReadTool โ
โ โโ FileEditTool โ
โ โโ BashTool โ
โ โโ GlobTool โ
โโโโโโโโโโโโโโโโโโโโโโโ
| ๆจกๅ | ๆไปถ | ่่ดฃ |
|---|---|---|
| QueryEngine | agent/query_engine.py |
Agent ไธปๅพช็ฏ๏ผๅ่ฏทๆฑโ่งฃๆโๆง่กโๅพช็ฏ |
| AgentContext | agent/context.py |
ๆถๆฏๅๅฒใtoken ไผฐ็ฎใไธไธๆๆชๆญ |
| ToolRegistry | tool/registry.py |
ๅ จๅฑๅทฅๅ ทๆณจๅ่กจ๏ผLLM ๆฅ่ฏขๅทฅๅ ท |
| Tool | tool/base.py |
ๅทฅๅ ทๆฝ่ฑกๅบ็ฑป + ๆ้ๆจกๅ |
| ๅ ็ฝฎๅทฅๅ ท | tool/builtin/*.py |
ๅ ทไฝๅทฅๅ ทๅฎ็ฐ |
QueryEngine ๆฏ Agent ็ไธปๅพช็ฏๅผๆ๏ผๅฏนๅบ Claude Code ๆบ็ ไธญ็ src/query.tsใๅฎ้ฉฑๅจๆดไธช Agent ็่ฟ่ฝฌ๏ผ
็จๆท่พๅ
ฅ โ QueryEngine.run() โ LLM API โ ่งฃๆๅๅบ โ ๆง่กๅทฅๅ
ท โ ่ฟๅ็ปๆ
โโโโโโโโโโโโโโโโโโโ
โ ็จๆท่พๅ
ฅๆถๆฏ โ
โโโโโโโโโโฌโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ self.context.add_user_ โ
โ message(msg) โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ while ๅพช็ฏๅผๅง โ
โ iteration++ โ
โ state = THINKING โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ self._call_llm() โ โโโโ ๆ ธๅฟ่ฐ็จ
โ ๆ้ ่ฏทๆฑ: โ
โ - model โ
โ - max_tokens โ
โ - messages (ไธไธๆ) โ
โ - tools (ๅทฅๅ
ทๅฎไน) โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ ่งฃๆ LLM ๅๅบ: โ
โ - content (ๆๆฌๅๅค) โ
โ - tool_calls (ๅทฅๅ
ท่ฐ็จ) โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ self.context.add_assistant_message โ
โ (ๅฐๅๅคๅ ๅ
ฅไธไธๆ) โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ ๆ tool_calls ? โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โโโโโโโโโดโโโโโโโโ
โ Yes โ No
โผ โผ
โโโโโโโโโโ โโโโโโโโโโโโโโโโโโโ
โๆง่กๅทฅๅ
ท โ โ ็ปๆ (break) โ
โๅพช็ฏ โ โโโโโโโโโโโโโโโโโโโ
โโโโโโฌโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ _execute_tool_calls() โ
โ ้ๅๆฏไธช tool_call: โ
โ 1. lookup tool in registryโ
โ 2. check_permission() โ
โ 3. prompt if ASK mode โ
โ 4. tool.execute() โ
โ 5. context.add_tool_resultโ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Check: โ
โ - iteration >= max_iterations โ
โ - _stop_event is set โ
โ - no more tool_calls โ
โโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โโโโโโโโดโโโโโโโ
โ Continue โโโโบ ๅๅฐ while ๅพช็ฏ
โ or Break โ
โโโโโโโโโโโโโโโ
@dataclass
class AgentConfig:
model: str = "claude-sonnet-4-20250514" # LLM ๆจกๅ
max_iterations: int = 100 # ๆๅคงๅพช็ฏๆฌกๆฐ
max_tool_calls_per_iteration: int = 128 # ๅๆฌกๆๅคงๅทฅๅ
ท่ฐ็จๆฐ
temperature: float = 0 # ๆธฉๅบฆๅๆฐ
system_prompt: Optional[str] = None # ็ณป็ปๆ็คบ่ฏ
stream: bool = True # ๆฏๅฆๆตๅผ่พๅบ
verbose: bool = False # ่ฏฆ็ปๆจกๅผclass AgentState(Enum):
IDLE = "idle" # ็ฉบ้ฒ
THINKING = "thinking" # ๆ่ไธญ๏ผ่ฐ็จ LLM๏ผ
TOOL_CALLING = "tool_calling" # ๆง่กๅทฅๅ
ทไธญ
AWAITING_PERMISSION = "awaiting_permission" # ็ญๅพ
็จๆทๆๆ
DONE = "done" # ๅฎๆ
ERROR = "error" # ้่ฏฏasync def _call_llm(self) -> Any:
# 1. ่ทๅๆๅๅปบ LLM ๅฎขๆท็ซฏ
if self.llm_client is None:
api_key = os.environ.get("ANTHROPIC_API_KEY")
self.llm_client = AsyncAnthropic(api_key=api_key)
# 2. ๆ้ ่ฏทๆฑๆถๆฏ
messages = self.context.get_messages() # ไปไธไธๆ่ทๅๅๅฒ
tools = self.tool_registry.get_llm_tools() # ่ทๅๅทฅๅ
ทๅฎไน
request_options = {
"model": self.config.model,
"max_tokens": 8192,
"messages": messages,
"tools": tools,
}
# 3. ๅ้่ฏทๆฑ๏ผๆฏๆๆตๅผ๏ผ
if self.config.stream:
async with self.llm_client.messages.stream(**request_options) as stream:
response = await stream.get_final_message()
else:
response = await self.llm_client.messages.create(**request_options)
return responseasync def _execute_tool_calls(self, tool_calls: List[Dict[str, Any]]) -> None:
for tc in tool_calls:
tool_name = tc["name"]
tool_input = tc["input"]
tool_id = tc["id"]
# Step 1: ไปๆณจๅ่กจๆฅๆพๅทฅๅ
ท
tool = self.tool_registry.get(tool_name)
if tool is None:
self.context.add_tool_result(f"Unknown tool: {tool_name}", tool_id)
continue
# Step 2: ๆ้ๆฃๆฅ
allowed, reason = tool.check_permission()
if not allowed:
self.context.add_tool_result(f"Permission denied: {reason}", tool_id)
continue
# Step 3: ASK ๆจกๅผ้็จๆท็กฎ่ฎค
if tool.permission.mode == PermissionMode.ASK:
if not self._prompt_permission(tool):
self.context.add_tool_result("User denied permission", tool_id)
continue
# Step 4: ๆง่กๅทฅๅ
ท
try:
result = await tool.execute(tool_input)
except Exception as e:
result = ToolResult.err(f"Tool execution error: {e}")
# Step 5: ๅฐ็ปๆๅ ๅ
ฅไธไธๆ
self.context.add_tool_result(result.content, tool_id)Context ็ณป็ปๆฏ Agent ็่ฎฐๅฟไธญๆข๏ผ่ด่ดฃ็ฎก็ๆดไธชๅฏน่ฏ็็ๅฝๅจๆใAgent ๅ็ไธๅๅณ็ญ้ฝๅบไบไธไธๆ๏ผๆฒกๆไธไธๆๅฐฑๆฒกๆๆบ่ฝใ
LLM ๆไธไธๆ็ชๅฃ้ๅถ๏ผClaude ็บฆ 200K tokens๏ผใๅฝๅฏน่ฏ่ถๆฅ่ถ้ฟๆถ๏ผ
้ฎ้ข๏ผ
1. ๅฏน่ฏๅๅฒ่ถๆฅ่ถ้ฟ โ ่ถ
่ฟไธไธๆ็ชๅฃ
2. Token ๆๆฌ่ถๆฅ่ถ้ซ โ ๆตช่ดน้ฑ
3. ๆจกๅๅฎนๆ"้ๅฟ"ๆฉๆไฟกๆฏ โ ๅณ็ญ่ดจ้ไธ้
Context ็ณป็ป่งฃๅณๆนๆก๏ผ
1. ่ท่ธชๆถๆฏๅๅฒ๏ผ่ชๅจ็ฎก็็ๅฝๅจๆ
2. ็ฒพ็กฎไผฐ็ฎ token๏ผๆฅ่ฟไธ้ๆถๆชๆญ
3. ไผๅ
ไฟ็ๆ้่ฆ็ไฟกๆฏ๏ผ็ณป็ปๆ็คบ + ๆ่ฟๅฏน่ฏ๏ผ
class MessageRole(Enum):
"""
ๆถๆฏ็ๅ้่
่ง่ฒ๏ผๅณๅฎไบ LLM ๅฆไฝ็่งฃ่ฟๆกๆถๆฏใ
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~Role~
ๅ็ง่ง่ฒ๏ผ
- system: ็ณป็ปๆไปค๏ผๅช่ฏป๏ผไธ่ฎกๅ
ฅๅทฅๅ
ท่ฐ็จ๏ผ
- user: ็จๆท่พๅ
ฅ
- assistant: LLM ่ชๅทฑ็ๅๅค
- tool: ๅทฅๅ
ทๆง่ก็ปๆ
"""
SYSTEM = "system" # ็ณป็ปๆ็คบ๏ผๆ้่ฆ๏ผๆฐธไธๅ ้ค
USER = "user" # ็จๆทๆถๆฏ
ASSISTANT = "assistant"# LLM ๅๅค๏ผๅซ tool_calls๏ผ
TOOL = "tool" # ๅทฅๅ
ท็ปๆ@dataclass
class Message:
"""
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~Message~ ๆฅๅฃ
ไธๆกๆถๆฏ็ๆ ธๅฟ็ปๆ๏ผ
- role: ่ฐ่ฏด็
- content: ่ฏดไบไปไน
- tool_calls: ๅทฅๅ
ท่ฐ็จๅ่กจ๏ผassistant ๆถๆฏไธ็จ๏ผ
- tool_call_id: ๆฌๆถๆฏๅฏนๅบ็ๅทฅๅ
ท่ฐ็จ ID๏ผtool ๆถๆฏไธ็จ๏ผ
"""
role: MessageRole
content: str = ""
tool_calls: Optional[List[Dict[str, Any]]] = None # assistant ไธ็จ
tool_call_id: Optional[str] = None # tool ไธ็จ
name: Optional[str] = None # ๅทฅๅ
ทๅ๏ผtool ไธ็จ๏ผ
timestamp: float = field(default_factory=time.time) # ๆถ้ดๆณๆถๆฏ็ๅฝๅจๆ็คบไพ๏ผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ ไธไธชๅฎๆด็ Tool Call ๅฏน่ฏ โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ โ
โ [User] โ
โ { role: "user", content: "Read ./README.md" } โ
โ โ
โ [Assistant + ToolCall] โ
โ { โ
โ role: "assistant", โ
โ content: "I'll read the file...", โ
โ tool_calls: [ โ
โ { id: "tool_1", name: "read", input: { path: "./README" } }โ
โ ] โ
โ } โ
โ โ
โ [Tool Result] โ
โ { โ
โ role: "tool", โ
โ content: "# Project\n\nThis is...", โ
โ tool_call_id: "tool_1", โ
โ name: "read" โ
โ } โ
โ โ
โ [Assistant Final] โ
โ { โ
โ role: "assistant", โ
โ content: "The README contains..." โ
โ } โ
โ โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
@dataclass
class ToolCall:
"""
ไปฃ่กจไธๆฌกๅทฅๅ
ท่ฐ็จ่ฏทๆฑใ
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~ToolCall~ ๆฅๅฃ
LLM ็ๆ tool_calls๏ผๆ ผๅผไธบ๏ผ
{
"id": "tool_1", # ๅฏไธๆ ่ฏ
"name": "read", # ๅทฅๅ
ทๅ
"input": { "path": "..." } # ๅทฅๅ
ทๅๆฐ
}
"""
id: str # ๅฏไธ ID
name: str # ๅทฅๅ
ทๅ
input_data: Dict[str, Any] # ๅทฅๅ
ทๅๆฐ@dataclass
class ToolResultBlock:
"""
ๅทฅๅ
ท็ปๆ็ๅ
ๅฎนๅใ
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~ToolResultBlock~
็จไบๅฐๅทฅๅ
ทๆง่ก็ปๆๆ ผๅผๅๅ่ฟๅ็ป LLMใ
"""
type: str = "tool_result"
tool_use_id: str = "" # ๅฏนๅบ็ tool_call id
content: str = "" # ๆง่ก็ปๆๅ
ๅฎนclass AgentContext:
"""
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~Context~ ็ฑป
่่ดฃ๏ผ
1. ๅญๅจๆถๆฏๅๅฒ๏ผ_messages ๅ่กจ๏ผ
2. ๅทฅๅ
ท็ปๆ็ผๅญ๏ผ_tool_results ๅญๅ
ธ๏ผ
3. Token ่ฎกๆฐไธไผฐ็ฎ
4. ไธไธๆๆชๆญ๏ผๅฝ่ถ
่ฟ้ๅถๆถ๏ผ
"""
DEFAULT_MAX_TOKENS = 200_000 # Claude ไธไธๆ็ชๅฃ็บฆ 200K
TOKEN_ESTIMATE_CHARS = 4 # ๅค็จไผฐ็ฎ๏ผ1 token โ 4 ๅญ็ฌฆ
def __init__(self, system_prompt: str = None):
self._messages: List[Message] = []
self._tool_results: Dict[str, str] = {} # tool_id โ result
self.max_tokens = self.DEFAULT_MAX_TOKENS
self._encoder = None
# ๅๅงๅ tiktoken๏ผ็ฒพ็กฎ token ่ฎกๆฐ๏ผ
try:
self._encoder = tiktoken.get_encoding("cl100k_base")
except:
pass # ๅ้ๅฐๅญ็ฌฆไผฐ็ฎdef add_user_message(self, content: str) -> None:
"""ๆทปๅ ็จๆทๆถๆฏ"""
self._messages.append(Message(role=MessageRole.USER, content=content))
def add_assistant_message(self, content: str, tool_calls: List[ToolCall] = None):
"""ๆทปๅ ๅฉๆๆถๆฏ๏ผๅฏๅธฆๅทฅๅ
ท่ฐ็จ๏ผ"""
tc_dicts = [tc.to_dict() for tc in tool_calls] if tool_calls else None
self._messages.append(Message(
role=MessageRole.ASSISTANT,
content=content,
tool_calls=tc_dicts
))
def add_tool_result(self, content: str, tool_call_id: str) -> None:
"""ๆทปๅ ๅทฅๅ
ท็ปๆ"""
self._tool_results[tool_call_id] = content
self._messages.append(Message(
role=MessageRole.TOOL,
content=content,
tool_call_id=tool_call_id
))def estimate_tokens(self, text: str) -> int:
"""
็ฒพ็กฎไผฐ็ฎ token ๆฐ้ใ
ไฝฟ็จ tiktoken๏ผcl100k_base็ผ็ ๅจ๏ผ๏ผ่ฟๆฏ Claude ๅ
ผๅฎน็็ผ็ ใ
1 token โ 4 ๅญ็ฌฆ๏ผ่ฑๆ๏ผ๏ผไธญๆ็บฆ 1-2 ๅญ็ฌฆ/token
"""
if self._encoder:
return len(self._encoder.encode(text))
return len(text) // self.TOKEN_ESTIMATE_CHARS # ๅค็จไผฐ็ฎ
def estimate_total_tokens(self) -> int:
"""ไผฐ็ฎๆดไธชไธไธๆ็ๆป token ๆฐ"""
total = 0
for msg in self._messages:
total += self.estimate_tokens(msg.content)
return totalToken ไผฐ็ฎ็คบไพ๏ผ
่พๅ
ฅ: "Hello, world!"
็ผ็ : [15339, 11, 1917, 0]
Token ๆฐ: 4
่พๅ
ฅ: "# Title\n่ฟๆฏไธญๆๅ
ๅฎน"
็ผ็ : [35, 5767, 4, 24826, 21487, ...]
Token ๆฐ: ~15
def truncate_if_needed(self, keep_recent: int = 20) -> List[Message]:
"""
ๅฝไธไธๆ่ถ
้ๆถ๏ผๆบ่ฝๆชๆญใ
็ญ็ฅ๏ผ
1. ไฟ็็ณป็ปๆถๆฏ๏ผๆฐธไธๅจ๏ผ
2. ไฟ็ๆ่ฟ N ๆกๆถๆฏ
3. ไปไธญ้ดๅผๅงๅ ้คๆๆง็ๆถๆฏ
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~truncateContext~
ไธบไปไน่ฟๆ ท่ฎพ่ฎก๏ผ
- ็ณป็ปๆ็คบๆฏ Agent ็"ไธชๆง"๏ผๅฟ
้กปไฟ็
- ๆ่ฟ็ๆถๆฏๅฏนๅฝๅไปปๅกๆ็ธๅ
ณ
- ๆฉๆๆถๆฏๅฏ่ฝ่ขซ"้ๅฟ"ๅฝฑๅๆๅฐ
"""
removed = []
while self.estimate_total_tokens() > self.max_tokens and len(self._messages) > 2:
# ่ทณ่ฟ็ณป็ปๆถๆฏ๏ผindex 0๏ผ๏ผไปๆๆง็็จๆท/ๅฉๆๆถๆฏๅผๅงๅ
for i, msg in enumerate(self._messages):
if msg.role != MessageRole.SYSTEM:
removed.append(self._messages.pop(i))
break
return removed # ่ฟๅ่ขซๅ ้ค็ๆถๆฏ๏ผๅฏ่ฎฐๅฝๆฅๅฟ๏ผๆชๆญ็คบๆๅพ๏ผ
ๆชๆญๅ๏ผๅ่ฎพ่ถ
้๏ผ:
โโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Sys โ [System] You are Claude Code... โ โโโ ไฟ็
โโโโโโโโผโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Msg โ [User] Read ./README.md โ
โ Msg โ [Assistant] I'll read it... โ
โ Msg โ [Tool] file content... โโโโบ ่ขซๅ ้ค
โ Msg โ [Assistant] The file shows... โโโโบ ่ขซๅ ้ค
โ Msg โ [User] Now edit main.py โโโโบ ่ขซๅ ้ค
โ Msg โ [Assistant] I'll edit... โ โโโ ไฟ็
โ Msg โ [Tool] edit complete โ โโโ ไฟ็
โโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
ๆชๆญๅ:
โโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Sys โ [System] You are Claude Code... โ
โโโโโโโโผโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ Msg โ [Assistant] I'll edit... โ
โ Msg โ [Tool] edit complete โ
โโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
def get_messages(self) -> List[Dict[str, Any]]:
"""
ๅฐๆถๆฏๆ ผๅผๅไธบ LLM API ๆ ผๅผใ
ๅฏนๅบ Claude Code ๆบ็ : src/context.ts ~formatMessages~
่ฟๅๆ ผๅผ็ฌฆๅ Anthropic API ่ฆๆฑ๏ผ
[
{ "role": "system", "content": "..." },
{ "role": "user", "content": "..." },
{ "role": "assistant", "content": "...", "tool_calls": [...] },
{ "role": "tool", "content": "...", "tool_call_id": "..." }
]
"""
return [msg.to_dict() for msg in self._messages]โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ QueryEngine.run() โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ 1. ็จๆท่พๅ
ฅ โ
โ engine.context.add_user_message("Read ./README.md") โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ 2. ้ฆๆฌก LLM ่ฐ็จ โ
โ messages = context.get_messages() # ่ทๅๆ ผๅผๅๆถๆฏ โ
โ response = await call_llm(messages, tools) โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ 3. LLM ่ฟๅ tool_call โ
โ context.add_assistant_message(response.content, tool_calls) โ
โ โ
โ # ๆญคๆถไธไธๆๅทฒๅ
ๅซ: โ
โ # - System โ
โ # - User: Read ./README.md โ
โ # - Assistant: I'll read it... + tool_calls โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ 4. ๆง่กๅทฅๅ
ท โ
โ result = await tool.execute(input) โ
โ context.add_tool_result(result, tool_call_id) โ
โ โ
โ # ไธไธๆๆฐๅข: โ
โ # - Tool: file content (tool_call_id: tool_1) โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ 5. ๅๆฌก LLM ่ฐ็จ๏ผๅธฆไธๅทฅๅ
ท็ปๆ๏ผ โ
โ # LLM ็ๅฐๅทฅๅ
ท็ปๆๅ๏ผ็ปๅบๆ็ปๅๅค โ
โ context.add_assistant_message("The file contains...") โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โผ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ 6. Token ๆฃๆฅ โ
โ if context.estimate_total_tokens() > context.max_tokens: โ
โ context.truncate_if_needed() # ๆบ่ฝๆชๆญ โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
| ๆฌ้กน็ฎ | Claude Code ๆบ็ | ่ฏดๆ |
|---|---|---|
MessageRole |
src/context.ts ~Role~ |
ๆถๆฏ่ง่ฒๆไธพ |
Message |
src/context.ts ~Message~ |
ๅๆกๆถๆฏ |
AgentContext |
src/context.ts ~Context~ |
ไธไธๆ็ฎก็ๅจ |
ToolCall |
src/context.ts ~ToolCall~ |
ๅทฅๅ ท่ฐ็จๅฏน่ฑก |
ToolResultBlock |
src/context.ts ~ToolResultBlock~ |
ๅทฅๅ ท็ปๆๅ |
get_messages() |
src/context.ts ~formatMessages~ |
ๆ ผๅผๅ API ๆถๆฏ |
truncate_if_needed() |
src/context.ts ~truncateContext~ |
ไธไธๆๆชๆญ |
class Tool(ABC):
"""ๆๆๅทฅๅ
ท็ๅบ็ฑป๏ผๅฎไน็ปไธๆฅๅฃ"""
name: str = "" # ๅทฅๅ
ทๅฏไธๆ ่ฏ
description: str = "" # ไพ LLM ็่งฃ็ๆ่ฟฐ
permission: Permission = Permission() # ๆ้้
็ฝฎ
@abstractmethod
def get_input_schema(self) -> Dict[str, Any]:
"""่ฟๅ JSON Schema๏ผLLM ๆฎๆญคๆ้ ่ฐ็จๅๆฐ"""
raise NotImplementedError
@abstractmethod
async def execute(self, input_data: Dict[str, Any]) -> ToolResult:
"""ๆ ธๅฟๆง่ก้ป่พ"""
raise NotImplementedErrorclass PermissionMode(Enum):
ASK = "ask" # ๆฏๆฌกๆง่กๅ่ฏข้ฎ็จๆท
AUTOMATIC = "automatic" # ่ชๅจๆง่ก๏ผๅฏไฟกๆไฝ๏ผ
NEVER = "never" # ็ฆ็จ
class PermissionScope(Enum):
READ = "read" # ๅช่ฏปๆไฝ
WRITE = "write" # ๅๆไฝ
NETWORK = "network" # ็ฝ็ป่ฎฟ้ฎ
ENVIRONMENT = "environment" # ็ฏๅขๅ้
ALL = "all" # ๅ
จ้จๆ้| ๅทฅๅ ท | ๆไปถ | ๆ ธๅฟๅ่ฝ |
|---|---|---|
| BashTool | tool/builtin/bash.py |
ๆง่ก shell ๅฝไปค |
| FileReadTool | tool/builtin/file_read.py |
่ฏปๅๆไปถๅ ๅฎน |
| FileEditTool | tool/builtin/file_edit.py |
Search-replace ็ผ่พ |
| GlobTool | tool/builtin/glob.py |
ๆจกๅผๅน้ ๆไปถ |
class FileEditTool(Tool):
"""
้่ฟ 'old_text' ็ฒพ็กฎๅน้
ๅฎ็ฐๆไปถ็ผ่พ
LLM ๅฟ
้กปๆไพ exact old_text ๆฅๅฎไฝ็ผ่พไฝ็ฝฎ
"""
async def execute(self, input_data: Dict[str, Any]) -> ToolResult:
file_path = input_data["file_path"]
old_text = input_data["old_text"]
new_text = input_data.get("new_text", "")
# 1. ่ฏปๅๅๆไปถ
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
# 2. ้ช่ฏ old_text ็ฒพ็กฎๅน้
if old_text not in content:
return ToolResult.err(f"old_text not found in file")
# 3. ๆง่กๆฟๆข
new_content = content.replace(old_text, new_text, 1)
# 4. ๅๅๆไปถ
with open(file_path, "w", encoding="utf-8") as f:
f.write(new_content)
return ToolResult.ok(f"Edited {file_path}")class ToolRegistry:
"""ๅ
จๅฑๅทฅๅ
ทๆณจๅ่กจ๏ผๅไพๆจกๅผ๏ผ"""
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
"""ๆณจๅๅทฅๅ
ท"""
self._tools[tool.name] = tool
def get(self, name: str) -> Optional[Tool]:
"""ๆ นๆฎๅ็งฐ่ทๅๅทฅๅ
ท"""
return self._tools.get(name)
def get_llm_tools(self) -> List[Dict[str, Any]]:
"""่ทๅๆๆๅทฅๅ
ท็ LLM ๅฎไน"""
return [tool.get_metadata() for tool in self._tools.values()]# main.py ๅฏๅจๆถ
registry = ToolRegistry()
# ๆณจๅๅ
็ฝฎๅทฅๅ
ท
registry.register(FileReadTool())
registry.register(FileEditTool())
registry.register(BashTool())
registry.register(GlobTool())
# ๅฐๆณจๅ่กจๆณจๅ
ฅ QueryEngine
engine = QueryEngine(tool_registry=registry)| ๆฌ้กน็ฎ | Claude Code ๆบ็ | ่ฏดๆ |
|---|---|---|
agent/query_engine.py |
src/query.ts |
ๆ ธๅฟ QueryEngine ็ฑป |
agent/context.py |
src/context.ts |
Context ็ฑป + Message |
tool/base.py |
src/Tool.ts |
Tool ๆฝ่ฑก็ฑป + PermissionModel |
tool/registry.py |
src/tools.ts |
ๅทฅๅ ทๆณจๅ่กจ |
tool/builtin/*.py |
src/tools/*.ts |
ๅ ็ฝฎๅทฅๅ ทๅฎ็ฐ |
src/
โโโ main.tsx # CLI ๅ
ฅๅฃ
โโโ query.ts # QueryEngine ๆ ธๅฟ
โโโ context.ts # ไธไธๆ็ฎก็
โโโ Tool.ts # Tool ๅบ็ฑป
โโโ tools.ts # ๅทฅๅ
ทๆณจๅ่กจ
โโโ tools/ # ๅ
็ฝฎๅทฅๅ
ท (~40 ไธช)
โ โโโ BashTool.ts
โ โโโ FileReadTool.ts
โ โโโ FileEditTool.ts
โ โโโ GlobTool.ts
โ โโโ GrepTool.ts
โ โโโ WebSearchTool.ts
โ โโโ AgentTool.ts # ๅญ Agent
โ โโโ TaskTool.ts # ไปปๅก็ฎก็
โ โโโ ...
โโโ state/ # Zustand ็ถๆ็ฎก็
โโโ coordinator/ # ๅค Agent ๅไฝ
โโโ skill/ # Skill ็ณป็ป
โโโ mcp/ # MCP ๅ่ฎฎ
โโโ hooks/ # ้ฉๅญ็ณป็ป
| ๆจกๅผ | ๅบ็จๅบๆฏ |
|---|---|
| Template Method | Tool ๅบ็ฑปๅฎไน้ชจๆถ๏ผๅญ็ฑปๅฎ็ฐ execute() |
| Strategy | ไธๅๅทฅๅ ทๅฐ่ฃ ไธๅ็ญ็ฅ |
| Registry/Singleton | ๅ จๅฑ ToolRegistry |
| Command | ๆฏไธช Tool ๆฏไธไธชๅฏๆง่ก็ๅฝไปคๅฏน่ฑก |
| Observer | ่ฟๅบฆๆก็ๅฌๅทฅๅ ทๆง่ก็ถๆ |
| Factory | Tool ๅญ็ฑปๅฎไพๅ |
| ๆไปถ | ๅ่ฝ |
|---|---|
memory/base.py |
่ฎฐๅฟๆฝ่ฑกๅบ็ฑป |
memory/session.py |
ไผ่ฏ่ฎฐๅฟ็ฎก็ |
memory/compact.py |
ไธไธๆๅ็ผฉ |
| ๆไปถ | ๅ่ฝ |
|---|---|
hook/base.py |
้ฉๅญๆฝ่ฑกๅบ็ฑป |
hook/registry.py |
้ฉๅญๆณจๅ่กจ |
hook/events.py |
ไบไปถ็ฑปๅๅฎไน |
hook/builtin/session_memory.py |
ไผ่ฏ่ฎฐๅฟ้ฉๅญ |
hook/builtin/command_logger.py |
ๅฝไปคๆฅๅฟ้ฉๅญ |
hook/builtin/boot_md.py |
ๅฏๅจ้ฉๅญ |
cd claaude-code-python
pip install -r requirements.txt
set ANTHROPIC_API_KEY=sk-ant-... # Windows
# export ANTHROPIC_API_KEY=sk-ant-... # Linux/Mac
# ่ฟ่ก
python main.py "Read ./README.md and explain"
python main.py list-toolsๆฅๅ็ๆๆถ้ด: 2026-04-01 ๆๅๆดๆฐ: 2026-04-01 05:22