「無料3,000リクエスト」を使い切る！ 最終回
※再現可能なソースコード全文は文末にあります。
SAKURA Internetは無料で3000リクエストも使えて太っ腹です。
夏休みの自由研究のため、AIエージェント作成の解説の記録！
400くらいリクエストを使っていました。
残りリクエストも夏休みも少なくこれが集大成。これで終わります。
結局3000リクエストは使いきれず余りました。
業務利用でなければ十分な量でした。
可能なら今後も継続して使いたいです！！！
AIエージェントを自作して、やっと「ループエンジニアリング」の意味がわかった
そりゃ、ループエンジニアリングの言い出しっぺのBoris Chernyさんも乗っかったPeter Steinbergerさんもエージェントの開発者。
この視点で語られることをイチアプリ作成で理解するのは無理かあったのだと。
そう気が付きました。
前回LLMチャット作ったので、AIエージェントを自作してみた。
前回、今回どちらも簡易であるわけだけど。(例によって動くコードは後半に)
最初は、
「LLM APIを呼んで、ツールを実行させればいいんでしょ？」
くらいに考えていた。
実際に作ってみる。
すると、思ったより普通に動く。
ファイルを読む。
ファイルを探す。
コードを書く。
コードを編集する。
コマンドを実行する。
そして、結果をまたLLMに渡す。
……あれ？
これ、Claude CodeやCodexの中で起きていることと、かなり近い。
AIエージェントの本体は、LLMそのものではない。
むしろ重要なのは、LLMをどう回すか。回し続けるかだ。
自作してみたら「ループ」が見えた
今回作ったエージェントは、かなりシンプル。
構成はこんな感じ。
ユーザー
↓
System Prompt
↓
Context
↓
LLM
↓
Tool Call
↓
Permission
↓
Tool実行
↓
Tool Result
↓
Contextへ追加
↓
LLM
↓
Tool Call
↓
……
つまり、
LLM → ツール → 結果 → LLM → ツール → 結果……
というループ。
実装がわかりやすい。（2重ループの箇所）
while self.turn_count < self.max_turns:
self.turn_count += 1
messages = self.context.build_messages()
tools = self.context.get_tool_definitions()
response = self.llm.chat(
messages,
tools=tools
)
if response.has_tool_calls():
for tc in response.tool_calls:
result = self.tools.execute(
tc.name,
tc.arguments
)
self.context.add_tool_result(
tc.id,
tc.name,
result
)
else:
final_response = response.content
break
これだけでも、AIエージェントの核心がかなり見える。
AIに一回質問して終わり、ではない。
AIが考える。
↓
ツールを使う。
↓
結果を見る。
↓
もう一度考える。
↓
またツールを使う。
↓
結果を見る。
↓
目的を達成するまで続ける。
この「回し方」がエージェントだった。
そして「4つのエンジニアリング」がつながった
ここで、やっと腑に落ちた。
- プロンプトエンジニアリング
- コンテキストエンジニアリング
- ハーネスエンジニアリング
- ループエンジニアリング
AIエージェントを作るための4つの概念でしょ
① プロンプトエンジニアリング
まず、AIに何をさせるか。
今回のエージェントにもSystem Promptがある。
You are MyAgent, an autonomous coding assistant.
You help users by reading code,
writing code,
editing files,
running commands,
and fixing bugs.
AIに「あなたは何者なのか」「何をするのか」を与える。
これが最初の入口。
② コンテキストエンジニアリング
次に重要なのが、
何をAIに見せるか。
今回の実装ではContextManagerを作った。
System Prompt。
ユーザーの入力。
AIの回答。
Tool Call。
Tool Result。
これらを履歴として保持して、次のLLM呼び出しに渡す。
def build_messages(self):
result = [
{
"role": "system",
"content": self.system_prompt
}
]
result.extend(self.messages)
return result
AIエージェントでは、①プロンプトと②コンテキストが
「①どんなプロンプトを書くか」 と「②次のAIに、何を記憶として渡すか」
として実装された。
③ ハーネスエンジニアリング
AIに自由にやらせるのは危ない。
例えば、rm -rf / なんて実行されたら終わる。
そこで、AIとツールの間にガードを入れる。
DANGEROUS_PATTERNS = [
"rm -rf /",
"rm -rf ~",
"rm -rf *",
"mkfs",
":(){ :|:& };:",
]
のような危険コマンドをブラックリストでブロック。
read_file → auto
list_files → auto
search_files → auto
write_file → ask
edit_file → ask
run_command → ask
のように、ツールごとに権限も設定できる。
AIに何をさせるかだけではなく、
AIが何をしてはいけないか
これがハーネス。
④ そしてループエンジニアリング
ここが今回、一番大きかった発見。
AIエージェントは、
Prompt → LLM → Answer
ではない。
Prompt → LLM → Tool → Result→Context → LLM → Tool → Result → Context → LLM → …
AIの能力を「1回の回答」ではなく「何回回せるか」で引き出す。
今回のMyAgentでは最大ターン数も設定している。
max_turns = 50
エージェントを回す。
while self.turn_count < self.max_turns:
できるだけ少ないループで回数を減らせるのがベスト。ここが設計の要でしょうね。
悪いループは、何やってんだAIになる
LLM → 間違える → LLM → さらに間違える → LLM → もっと間違える
対策は ループ × コンテキスト × ツール × ガードレールの設計が必要
AIエージェントを作ると、LLMの見え方が変わる
ChatGPTのようなチャットだけを使っていると、
「AIが回答している」
ように見える。
でもエージェントを作ってみると、景色が変わる。
実際に起きていることは、
┌──────────────┐
│ LLM │
└──────┬───────┘
│
Tool Call
↓
┌──────────────┐
│ Tool │
└──────┬───────┘
│
Tool Result
↓
┌──────────────┐
│ Context │
└──────┬───────┘
│
└────────→ LLM
これを延々と回している。
だから、
「AIエージェントを作る」＝「LLMを賢くする」
ではない。
「LLMをどういう環境に置いて、どう回すかを設計する」
という話だった。
Claude CodeやCodexがすごい理由も少し見えてきた
ここまで自作してみると、
Claude CodeやCodexのようなCoding Agentが単なる「LLMチャット」ではないことがよくわかる。
重要なのは、
- コンテキスト管理
- ツール定義
- ツール実行
- 権限制御
- エラー処理
- プロンプト
- 状態管理
- ループ制御
- 最大ターン数
- ログ
- ユーザー確認
などを全部まとめて設計していること。
今回の自作エージェントでも、実際に、
ReadFileTool
WriteFileTool
EditFileTool
ListFilesTool
SearchFilesTool
RunCommandTool
というツールを登録している。
そしてLLMが必要に応じて呼び出す。
Coding Agentは「制御システム + LLM + ツール + ループ 」が肝で
制御システムがコンテキストとハーネス。ループはループ。
結局、AIエージェント開発で重要なのは何なのか
今回、自作してみて整理するとこうなった。
| レイヤー | やること |
|---|---|
| プロンプト | AIに何をさせるか |
| コンテキスト | AIに何を見せるか |
| ハーネス | AIをどう安全に制御するか |
| ツール | AIに何ができるようにするか |
| ループ | AIをどう繰り返し動かすか |
そして全部をつなぐと、
┌─────────────┐
│ Prompt │
└──────┬──────┘
↓
┌─────────────┐
│ Context │
└──────┬──────┘
↓
┌─────────────┐
│ LLM │
└──────┬──────┘
↓
┌─────────────┐
│ Tool │
└──────┬──────┘
↓
┌─────────────┐
│ Harness │
└──────┬──────┘
↓
┌─────────────┐
│ Result │
└──────┬──────┘
│
└────────→ Context
│
└──→ LLM
という循環になる。
これが、今回自作して一番腹落ちした部分。
「エージェントを作る」とは何だったのか
最初は、
「PythonでLLM APIを呼んで、ファイル操作できるようにする」
くらいのつもりだった。
でも実際に作ってみたら違った。
AIに能力を与えることではなく、AIが能力を使い続けられる仕組みを作ること。
AIが暴走しないように囲いを作ること。
必要な情報を次のループへ渡すこと。
その全部を設計する。
プロンプトエンジニアリング
コンテキストエンジニアリング
ハーネスエンジニアリング
ループエンジニアリング
という言葉が、やっと一つにつながった。
AIエージェント開発とは、LLMを中心にした「制御ループ」を設計すること。
ここまでわかると、Claude CodeやCodexを見る目も変わる。
「このループ、どう設計してるんだ？」になる。
だから、ここが妙に腹落ち。
ループエンジニアリングの言い出しっぺのBoris Chernyさんも
乗っかったPeter Steinbergerさんもエージェントの開発者。
ここで満足し終了した。
付録 ソースコード全文
クリックで展開
結構ボリューム大きいですが、Linuxで動作確認済み。
#初期化
Myagent.py --init
#設定ファイルにAPIキーを記載
vi config.json
#起動
Myagent.py
次の改善のためソースの更新しました。
APIキー入力のUX改善 → Sakura互換対応・エラー可視化 → ログレベル制御 → read_file改善
Myagent.py
_BUNDLED_PROMPTS = {"01-system": "You are MyAgent, an autonomous coding assistant. You help users by reading code, writing code, editing files, running commands, and fixing bugs.You have access to the following tools to interact with the user's filesystem and environment. Always think step by step before taking action.Guidelines:- Prefer reading files before modifying them- Write clean, correct code- Run commands to verify your work when appropriate- If a task seems complex, break it down into smaller steps- Always report what you did and what the results were", "02-tools": "You can invoke tools by emitting a JSON object in your response with this exact format:```json{\"tool\": \"tool_name\", \"arguments\": {\"arg1\": \"value1\", ...}}```After each tool execution, you will receive the result and can decide the next step.Read the contents of a file.- path (string, required): File path to readWrite content to a file (overwrites existing).- path (string, required): File path- content (string, required): File contentReplace text in an existing file.- path (string, required): File path- old_string (string, required): Text to replace- new_string (string, required): Replacement textList files in a directory.- path (string, required): Directory path- recursive (boolean, optional): List recursivelySearch for a pattern in files using grep-like search.- pattern (string, required): Search pattern- path (string, optional): Directory or file to search (default: current)Execute a shell command in the working directory.- command (string, required): Command to execute- timeout (integer, optional): Timeout in seconds", "03-rules": "1. Think-before-act: Always consider what you need to know before modifying anything.2. Verify after change: After editing a file, verify syntax or run tests if applicable.3. Prefer explicit: Do not make hidden changes; only modify what the user asked.4. Safety first: Never run destructive commands without user confirmation.5. Report clearly: Summarize actions and results in the final response.6. One tool at a time: Wait for the result of each tool before deciding the next.When you finish the task, provide a concise summary of what was done.", "10-observer-guide": "あなたはファイルの内容を深く理解することが目的です。読み取った内容を分析し、重要なポイントを整理してください。", "11-implementer-guide": "あなたはコードの作成・変更を行います。既存コードとの整合性を保ちながら、目的を達成する最小限の変更を心がけてください。", "12-researcher-guide": "あなたはコードベースを探索・調査しています。検索結果を整理し、必要な情報を正確に特定してください。", "13-executor-guide": "あなたはシェルコマンドを実行して結果を確認しています。出力を正確に解釈し、エラーがあれば原因を特定してください。", "14-summarizer-guide": "あなたはこれまでの作業を要約して報告します。重要なポイントと次のアクションを明確にしてください。", "15-archivist-guide": "あなたはこれまでの作業記録を整理し、次回のタスク引継に備えます。重要な決定事項と未解決項目を記録してください。"}
_BUNDLED_SEQ = ["01-system", "02-tools", "03-rules", "---role:observer---", "10-observer-guide", "---role:implementer---", "11-implementer-guide", "---role:researcher---", "12-researcher-guide", "---role:executor---", "13-executor-guide", "---role:summarizer---", "14-summarizer-guide", "---role:archivist---", "15-archivist-guide"]
# Module: myagent
__version__ = "0.7.0"
# Module: myagent.archivist
import json
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
class ContextCompactor:
COMPACTION_THRESHOLD = 0.7
KEEP_RATIO = 0.3
def __init__(self, max_history: int = 100):
self.max_history = max_history
self.compaction_threshold = int(max_history * self.COMPACTION_THRESHOLD)
self._compressed_summary: Optional[str] = None
self._compression_count = 0
self._cumulative_files_accessed: set = set()
self._cumulative_files_modified: set = set()
self._cumulative_tools: set = set()
self._cumulative_errors: List[str] = []
self._cumulative_decisions: List[str] = []
def is_compaction_needed(self, message_count: int) -> bool:
return message_count >= self.compaction_threshold
def compact(self, messages: List[Dict]) -> Tuple[List[Dict], str]:
keep_count = max(int(len(messages) * self.KEEP_RATIO), 4)
recent = messages[-keep_count:]
older = messages[:-keep_count]
for msg in older:
self._extract_metadata(msg)
summary = self._build_summary()
self._compressed_summary = summary
self._compression_count += 1
summary_msg = {
"role": "assistant",
"content": f"[Context compaction #{self._compression_count}]\n{summary}",
}
return [summary_msg] + recent, summary
def _extract_metadata(self, msg: Dict[str, Any]) -> None:
role = msg.get("role", "")
content = msg.get("content", "") or ""
name = msg.get("name", "")
if role == "user":
text = content if len(content) < 80 else content[:77] + "..."
self._cumulative_decisions.append(f"Q: {text}")
elif role == "assistant":
if content and len(content) < 120:
self._cumulative_decisions.append(f"A: {content}")
if "tool_calls" in msg:
for tc in msg["tool_calls"]:
func = tc.get("function", {})
tname = func.get("name", "")
args = func.get("arguments", "{}")
if tname:
self._cumulative_tools.add(tname)
try:
arg_dict = json.loads(args)
path = arg_dict.get("path")
if path:
if tname in ("read_file", "list_files", "search_files"):
self._cumulative_files_accessed.add(path)
elif tname in ("write_file", "edit_file"):
self._cumulative_files_modified.add(path)
except json.JSONDecodeError:
pass
elif role == "tool":
if content.startswith("ERROR:") or content.startswith("PERMISSION_DENIED:"):
self._cumulative_errors.append(f"{name}: {content[:60]}")
def _build_summary(self) -> str:
parts: List[str] = []
if self._cumulative_files_modified:
parts.append(f"Files modified: {', '.join(sorted(self._cumulative_files_modified))}")
if self._cumulative_files_accessed:
parts.append(f"Files accessed: {', '.join(sorted(self._cumulative_files_accessed))}")
if self._cumulative_tools:
parts.append(f"Tools: {', '.join(sorted(self._cumulative_tools))}")
if self._cumulative_decisions:
parts.append("Exchange:")
for d in self._cumulative_decisions[-6:]:
parts.append(f" {d}")
if self._cumulative_errors:
parts.append("Errors:")
for e in self._cumulative_errors[-3:]:
parts.append(f" {e}")
return "\n".join(parts) if parts else "Previous context condensed."
def save_state(self, summary: str, messages: List[Dict], state_dir: str = ".myagent") -> Path:
state_path = Path(state_dir)
state_path.mkdir(parents=True, exist_ok=True)
state_file = state_path / "state.json"
recent_context = []
for msg in messages[-6:]:
recent_context.append({
"role": msg.get("role", ""),
"content": msg.get("content", "")[:200],
})
data = {
"summary": summary,
"timestamp": datetime.now().isoformat(),
"compression_count": self._compression_count,
"recent_context": recent_context,
}
state_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
return state_file
def load_state(self, state_dir: str = ".myagent") -> Optional[Dict]:
state_file = Path(state_dir) / "state.json"
if state_file.exists():
return json.loads(state_file.read_text(encoding="utf-8"))
return None
def get_summary(self) -> Optional[str]:
return self._compressed_summary
# Module: myagent.llm
import json
import logging
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional
import requests
logger = logging.getLogger("myagent.llm")
class OptimizeStrategy(Enum):
MINIMIZE_TOKENS = "min_tokens"
MINIMIZE_REQUESTS = "min_requests"
class Counters:
def __init__(self):
self.total_requests = 0
self.total_prompt_tokens = 0
self.total_completion_tokens = 0
self.total_tokens = 0
def accumulate(self, usage: Optional[Dict[str, int]]) -> None:
if usage is None:
return
self.total_requests += 1
self.total_prompt_tokens += usage.get("prompt_tokens", 0)
self.total_completion_tokens += usage.get("completion_tokens", 0)
self.total_tokens += usage.get("total_tokens", 0)
def log_current(self) -> None:
logger.info(
"Cumulative: requests=%d, prompt_tokens=%d, completion_tokens=%d, total_tokens=%d",
self.total_requests,
self.total_prompt_tokens,
self.total_completion_tokens,
self.total_tokens,
)
def summary(self) -> str:
return (
f"Requests: {self.total_requests}, "
f"Tokens: {self.total_prompt_tokens}p/{self.total_completion_tokens}c = {self.total_tokens}"
)
@dataclass
class ToolCall:
id: str
name: str
arguments: Dict[str, Any]
@dataclass
class OptimizedRequest:
messages: List[Dict[str, Any]]
tools: Optional[List[Dict]]
model: Optional[str]
@dataclass
class LLMResponse:
content: Optional[str] = None
tool_calls: List[ToolCall] = field(default_factory=list)
raw_response: Optional[Dict] = None
usage: Optional[Dict[str, int]] = None
def has_tool_calls(self) -> bool:
return len(self.tool_calls) > 0
class LLMProvider(ABC):
@abstractmethod
def chat(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> LLMResponse:
pass
@abstractmethod
def optimize_request(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> OptimizedRequest:
pass
@abstractmethod
def get_counters(self) -> Counters:
pass
class OpenAICompatibleProvider(LLMProvider):
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.openai.com/v1",
default_model: str = "gpt-4o-mini",
timeout: int = 120,
optimize: OptimizeStrategy = OptimizeStrategy.MINIMIZE_TOKENS,
convert_tool_role: bool = False,
):
self.api_key = api_key or os.environ.get("OPENAI_API_KEY", "")
self.base_url = base_url.rstrip("/")
self.default_model = default_model
self.timeout = timeout
self.optimize = optimize
self.convert_tool_role = convert_tool_role
self.counters = Counters()
if not self.api_key:
raise ValueError(
"API key is required. Set OPENAI_API_KEY env var or pass api_key."
)
masked_key = self.api_key[:8] + "..." + self.api_key[-4:] if len(self.api_key) > 12 else "..."
logger.info(
"Provider initialized: base_url=%s, model=%s, optimize=%s",
self.base_url,
self.default_model,
self.optimize.name,
)
logger.debug("API key: %s", masked_key)
def _prepare_tools(self, tools: List[Dict]) -> List[Dict]:
return [
{
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t.get("parameters", {"type": "object", "properties": {}}),
},
}
for t in tools
]
def optimize_request(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> OptimizedRequest:
if self.optimize == OptimizeStrategy.MINIMIZE_TOKENS:
return self._optimize_min_tokens(messages, tools, model)
else:
return self._optimize_min_requests(messages, tools, model)
def _optimize_min_tokens(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> OptimizedRequest:
tool_indices = [i for i, m in enumerate(messages) if m.get("role") == "tool"]
if len(tool_indices) <= 4:
return OptimizedRequest(messages=messages, tools=tools, model=model)
keep_from = tool_indices[-4]
optimized = messages[:1] + messages[keep_from:]
saved = len(messages) - len(optimized)
logger.info("Token optimization: removed %d older messages (%d -> %d)", saved, len(messages), len(optimized))
return OptimizedRequest(messages=optimized, tools=tools, model=model)
def _optimize_min_requests(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> OptimizedRequest:
if len(messages) < 3:
return OptimizedRequest(messages=messages, tools=tools, model=model)
optimized: List[Dict[str, Any]] = []
i = 0
while i < len(messages):
msg = messages[i]
role = msg.get("role", "")
if role == "tool":
tool_results: List[str] = []
tool_names: List[str] = []
while i < len(messages) and messages[i].get("role") == "tool":
content = messages[i].get("content", "")
name = messages[i].get("name", "tool")
tool_results.append(f"[{name}]: {content[:150]}" + ("..." if len(content) > 150 else ""))
tool_names.append(name)
i += 1
if len(tool_results) > 1:
merged = {
"role": "user",
"content": f"Tool execution results ({len(tool_results)} tools):\n" + "\n".join(tool_results),
}
optimized.append(merged)
logger.debug("Request optimization: merged %d tool results into 1 message", len(tool_results))
else:
optimized.append(msg)
elif role == "assistant" and i + 1 < len(messages) and messages[i + 1].get("role") != "tool":
optimized.append(msg)
i += 1
else:
optimized.append(msg)
i += 1
saved = len(messages) - len(optimized)
if saved > 0:
logger.info("Request optimization: merged %d messages (%d -> %d)", saved, len(messages), len(optimized))
return OptimizedRequest(messages=optimized, tools=tools, model=model)
def _convert_messages_for_sakura(
self, messages: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
converted: List[Dict[str, Any]] = []
for msg in messages:
if msg.get("role") == "tool":
name = msg.get("name", "tool")
content = msg.get("content", "")
converted.append({
"role": "user",
"content": f"[Tool result from {name}]\n{content}",
})
else:
converted.append(msg)
return converted
def chat(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> LLMResponse:
optimized = self.optimize_request(messages, tools, model)
final_messages = optimized.messages
if self.convert_tool_role or "sakura" in self.base_url.lower():
before = len([m for m in final_messages if m.get("role") == "tool"])
final_messages = self._convert_messages_for_sakura(final_messages)
if before > 0:
logger.debug("Sakura compat: converted %d tool messages to user role", before)
payload: Dict[str, Any] = {
"model": model or self.default_model,
"messages": final_messages,
}
if optimized.tools:
payload["tools"] = self._prepare_tools(optimized.tools)
payload["tool_choice"] = "auto"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
url = f"{self.base_url}/chat/completions"
logger.debug(
"LLM request: model=%s, messages=%d, tools=%s",
payload["model"],
len(payload["messages"]),
"yes" if optimized.tools else "no",
)
logger.debug(
"Request payload:\n%s",
json.dumps(payload, ensure_ascii=False, indent=2),
)
start_time = datetime.now()
try:
response = requests.post(
url, headers=headers, json=payload, timeout=self.timeout
)
response.raise_for_status()
data = response.json()
except requests.exceptions.HTTPError as e:
logger.error("=" * 60)
logger.error("HTTP ERROR: %s", e.response.status_code)
logger.error("Response body: %s", e.response.text)
logger.error("--- Request details ---")
logger.error("URL: %s", url)
logger.error("Headers: %s", {k: v[:20] + "..." if k.lower() == "authorization" and len(v) > 30 else v for k, v in headers.items()})
for i, msg in enumerate(payload.get("messages", [])):
role = msg.get("role", "unknown")
content = msg.get("content", "") or ""
tool_calls = msg.get("tool_calls", [])
content_preview = content[:100].replace("\n", " ") if isinstance(content, str) else str(content)[:100]
if tool_calls:
tc_names = [tc.get("function", {}).get("name", "?") for tc in tool_calls]
logger.error(" msg[%d] role=%s tool_calls=%s", i, role, tc_names)
else:
logger.error(" msg[%d] role=%s content=%s", i, role, content_preview)
logger.error("=" * 60)
raise
except requests.exceptions.RequestException as e:
logger.error("Request failed: %s", e)
raise
elapsed_ms = (datetime.now() - start_time).total_seconds() * 1000
logger.info("LLM response received in %.1f ms", elapsed_ms)
logger.debug(
"Response body:\n%s",
json.dumps(data, ensure_ascii=False, indent=2),
)
choice = data["choices"][0]
message = choice["message"]
content = message.get("content")
tool_calls = []
for tc in message.get("tool_calls", []):
if tc["type"] == "function":
func = tc["function"]
tool_calls.append(
ToolCall(
id=tc["id"],
name=func["name"],
arguments=json.loads(func["arguments"]),
)
)
usage = data.get("usage")
if usage:
logger.info(
"Tokens: prompt=%d, completion=%d, total=%d",
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0),
usage.get("total_tokens", 0),
)
self.counters.accumulate(usage)
self.counters.log_current()
if content:
logger.debug("Response content: %s", content[:500])
if tool_calls:
for tc in tool_calls:
logger.debug("Tool call: %s(%s)", tc.name, json.dumps(tc.arguments, ensure_ascii=False))
return LLMResponse(
content=content,
tool_calls=tool_calls,
raw_response=data,
usage=usage,
)
def get_counters(self) -> Counters:
return self.counters
class DebugProvider(LLMProvider):
def __init__(self, default_model: str = "debug"):
self.default_model = default_model
self.counters = Counters()
def optimize_request(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> OptimizedRequest:
return OptimizedRequest(messages=messages, tools=tools, model=model)
def chat(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> LLMResponse:
logger.info("DEBUG mode: skipping API call")
logger.debug("Messages: %d, Tools: %s", len(messages), "yes" if tools else "no")
if tools:
tool_names = [t["name"] for t in tools]
logger.debug("Available tools: %s", ", ".join(tool_names))
for msg in reversed(messages):
role = msg.get("role", "")
if role in ("user", "assistant"):
content = msg.get("content", "") or ""
logger.debug("Last %s message: %s", role, content[:200])
break
simulated_usage = {
"prompt_tokens": sum(len(str(m.get("content", ""))) // 4 for m in messages),
"completion_tokens": 20,
"total_tokens": sum(len(str(m.get("content", ""))) // 4 for m in messages) + 20,
}
self.counters.accumulate(simulated_usage)
self.counters.log_current()
return LLMResponse(
content="DEBUG: This is a test response. No API call was made.",
usage=simulated_usage,
)
def get_counters(self) -> Counters:
return self.counters
def create_provider(config: Dict[str, Any]) -> LLMProvider:
api_key = config.get("api_key", "")
if api_key == "DEBUG":
logger.info("Using DebugProvider (dry-run, no API calls)")
return DebugProvider(default_model=config.get("model", "debug"))
provider_type = config.get("provider", "openai")
optimize = OptimizeStrategy.MINIMIZE_TOKENS
if provider_type in ("sakura", "local", "ollama"):
optimize = OptimizeStrategy.MINIMIZE_REQUESTS
logger.info("Provider '%s' detected: using MINIMIZE_REQUESTS strategy", provider_type)
if provider_type in ("openai", "deepseek"):
return OpenAICompatibleProvider(
api_key=api_key,
base_url=config.get("base_url", "https://api.openai.com/v1"),
default_model=config.get("model", "gpt-4o-mini"),
timeout=config.get("timeout", 120),
optimize=optimize,
convert_tool_role=config.get("convert_tool_role", False),
)
raise ValueError(f"Unknown provider type: {provider_type}")
# Module: myagent.permission
from typing import Dict
class PermissionManager:
DEFAULT_LEVELS = {
"read_file": "auto",
"list_files": "auto",
"search_files": "auto",
"write_file": "ask",
"edit_file": "ask",
"run_command": "ask",
}
DANGEROUS_PATTERNS = [
"rm -rf /",
"rm -rf ~",
"rm -rf *",
"mkfs",
":(){ :|:& };:",
]
def __init__(self, overrides: Dict[str, str] = None):
self.levels = dict(self.DEFAULT_LEVELS)
if overrides:
self.levels.update(overrides)
def check(self, tool_name: str, arguments: Dict = None) -> bool:
level = self.levels.get(tool_name, "ask")
if tool_name == "run_command" and arguments:
cmd = arguments.get("command", "")
for pattern in self.DANGEROUS_PATTERNS:
if pattern in cmd:
print(f"[Permission] BLOCKED dangerous command: {cmd}")
return False
if level == "auto":
return True
if level == "deny":
print(f"[Permission] Tool '{tool_name}' is denied by policy.")
return False
arg_summary = ""
if arguments:
if tool_name in ("write_file", "edit_file"):
arg_summary = f" path={arguments.get('path', '?')}"
elif tool_name == "run_command":
arg_summary = f" command={arguments.get('command', '?')[:60]}"
prompt = f"Allow {tool_name}{arg_summary}? [y/N/s(always Skip)/a(always Allow)] "
response = input(prompt).strip().lower()
if response in ("a", "always"):
self.levels[tool_name] = "auto"
return True
if response in ("s", "skip"):
self.levels[tool_name] = "deny"
return False
return response in ("y", "yes")
def set_level(self, tool_name: str, level: str) -> None:
if level not in ("auto", "ask", "deny"):
raise ValueError("Level must be auto, ask, or deny")
self.levels[tool_name] = level
def get_summary(self) -> str:
lines = ["Permission levels:"]
for tool, level in sorted(self.levels.items()):
lines.append(f" {tool}: {level}")
return "\n".join(lines)
# Module: myagent.prompt_loader
import os
from pathlib import Path
from typing import Dict, List, Optional, Tuple
class PromptLoader:
DEFAULT_PROMPT = (
"You are MyAgent, an autonomous coding assistant. "
"You help users by reading code, writing code, editing files, "
"running commands, and fixing bugs."
)
_ROLE_SECTION_PREFIX = "---role:"
def __init__(self, prompts_dir: str):
self.prompts_dir = Path(prompts_dir)
self.seq_file = self.prompts_dir / "00-seq.txt"
self._base_ids: Optional[List[str]] = None
self._role_map: Optional[Dict[str, List[str]]] = None
def _parse_sequence(self) -> Tuple[List[str], Dict[str, List[str]]]:
if not self.seq_file.exists():
return [], {}
base_ids: List[str] = []
role_map: Dict[str, List[str]] = {}
current_role: Optional[str] = None
with open(self.seq_file, "r", encoding="utf-8") as f:
for line in f:
stripped = line.split("#")[0].strip()
if not stripped:
continue
if stripped.startswith(self._ROLE_SECTION_PREFIX):
role_name = stripped[
len(self._ROLE_SECTION_PREFIX) :
].rstrip("-").strip()
current_role = role_name
continue
if current_role is not None:
role_map.setdefault(current_role, []).append(stripped)
else:
base_ids.append(stripped)
return base_ids, role_map
def _ensure_parsed(self) -> Tuple[List[str], Dict[str, List[str]]]:
if self._base_ids is None or self._role_map is None:
self._base_ids, self._role_map = self._parse_sequence()
return self._base_ids, self._role_map
def _read_sequence(self) -> List[str]:
base_ids, _ = self._ensure_parsed()
return list(base_ids)
def _load_prompt_file(self, prompt_id: str) -> Optional[str]:
file_path = self.prompts_dir / f"{prompt_id}.txt"
if not file_path.exists():
print(f"[PromptLoader] Warning: {file_path} not found, skipping")
return None
with open(file_path, "r", encoding="utf-8") as f:
lines = f.readlines()
filtered = [line for line in lines if not line.strip().startswith("#")]
return "".join(filtered).strip()
def load(self, separator: str = "\n\n---\n\n") -> str:
if "_BUNDLED_PROMPTS" in globals() and "_BUNDLED_SEQ" in globals():
bp = globals()["_BUNDLED_PROMPTS"]
bs = globals()["_BUNDLED_SEQ"]
parts = []
for pid in bs:
if pid in bp:
parts.append(bp[pid])
if parts:
return separator.join(parts)
return self.DEFAULT_PROMPT
prompt_ids = self._read_sequence()
if not prompt_ids:
print("[PromptLoader] No sequence file found, using default prompt")
return self.DEFAULT_PROMPT
parts = []
for pid in prompt_ids:
content = self._load_prompt_file(pid)
if content is not None:
parts.append(content)
if not parts:
print("[PromptLoader] No prompt files loaded, using default prompt")
return self.DEFAULT_PROMPT
return separator.join(parts)
def load_role(self, role: str, separator: str = "\n\n---\n\n") -> Optional[str]:
if "_BUNDLED_PROMPTS" in globals() and "_BUNDLED_SEQ" in globals():
bp = globals()["_BUNDLED_PROMPTS"]
bs = globals()["_BUNDLED_SEQ"]
role_map: Dict[str, List[str]] = {}
current_role: Optional[str] = None
for entry in bs:
if entry.startswith("role:"):
current_role = entry[5:]
continue
if current_role is not None:
role_map.setdefault(current_role, []).append(entry)
if role in role_map:
parts = []
for pid in role_map[role]:
if pid in bp:
parts.append(bp[pid])
if parts:
return separator.join(parts)
return None
_, role_map = self._ensure_parsed()
if role not in role_map:
return None
parts = []
for pid in role_map[role]:
content = self._load_prompt_file(pid)
if content is not None:
parts.append(content)
if not parts:
return None
return separator.join(parts)
def list_roles(self) -> List[str]:
_, role_map = self._ensure_parsed()
return sorted(role_map.keys())
# Module: myagent.tools.base
from abc import ABC, abstractmethod
from typing import Any, Dict
class Tool(ABC):
@property
@abstractmethod
def name(self) -> str:
pass
@property
@abstractmethod
def description(self) -> str:
pass
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {},
}
@abstractmethod
def execute(self, arguments: Dict[str, Any]) -> str:
pass
def to_definition(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
def _require(self, arguments: Dict[str, Any], key: str) -> Any:
if key not in arguments:
raise ValueError(f"Missing required argument: '{key}' for tool '{self.name}'")
return arguments[key]
# Module: myagent.logger
import logging
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
DEFAULT_LOG_DIR = "logs"
_logger_cache: dict[str, logging.Logger] = {}
def get_logger(name: str) -> logging.Logger:
if name in _logger_cache:
return _logger_cache[name]
logger = logging.getLogger(name)
_logger_cache[name] = logger
return logger
def _resolve_level(level: Optional[str]) -> int:
if level is None:
return logging.NOTSET
mapping = {
"DEBUG": logging.DEBUG,
"INFO": logging.INFO,
"WARNING": logging.WARNING,
"WARN": logging.WARNING,
"ERROR": logging.ERROR,
"CRITICAL": logging.CRITICAL,
}
return mapping.get(level.upper(), logging.NOTSET)
def setup_logging(
log_dir: Optional[str] = None,
console_level: int = logging.INFO,
file_level: int = logging.DEBUG,
log_level: Optional[str] = None,
) -> Path:
root_logger = logging.getLogger("myagent")
root_logger.setLevel(logging.DEBUG)
for handler in list(root_logger.handlers):
root_logger.removeHandler(handler)
if log_level is not None:
resolved = _resolve_level(log_level)
effective_console = resolved
effective_file = resolved
else:
effective_console = console_level
effective_file = file_level
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(effective_console)
console_format = logging.Formatter(
"%(asctime)s [%(levelname)s] %(message)s",
datefmt="%H:%M:%S",
)
console_handler.setFormatter(console_format)
root_logger.addHandler(console_handler)
log_dir_path = Path(log_dir or DEFAULT_LOG_DIR)
log_dir_path.mkdir(parents=True, exist_ok=True)
date_str = datetime.now().strftime("%Y-%m-%d")
log_file = log_dir_path / f"myagent_{date_str}.log"
file_handler = logging.FileHandler(str(log_file), encoding="utf-8")
file_handler.setLevel(effective_file)
file_format = logging.Formatter(
"%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
file_handler.setFormatter(file_format)
root_logger.addHandler(file_handler)
root_logger.info("Logging initialized. File: %s, level: %s", log_file, log_level or "default")
return log_file
# Module: myagent.context
from pathlib import Path
from typing import Any, Dict, List, Optional
class ContextManager:
def __init__(self, system_prompt: str, max_history: int = 100):
self.system_prompt = system_prompt
self.max_history = max_history
self.messages: List[Dict[str, Any]] = []
self.tool_definitions: List[Dict[str, Any]] = []
self._current_role: Optional[str] = None
self.role_prompt: Optional[str] = None
self._compactor = ContextCompactor(max_history=max_history)
def set_tool_definitions(self, tools: List[Dict[str, Any]]) -> None:
self.tool_definitions = tools
def build_messages(self) -> List[Dict[str, Any]]:
result: List[Dict[str, Any]] = [
{"role": "system", "content": self.system_prompt}
]
if self.role_prompt:
result.append({"role": "system", "content": self.role_prompt})
result.extend(self.messages)
return result
def set_role(self, role: Optional[str], role_prompt: Optional[str] = None) -> None:
self._current_role = role
self.role_prompt = role_prompt
def get_role(self) -> Optional[str]:
return self._current_role
def add_user_message(self, content: str) -> None:
self.messages.append({"role": "user", "content": content})
self._trim_history()
def add_assistant_message(
self, content: Optional[str] = None, tool_calls: Optional[List[Dict]] = None
) -> None:
msg: Dict[str, Any] = {"role": "assistant"}
if content is not None:
msg["content"] = content
if tool_calls:
msg["tool_calls"] = tool_calls
self.messages.append(msg)
self._trim_history()
def add_tool_result(self, tool_call_id: str, name: str, result: str) -> None:
self.messages.append(
{
"role": "tool",
"tool_call_id": tool_call_id,
"name": name,
"content": str(result),
}
)
self._trim_history()
def get_tool_definitions(self) -> List[Dict[str, Any]]:
return self.tool_definitions
def get_messages(self) -> List[Dict[str, Any]]:
return list(self.messages)
def clear(self) -> None:
self.messages = []
def _trim_history(self) -> None:
if self._compactor.is_compaction_needed(len(self.messages)):
self.messages, summary = self._compactor.compact(self.messages)
def get_compaction_summary(self) -> Optional[str]:
return self._compactor.get_summary()
def save_state(self) -> Optional["Path"]:
summary = self._compactor.get_summary()
if summary:
return self._compactor.save_state(summary, self.messages)
return None
def load_state(self) -> Optional[Dict]:
return self._compactor.load_state()
def estimate_tokens(self) -> int:
total = 0
for msg in self.messages:
content = msg.get("content", "")
if content:
total += len(content) // 4
return total
# Module: myagent.model_router
import logging
from typing import Any, Dict, List, Optional
logger = logging.getLogger("myagent.model_router")
class ModelRouter:
def __init__(
self,
default_config: Dict[str, Any],
role_configs: Optional[Dict[str, Dict[str, Any]]] = None,
):
self.default_config = default_config
self.role_configs = role_configs or {}
self._providers: Dict[str, LLMProvider] = {}
self._default_provider = self._create_provider("default", default_config)
def _create_provider(self, name: str, config: Dict[str, Any]) -> LLMProvider:
merged = dict(self.default_config)
merged.update(config)
try:
provider = create_provider(merged)
logger.info("ModelRouter: created '%s' provider (model=%s)", name, merged.get("model", "?"))
return provider
except Exception as e:
logger.error("ModelRouter: failed to create '%s' provider: %s", name, e)
raise
def get_provider(self, role: Optional[str] = None) -> LLMProvider:
if role is None or role not in self.role_configs:
return self._default_provider
if role not in self._providers:
self._providers[role] = self._create_provider(role, self.role_configs[role])
return self._providers[role]
def chat(
self,
role: Optional[str],
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
):
provider = self.get_provider(role)
return provider.chat(messages, tools=tools, model=model)
def get_all_counters(self) -> List[Dict[str, Any]]:
results = [{"name": "default", **self._default_provider.get_counters().__dict__}]
for role, provider in self._providers.items():
results.append({"name": role, **provider.get_counters().__dict__})
return results
def get_total_counters(self) -> Counters:
total = Counters()
total.__dict__.update(self._default_provider.get_counters().__dict__)
for provider in self._providers.values():
pc = provider.get_counters()
total.total_requests += pc.total_requests
total.total_prompt_tokens += pc.total_prompt_tokens
total.total_completion_tokens += pc.total_completion_tokens
total.total_tokens += pc.total_tokens
return total
# Module: myagent.tools.registry
from typing import Any, Dict, List
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
if tool.name in self._tools:
raise ValueError(f"Tool '{tool.name}' is already registered")
self._tools[tool.name] = tool
def register_all(self, *tools: Tool) -> None:
for tool in tools:
self.register(tool)
def get(self, name: str) -> Tool:
if name not in self._tools:
raise KeyError(f"Tool '{name}' not found in registry")
return self._tools[name]
def execute(self, tool_name: str, arguments: Dict[str, Any]) -> str:
tool = self.get(tool_name)
try:
return tool.execute(arguments)
except Exception as e:
return f"ERROR: {type(e).__name__}: {e}"
def list_tools(self) -> List[str]:
return list(self._tools.keys())
def get_definitions(self) -> List[Dict[str, Any]]:
return [tool.to_definition() for tool in self._tools.values()]
def __contains__(self, name: str) -> bool:
return name in self._tools
# Module: myagent.tools.search_files
import fnmatch
import math
import os
import re
from typing import Any, Dict, List, Tuple
class SearchFilesTool(Tool):
@property
def name(self) -> str:
return "search_files"
@property
def description(self) -> str:
return (
"Search for a text pattern in files using substring or regex search. "
"Results are ranked by relevance score (higher = more important)."
)
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Text or regex pattern to search for",
},
"path": {
"type": "string",
"description": "Directory or file to search (default: current directory)",
},
"regex": {
"type": "boolean",
"description": "Use regex matching (default: false = literal search)",
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return (default: 50)",
},
"file_pattern": {
"type": "string",
"description": "Glob pattern for files to search (e.g., '*.py')",
},
"rank": {
"type": "boolean",
"description": "Rank results by BM25-like relevance score (default: true)",
},
},
"required": ["pattern"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
pattern = self._require(arguments, "pattern")
path = arguments.get("path", ".")
use_regex = arguments.get("regex", False)
max_results = arguments.get("max_results", 50)
file_pattern = arguments.get("file_pattern")
rank_results = arguments.get("rank", True)
try:
if use_regex:
compiled = re.compile(pattern)
else:
compiled = None
target = os.path.abspath(path)
if os.path.isfile(target):
results = self._search_file(target, pattern, compiled, use_regex)
ranked = self._rank_results(results, pattern, use_regex, rank=rank_results)
return self._format_results(ranked, max_results)
if os.path.isdir(target):
results = self._search_dir(target, pattern, compiled, use_regex, file_pattern, max_results)
ranked = self._rank_results(results, pattern, use_regex, rank=rank_results)
return self._format_results(ranked, max_results)
return f"ERROR: Path not found: {target}"
except re.error as e:
return f"ERROR: Invalid regex pattern: {e}"
except Exception as e:
return f"ERROR: {e}"
def _search_file(self, file_path: str, pattern: str, compiled, use_regex: bool) -> List[Tuple[str, int, str]]:
results = []
if self._is_binary(file_path):
return results
try:
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
for line_num, line in enumerate(f, start=1):
try:
if use_regex:
if compiled.search(line):
results.append((file_path, line_num, line.rstrip("\n")))
else:
if pattern in line:
results.append((file_path, line_num, line.rstrip("\n")))
except UnicodeDecodeError:
continue
except (OSError, PermissionError):
pass
return results
def _search_dir(self, root: str, pattern: str, compiled, use_regex: bool, file_pattern: str, max_results: int) -> List[Tuple[str, int, str]]:
results = []
exclude_dirs = {".git", ".hg", ".svn", "node_modules", "__pycache__", ".venv", "venv", ".claude", ".idea", ".vscode", "dist", "build"}
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if d not in exclude_dirs]
for filename in filenames:
if file_pattern and not fnmatch.fnmatch(filename, file_pattern):
continue
file_path = os.path.join(dirpath, filename)
results.extend(self._search_file(file_path, pattern, compiled, use_regex))
if len(results) >= max_results * 3:
return results[:max_results * 3]
return results
def _rank_results(self, results: List[Tuple[str, int, str]], pattern: str, use_regex: bool, rank: bool = True) -> List[Tuple[str, int, str, float]]:
if not results:
return []
if not rank:
return [(p, n, l, 0.0) for p, n, l in results]
scored = []
pattern_words = self._tokenize(pattern)
pattern_lower = pattern.lower()
for file_path, line_num, line in results:
line_lower = line.lower()
score = self._compute_score(line_lower, pattern_lower, pattern_words, line)
scored.append((file_path, line_num, line, score))
scored.sort(key=lambda x: x[3], reverse=True)
return scored
def _tokenize(self, text: str) -> List[str]:
tokens = re.findall(r"\b\w+\b", text.lower())
return [t for t in tokens if len(t) > 1]
def _compute_score(self, line_lower: str, pattern_lower: str, pattern_words: List[str], raw_line: str) -> float:
score = 0.0
if pattern_lower in line_lower:
score += 10.0
occurrences = line_lower.count(pattern_lower)
score += min(occurrences - 1, 3) * 2.0
line_words = self._tokenize(raw_line)
matched_words = sum(1 for pw in pattern_words if pw in line_words)
if pattern_words:
word_match_ratio = matched_words / len(pattern_words)
score += word_match_ratio * 5.0
stripped = raw_line.strip()
if stripped.startswith("class ") or stripped.startswith("def "):
score += 3.0
if stripped.startswith("import ") or stripped.startswith("from "):
score += 1.0
if stripped.startswith("#"):
score += 0.5
if len(raw_line) > 200:
score *= 0.7
elif len(raw_line) < 20:
score *= 0.9
return score
def _format_results(self, results: List[Tuple[str, int, str, float]], max_results: int) -> str:
if not results:
return "No matches found."
lines = [f"Found {len(results)} matches (top {min(len(results), max_results)} by relevance):"]
display_order = sorted(results[:max_results], key=lambda x: (x[0], x[1]))
for file_path, line_num, line_text, score in display_order:
display_line = line_text
if len(display_line) > 200:
display_line = display_line[:200] + " ... [truncated]"
lines.append(f" {file_path}:{line_num} [score:{score:.1f}] {display_line}")
if len(results) > max_results:
lines.append(f"\n... and {len(results) - max_results} more matches")
return "\n".join(lines)
def _is_binary(self, file_path: str) -> bool:
try:
with open(file_path, "rb") as f:
chunk = f.read(4096)
return b"\x00" in chunk
except:
return True
# Module: myagent.tools.list_files
import os
from typing import Any, Dict
class ListFilesTool(Tool):
@property
def name(self) -> str:
return "list_files"
@property
def description(self) -> str:
return "List files and directories at a given path. Can list recursively."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Directory path to list (default: current directory)",
},
"recursive": {
"type": "boolean",
"description": "List recursively (default: false)",
},
"max_depth": {
"type": "integer",
"description": "Maximum recursion depth (default: 3 for recursive)",
},
},
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = arguments.get("path", ".")
recursive = arguments.get("recursive", False)
max_depth = arguments.get("max_depth", 3)
try:
target = os.path.abspath(path)
if not os.path.exists(target):
return f"ERROR: Path not found: {target}"
if not os.path.isdir(target):
return f"ERROR: Not a directory: {target}"
lines = []
self._list_dir(target, "", lines, recursive, 0, max_depth)
return "\n".join(lines) if lines else "(empty directory)"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
def _list_dir(self, root: str, prefix: str, lines: list, recursive: bool, depth: int, max_depth: int) -> None:
try:
entries = sorted(os.listdir(root))
except PermissionError:
lines.append(f"{prefix}[permission denied]")
return
for i, entry in enumerate(entries):
entry_path = os.path.join(root, entry)
is_last = i == len(entries) - 1
connector = "└── " if is_last else "├── "
indent = " " if is_last else "│ "
if os.path.isdir(entry_path):
lines.append(f"{prefix}{connector}{entry}/")
if recursive and depth < max_depth:
self._list_dir(entry_path, prefix + indent, lines, recursive, depth + 1, max_depth)
elif recursive and depth >= max_depth:
lines.append(f"{prefix}{indent}...")
else:
lines.append(f"{prefix}{connector}{entry}")
# Module: myagent.tools.write_file
import os
from typing import Any, Dict
class WriteFileTool(Tool):
@property
def name(self) -> str:
return "write_file"
@property
def description(self) -> str:
return "Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Can optionally create parent directories."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to write",
},
"content": {
"type": "string",
"description": "Content to write to the file",
},
"create_dirs": {
"type": "boolean",
"description": "Create parent directories if they don't exist (default: true)",
},
},
"required": ["path", "content"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = self._require(arguments, "path")
content = self._require(arguments, "content")
create_dirs = arguments.get("create_dirs", True)
try:
if create_dirs:
parent = os.path.dirname(os.path.abspath(path))
if parent and not os.path.exists(parent):
os.makedirs(parent, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(content)
lines = content.count("\n") + 1 if content else 0
return f"Successfully wrote {len(content)} characters, {lines} lines to {path}"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
# Module: myagent.tools.run_command
import os
import subprocess
import tempfile
from typing import Any, Dict
class RunCommandTool(Tool):
@property
def name(self) -> str:
return "run_command"
@property
def description(self) -> str:
return "Execute a shell command in the current working directory. Use this to run tests, check syntax, or use git."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "Shell command to execute",
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds (default: 60)",
},
"cwd": {
"type": "string",
"description": "Working directory for the command (default: current directory)",
},
},
"required": ["command"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
command = self._require(arguments, "command")
timeout = arguments.get("timeout", 60)
try:
cwd = arguments.get("cwd", os.getcwd())
except FileNotFoundError:
cwd = arguments.get("cwd", "/tmp")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
cwd=cwd,
)
output_lines = []
if result.stdout:
output_lines.append("STDOUT:")
output_lines.append(result.stdout.rstrip())
if result.stderr:
output_lines.append("STDERR:")
output_lines.append(result.stderr.rstrip())
output_lines.append(f"\nExit code: {result.returncode}")
return "\n".join(output_lines)
except subprocess.TimeoutExpired:
return f"ERROR: Command timed out after {timeout} seconds"
except Exception as e:
return f"ERROR: {type(e).__name__}: {e}"
# Module: myagent.tools.edit_file
from typing import Any, Dict
class EditFileTool(Tool):
@property
def name(self) -> str:
return "edit_file"
@property
def description(self) -> str:
return "Replace text in an existing file. old_string must match exactly (including whitespace)."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to edit",
},
"old_string": {
"type": "string",
"description": "Exact text to replace",
},
"new_string": {
"type": "string",
"description": "Replacement text",
},
},
"required": ["path", "old_string", "new_string"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = self._require(arguments, "path")
old_string = self._require(arguments, "old_string")
new_string = self._require(arguments, "new_string")
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
content = f.read()
if old_string not in content:
if old_string.strip() in content.strip():
return (
f"ERROR: old_string not found exactly in {path}. "
"Note: old_string must match exactly including surrounding whitespace."
)
return f"ERROR: old_string not found in {path}"
count = content.count(old_string)
if count > 1:
return (
f"ERROR: Found {count} occurrences of old_string in {path}. "
"Please make the old_string more specific (include more context)."
)
new_content = content.replace(old_string, new_string)
with open(path, "w", encoding="utf-8") as f:
f.write(new_content)
old_lines = old_string.count("\n")
new_lines = new_string.count("\n")
delta = new_lines - old_lines
return f"Successfully edited {path} (+{delta} lines)"
except FileNotFoundError:
return f"ERROR: File not found: {path}"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
# Module: myagent.tools.read_file
from typing import Any, Dict
DEFAULT_PAGE_SIZE = 25
class ReadFileTool(Tool):
@property
def name(self) -> str:
return "read_file"
@property
def description(self) -> str:
return (
"Read the contents of a file at the given path. "
"By default reads the entire file. "
"Use limit=N (or page=N) to read in chunks for large files."
)
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute or relative path to the file",
},
"offset": {
"type": "integer",
"description": f"Line number to start reading from (1-indexed). Ignored if page is set.",
},
"limit": {
"type": "integer",
"description": "Maximum number of lines to read. Set explicitly for large files; omitted reads entire file.",
},
"page": {
"type": "integer",
"description": f"Optional page number (1-indexed). Each page is {DEFAULT_PAGE_SIZE} lines. Only used when pagination is needed.",
},
},
"required": ["path"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = self._require(arguments, "path")
offset = arguments.get("offset")
limit = arguments.get("limit")
page = arguments.get("page")
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
all_lines = f.readlines()
except FileNotFoundError:
return f"ERROR: File not found: {path}"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
total_lines = len(all_lines)
if total_lines == 0:
return "(empty file)"
paginate = page is not None or limit is not None
effective_limit = DEFAULT_PAGE_SIZE
if limit is not None and limit > 0:
effective_limit = limit
if page is not None and page > 0:
start = (page - 1) * effective_limit
elif offset is not None and offset > 0:
start = offset - 1
else:
start = 0
start = max(0, min(start, total_lines))
if paginate:
end = min(start + effective_limit, total_lines)
else:
end = total_lines
if start >= total_lines:
return (
f"ERROR: Requested position exceeds file length "
f"({total_lines} lines). Use a smaller offset/page."
)
selected = all_lines[start:end]
numbered_lines = []
for i, line in enumerate(selected, start=start + 1):
numbered_lines.append(f"{i:4d}: {line}")
result = "".join(numbered_lines)
if paginate and end < total_lines:
total_pages = (total_lines + effective_limit - 1) // effective_limit
current_page = (start // effective_limit) + 1
footer_parts = [
f"... ({total_lines - end} more lines)",
f"[Page {current_page}/{total_pages} — {total_lines} lines total]",
]
if current_page < total_pages:
footer_parts.append(
f"Next: page={current_page + 1} (or offset={end + 1})"
)
result += "\n" + "\n".join(footer_parts)
elif paginate:
current_page = (start // effective_limit) + 1
total_pages = (total_lines + effective_limit - 1) // effective_limit
footer_parts = [
f"[Page {current_page}/{total_pages} — {total_lines} lines total]",
]
result += "\n" + "\n".join(footer_parts)
return result.rstrip()
# Module: myagent.agent
import json
import logging
import os
from typing import Any, Dict, List, Optional, Tuple, Union
logger = logging.getLogger("myagent.agent")
ROLE_MAP: Dict[str, str] = {
"read_file": "observer",
"list_files": "observer",
"write_file": "implementer",
"edit_file": "implementer",
"search_files": "researcher",
"run_command": "executor",
}
KEYWORD_ROLE_MAP: List[Tuple[Tuple[str, ...], str]] = [
(("write", "create", "edit", "add", "implement", "modify", "update"), "implementer"),
(("read", "show", "list", "display", "look", "view"), "observer"),
(("search", "find", "grep", "locate", "lookup"), "researcher"),
(("run", "execute", "build", "test", "deploy", "compile"), "executor"),
(("summarize", "explain", "report", "describe"), "summarizer"),
(("archive", "save", "compress", "compact", "persist"), "archivist"),
(("refactor", "restructure", "rewrite", "improve"), "refactor"),
(("review", "check", "audit", "inspect"), "reviewer"),
(("debug", "fix", "troubleshoot", "diagnose"), "debugger"),
]
class Agent:
def __init__(
self,
llm: Union[LLMProvider, ModelRouter],
tool_registry: ToolRegistry,
permission: PermissionManager,
prompt_loader: PromptLoader,
max_turns: int = 50,
log_level: Optional[str] = None,
):
self.llm = llm
self.tools = tool_registry
self.permission = permission
self.prompt_loader = prompt_loader
self.max_turns = max_turns
self.log_level = log_level
self._using_router = isinstance(llm, ModelRouter)
system_prompt = self.prompt_loader.load()
logger.debug("System prompt length: %d chars", len(system_prompt))
self.context = ContextManager(system_prompt=system_prompt)
self.context.set_tool_definitions(self.tools.get_definitions())
state = self.context.load_state()
if state:
prev_summary = state.get("summary", "")
if prev_summary:
self.context.system_prompt += (
f"\n\n[Previous session #{state.get('compression_count', 0)}]\n"
f"{prev_summary}"
)
logger.info(
"Loaded previous session state (compression #%d)",
state.get("compression_count", 0),
)
self.turn_count = 0
logger.info("Agent initialized. max_turns=%d", max_turns)
def _chat(self, messages, tools=None):
if self._using_router:
return self.llm.chat(self.context.get_role(), messages, tools=tools)
return self.llm.chat(messages, tools=tools)
def _get_counters(self):
if self._using_router:
return self.llm.get_total_counters()
return self.llm.get_counters()
def _resolve_role_from_keywords(self, text: str) -> Optional[str]:
text_lower = text.lower()
for keywords, role in KEYWORD_ROLE_MAP:
if any(kw in text_lower for kw in keywords):
return role
return None
def _resolve_role_from_tool(self, tool_name: str) -> Optional[str]:
return ROLE_MAP.get(tool_name)
def _apply_role(self, role: Optional[str]) -> None:
if role is None:
if self.context.get_role() is not None:
logger.info("Role cleared (back to base)")
self.context.set_role(None)
return
role_prompt = self.prompt_loader.load_role(role)
if role_prompt is None:
if self.context.get_role() is not None:
logger.debug("Role '%s' not defined, clearing previous role", role)
self.context.set_role(None)
return
if self.context.get_role() != role:
logger.info("Role switched: %s", role)
self.context.set_role(role, role_prompt)
def run(self, user_request: str) -> str:
logger.info(
"Starting task: %s%s",
user_request[:60],
"..." if len(user_request) > 60 else "",
)
print(f"\n{'='*60}")
print(f" MyAgent - Task: {user_request[:60]}{'...' if len(user_request) > 60 else ''}")
print(f"{'='*60}\n")
if self.log_level is not None:
header_task = user_request[:80] + ("..." if len(user_request) > 80 else "")
logger.info("===== Agent started: %s =====", header_task)
if self._is_archivist_request(user_request):
self._handle_archivist_request(user_request)
return "Session state saved."
self.context.add_user_message(user_request)
initial_role = self._resolve_role_from_keywords(user_request)
self._apply_role(initial_role)
final_response = ""
last_tool_name: Optional[str] = None
while self.turn_count < self.max_turns:
self.turn_count += 1
logger.info("--- Turn %d ---", self.turn_count)
print(f"\n--- Turn {self.turn_count} ---")
if self.log_level is not None:
logger.info("--- Turn %d ---", self.turn_count)
summary = self.context.get_compaction_summary()
if summary:
logger.info("Context compacted: %s", summary[:100])
print("[Context compacted]")
if last_tool_name and self.turn_count > 1:
tool_role = self._resolve_role_from_tool(last_tool_name)
self._apply_role(tool_role)
last_tool_name = None
messages = self.context.build_messages()
tools = self.context.get_tool_definitions()
logger.debug("Sending %d messages to LLM (role=%s)", len(messages), self.context.get_role() or "none")
try:
response = self._chat(messages, tools=tools)
except Exception as e:
logger.error("LLM error at turn %d: %s", self.turn_count, e, exc_info=True)
return f"LLM Error: {e}"
if response.has_tool_calls():
logger.info("Received %d tool call(s)", len(response.tool_calls))
history_tool_calls = []
for tc in response.tool_calls:
history_tool_calls.append({
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments),
},
})
self.context.add_assistant_message(
content=response.content,
tool_calls=history_tool_calls,
)
for tc in response.tool_calls:
last_tool_name = tc.name
arg_str = json.dumps(tc.arguments, ensure_ascii=False)
logger.info("Executing tool: %s(%s)", tc.name, arg_str)
print(f"\n[Tool] {tc.name}({arg_str})")
if self.log_level is not None:
logger.info("[Tool] %s(%s)", tc.name, arg_str)
if tc.name not in self.tools:
result = f"ERROR: Unknown tool '{tc.name}'"
logger.warning("Unknown tool requested: %s", tc.name)
print(f"[Result] {result}")
self.context.add_tool_result(tc.id, tc.name, result)
continue
permitted = self.permission.check(tc.name, tc.arguments)
logger.debug("Permission check for %s: %s", tc.name, permitted)
if not permitted:
result = f"PERMISSION_DENIED: User declined to execute {tc.name}"
logger.warning("Permission denied for %s", tc.name)
print(f"[Result] {result}")
self.context.add_tool_result(tc.id, tc.name, result)
continue
result = self.tools.execute(tc.name, tc.arguments)
result_preview = result[:200] + ("..." if len(result) > 200 else "")
logger.info("Tool result: %s", result_preview)
print(f"[Result] {result[:500]}{'...' if len(result) > 500 else ''}")
if self.log_level is not None:
logger.info("[Result] %s", result_preview)
self.context.add_tool_result(tc.id, tc.name, result)
else:
self.context.add_assistant_message(content=response.content)
final_response = response.content or "(no response)"
logger.info("Agent finished at turn %d", self.turn_count)
if self.log_level is not None:
logger.info("===== Agent finished =====")
print(f"\n{'='*60}")
print(" Agent finished")
print(f"{'='*60}")
break
else:
final_response = f"Reached max turns ({self.max_turns}). Last response: {response.content or '(none)'}"
logger.warning("Reached max_turns limit (%s)", self.max_turns)
counters = self._get_counters()
summary = counters.summary()
logger.info("Task complete — %s", summary)
print(f"\n[{summary}]\n")
return final_response
def reset(self) -> None:
self.context.clear()
self.turn_count = 0
logger.info("Conversation reset")
def _is_archivist_request(self, text: str) -> bool:
text_lower = text.lower().strip()
return text_lower in ("archive", "save", "persist")
def _handle_archivist_request(self, user_request: str) -> None:
state_path = self.context.save_state()
summary = self.context.get_compaction_summary()
saved_msg = f"State saved"
if state_path:
saved_msg += f" to {state_path}"
if summary:
saved_msg += f" (compression summary generated)"
logger.info("Archivist: %s", saved_msg)
print(saved_msg)
def get_conversation_summary(self) -> str:
messages = self.context.get_messages()
user_msgs = sum(1 for m in messages if m.get("role") == "user")
assistant_msgs = sum(1 for m in messages if m.get("role") == "assistant")
tool_results = sum(1 for m in messages if m.get("role") == "tool")
current_role = self.context.get_role()
compaction = self.context.get_compaction_summary()
summary = (
f"Conversation: {len(messages)} messages, "
f"{user_msgs} user, {assistant_msgs} assistant, {tool_results} tool results"
f"{f', role={current_role}' if current_role else ''}"
f"{f' [compacted]' if compaction else ''}"
)
logger.debug("Summary: %s", summary)
return summary
# Module: myagent.main
import argparse
import getpass
import json
import os
import sys
from pathlib import Path
logger = get_logger(__name__)
def _log_print(level: Optional[str], msg: str, *args) -> None:
if level is not None:
logger.info(msg, *args)
def load_config(config_path: str) -> dict:
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
def resolve_api_key(llm_config: dict) -> str:
api_key = llm_config.get("api_key", "")
if api_key == "DEBUG":
logger.info("DEBUG mode: skipping API key resolution")
return api_key
if isinstance(api_key, str) and api_key.startswith("${") and api_key.endswith("}"):
env_var = api_key[2:-1]
api_key = os.environ.get(env_var, "")
llm_config["api_key"] = api_key
logger.debug(
"Resolved API key from env var %s: %s",
env_var,
"set" if api_key else "empty",
)
if not api_key:
provider = llm_config.get("provider", "openai")
print(f"API key not configured. Enter your API key for {provider}.")
print("(入力は画面に表示されません — そのまま貼り付けてEnterを押してください)")
prompt_msg = "API key: "
try:
api_key = getpass.getpass(prompt_msg)
except (EOFError, OSError):
api_key = input(prompt_msg).strip()
llm_config["api_key"] = api_key
logger.info("API key entered via interactive prompt.")
return api_key
def setup_tools() -> ToolRegistry:
registry = ToolRegistry()
registry.register_all(
ReadFileTool(),
WriteFileTool(),
EditFileTool(),
ListFilesTool(),
SearchFilesTool(),
RunCommandTool(),
)
return registry
CONFIG_SKELETON = {
"llm": {
"provider": "deepseek",
"api_key": "${API_KEY}",
"base_url": "https://api.openai.com/v1",
"model": "gpt-4o-mini",
"timeout": 120,
},
"permissions": {
"read_file": "auto",
"list_files": "auto",
"search_files": "auto",
"write_file": "ask",
"edit_file": "ask",
"run_command": "ask",
},
}
def init_config(config_path: str, force: bool = False) -> int:
path = Path(config_path)
if path.exists() and not force:
print(f"Config file already exists: {path}")
answer = input("Overwrite? [y/N]: ").strip().lower()
if answer not in ("y", "yes"):
print("Aborted.")
return 1
with open(path, "w", encoding="utf-8") as f:
json.dump(CONFIG_SKELETON, f, indent=2, ensure_ascii=False)
f.write("\n")
print(f"Created: {path}")
print("\nNext steps:")
print(" 1. Set your API key via environment variable (API_KEY)")
print(" 2. Or edit the file directly to set 'api_key'")
print(f" 3. Run: python -m myagent.main -i")
return 0
def main():
parser = argparse.ArgumentParser(
description="MyAgent - Autonomous Coding Agent",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"request",
nargs="?",
help="Task to perform (e.g., 'create hello.py')",
)
parser.add_argument(
"--config",
default="config.json",
help="Path to config file (default: config.json)",
)
parser.add_argument(
"--prompts",
default="prompts",
help="Path to prompts directory (default: prompts)",
)
parser.add_argument(
"--max-turns",
type=int,
default=50,
help="Maximum agent turns (default: 50)",
)
parser.add_argument(
"--interactive",
"-i",
action="store_true",
help="Run in interactive mode",
)
parser.add_argument(
"--log-dir",
default="logs",
help="Directory for log files (default: logs)",
)
parser.add_argument(
"--init",
action="store_true",
help="Create a skeleton config.json and exit",
)
parser.add_argument(
"--force",
"-f",
action="store_true",
help="Overwrite existing config when using --init",
)
args = parser.parse_args()
if args.init:
return init_config(args.config, force=args.force)
config_path = os.path.abspath(args.config)
if not os.path.exists(config_path):
print(f"Error: Config file not found: {config_path}")
print("Create a config.json with your API settings.")
sys.exit(1)
config = load_config(config_path)
log_level = config.get("log_level")
log_file = setup_logging(log_dir=args.log_dir, log_level=log_level)
llm_config = config.get("llm", {})
resolve_api_key(llm_config)
logger.debug("Initializing LLM provider")
role_configs = config.get("models")
if role_configs:
llm = ModelRouter(llm_config, role_configs)
logger.info("ModelRouter enabled with %d role configs", len(role_configs))
else:
llm = create_provider(llm_config)
logger.debug("Registering tools")
tools = setup_tools()
perm_overrides = config.get("permissions", {})
permission = PermissionManager(perm_overrides)
logger.debug("Loading prompts from: %s", args.prompts)
prompt_loader = PromptLoader(args.prompts)
logger.debug("Creating agent with max_turns=%d", args.max_turns)
agent = Agent(
llm=llm,
tool_registry=tools,
permission=permission,
prompt_loader=prompt_loader,
max_turns=args.max_turns,
log_level=log_level,
)
_log_print(log_level, "MyAgent v%s", VERSION)
print(f"MyAgent v{VERSION}")
print(f" Config: {config_path}")
print(f" Prompts: {args.prompts}/")
print()
_log_print(log_level, "System prompt loaded from: %s/", args.prompts)
print(f"System prompt loaded from: {args.prompts}/")
_log_print(log_level, "Log file: %s", log_file)
print(f"Log file: {log_file}")
_log_print(log_level, "Available tools: %s", ", ".join(tools.list_tools()))
print(f"Available tools: {', '.join(tools.list_tools())}")
summary_text = permission.get_summary()
_log_print(log_level, "%s", summary_text)
print(summary_text)
print()
if args.interactive or not args.request:
_log_print(log_level, "MyAgent Interactive Mode")
_log_print(log_level, "Type 'exit' or 'quit' to exit, 'reset' to clear history.")
print("MyAgent Interactive Mode")
print("Type 'exit' or 'quit' to exit, 'reset' to clear history.\n")
while True:
try:
user_input = input(">>> ").strip()
except (EOFError, KeyboardInterrupt):
_log_print(log_level, "Exiting...")
print("\nExiting...")
break
if not user_input:
continue
if user_input.lower() in ("exit", "quit"):
_log_print(log_level, "Exiting...")
print("Exiting...")
break
if user_input.lower() == "reset":
agent.reset()
_log_print(log_level, "Conversation reset.")
print("Conversation reset.")
continue
if user_input.lower() == "summary":
_log_print(log_level, "Summary: %s", agent.get_conversation_summary())
print(agent.get_conversation_summary())
continue
if user_input.lower() == "stats":
counters = agent.llm.get_counters()
_log_print(log_level, "Requests: %d, Prompt: %d, Completion: %d, Total: %d",
counters.total_requests, counters.total_prompt_tokens,
counters.total_completion_tokens, counters.total_tokens)
print(f"API Statistics:")
print(f" Requests: {counters.total_requests}")
print(f" Prompt tokens: {counters.total_prompt_tokens}")
print(f" Completion: {counters.total_completion_tokens}")
print(f" Total tokens: {counters.total_tokens}")
continue
if user_input.lower() in ("archive", "save"):
state_path = agent.context.save_state()
if state_path:
_log_print(log_level, "Session archived: %s", state_path)
print(f"Session archived: {state_path}")
else:
_log_print(log_level, "Session not yet compacted; nothing to archive.")
print("Session not yet compacted; nothing to archive.")
continue
result = agent.run(user_input)
print(f"\n{result}\n")
else:
result = agent.run(args.request)
print(result)
if __name__ == "__main__":
main()
