Hugging Face's logo

Tayrrr
/
KaliSupport

GGUF
conversational
Model card Files
xet
 Community
1
KaliSupport / tool_call.py
Tayrrr's picture
Tayrrr
Upload folder using huggingface_hub
7ea1cf8 verified
Raw History Blame Contribute Delete
14.1 kB
"""ReAct stili tool cagri ayriciligi + dispatcher.
Akis (genel)
------------
1. Model ureten bir metin doner. Genellikle su sablon:
Thought: hedefin portlarini taramaliyim.
Action: run_cmd
Action Input: nmap -sV -p- 10.0.0.5
ya da bitirirken:
Thought: yeterli bilgi var.
Final Answer: Acik portlar: 22/ssh, 80/http ...
2. `parse_react()` bu metni `ParsedResponse`'a cevirir.
3. agent.py birden fazla aksiyon varsa kullaniciya sorar (proje karari).
4. `ToolDispatcher.dispatch(call)` -> safety -> KaliSSH -> Observation string.
5. Observation modele "Observation: ..." prefix'i ile geri verilir.
Bu modul sadece parsing ve dispatch'i yapar; LLM cagrisi agent.py'da.
Modelfile uyumu
---------------
Modelfile'da `Observation:` ve `Tool Output:` stop token'lari var. Bu sayede
model `Action Input: ...` yazdiktan sonra durur ve biz observation'i
ekleyene kadar uretmez. parse_react bu sayede tek seferde bir aksiyona
odaklanir.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Callable, Optional
from safety import SafetyChecker, Verdict
from ssh_client import KaliSSH, SSHConnectionError
if TYPE_CHECKING:
from logger import SessionLogger
# -----------------------------------------------------------------------------
# Veri tipleri
# -----------------------------------------------------------------------------
@dataclass
class ToolCall:
"""Model tarafindan istenen tek bir tool cagrisi."""
tool: str
args: dict[str, str]
raw: str = "" # ham metin (debug)
def short(self, n: int = 120) -> str:
body = self.args.get("cmd") or json.dumps(self.args, ensure_ascii=False)
body = body.replace("\n", " ")
return f"{self.tool}({body if len(body) <= n else body[: n - 3] + '...'})"
@dataclass
class ParsedResponse:
"""Modelin tek bir uretiminden cikarilmis yapisal bilgi."""
thought: Optional[str]
actions: list[ToolCall] = field(default_factory=list)
final_answer: Optional[str] = None
raw: str = ""
@property
def has_action(self) -> bool:
return bool(self.actions)
@property
def is_final(self) -> bool:
return self.final_answer is not None and not self.actions
# -----------------------------------------------------------------------------
# Parser
# -----------------------------------------------------------------------------
# Bir bolumun bittigini gosteren etiketler. Lookahead'de kullanilir.
_SECTION_LOOKAHEAD = (
r"(?=\n\s*(?:Thought\s*:|Action\s*:|Action\s*Input\s*:|"
r"Observation\s*:|Tool\s*Output\s*:|Final\s*Answer\s*:)|\Z)"
)
_THOUGHT_RE = re.compile(
r"Thought\s*:\s*(?P<body>.*?)" + _SECTION_LOOKAHEAD,
re.DOTALL | re.IGNORECASE,
)
# Klasik ReAct format:
# Action: run_cmd
# Action Input: nmap -sV 1.2.3.4
_ACTION_BLOCK_RE = re.compile(
r"Action\s*:\s*(?P<tool>[A-Za-z_][\w\-]*)\s*\n"
r"\s*Action\s*Input\s*:\s*(?P<input>.*?)" + _SECTION_LOOKAHEAD,
re.DOTALL | re.IGNORECASE,
)
# Fonksiyon-cagri stili (model bu sekilde de yazabilir):
# Action: run_cmd(nmap -sV 1.2.3.4)
# Parantez ici icerigi tolere etmek icin nested 1 seviye paren'e izin veriyoruz.
# Onemli: bu regex \n veya '(' ile baslayan satirlari yakalamali, klasik
# formati da yanlis match etmesin diye lookahead ile parantez sart.
_ACTION_FUNCTION_RE = re.compile(
r"Action\s*:\s*(?P<tool>[A-Za-z_][\w\-]*)\s*"
r"\(\s*(?P<input>(?:[^()]|\([^()]*\))*?)\s*\)",
re.DOTALL | re.IGNORECASE,
)
_FINAL_RE = re.compile(
r"Final\s*Answer\s*:\s*(?P<body>.*?)\Z",
re.DOTALL | re.IGNORECASE,
)
def _strip_input(raw: str) -> str:
"""Action Input'u temizle: kod blogu / tirnak isaretlerini sok."""
text = raw.strip()
if not text:
return ""
# ```bash\n...\n``` veya ```\n...\n```
fenced = re.match(r"^```(?:[a-zA-Z0-9_+-]*)\s*\n(?P<body>.*?)\n```\s*$", text, re.DOTALL)
if fenced:
text = fenced.group("body").strip()
return text
# `tek-satir`
if text.startswith("`") and text.endswith("`") and text.count("`") == 2:
text = text[1:-1].strip()
return text
# "cift tirnak" veya 'tek tirnak' (sarmalayici)
if (text.startswith('"') and text.endswith('"')) or (
text.startswith("'") and text.endswith("'")
):
text = text[1:-1].strip()
return text
def parse_react(text: str) -> ParsedResponse:
"""ReAct stilinde uretilmis metni yapisal hale getir.
Notlar:
* Birden fazla Action blogu varsa hepsi listeye girer.
* Final Answer varsa actions yine de toplanir; agent.py karari verir.
* Buyuk/kucuk harf duyarsiz; etiketlerin once gelen bosluklari onemsiz.
"""
raw = text or ""
thought_match = _THOUGHT_RE.search(raw)
thought = thought_match.group("body").strip() if thought_match else None
# Iki farkli format'i da yakaliyoruz; pozisyona gore sirala ki model'in
# uretim sirasi korunsun. Ayni pozisyonda klasik bulunduysa fonksiyon
# cagrisi versiyonunu yoksay (cunku zaten klasik match ediyor demektir).
raw_matches: list[tuple[int, str, str, str]] = [] # (pos, tool, input, raw_text)
classic_spans: list[tuple[int, int]] = []
for m in _ACTION_BLOCK_RE.finditer(raw):
classic_spans.append((m.start(), m.end()))
raw_matches.append(
(m.start(), m.group("tool").strip(), _strip_input(m.group("input")), m.group(0))
)
def _overlaps_classic(start: int, end: int) -> bool:
return any(not (end <= s or start >= e) for s, e in classic_spans)
for m in _ACTION_FUNCTION_RE.finditer(raw):
if _overlaps_classic(m.start(), m.end()):
continue
raw_matches.append(
(m.start(), m.group("tool").strip(), _strip_input(m.group("input")), m.group(0))
)
raw_matches.sort(key=lambda t: t[0])
actions: list[ToolCall] = [
ToolCall(tool=tool, args={"cmd": cmd_input}, raw=raw_text)
for _, tool, cmd_input, raw_text in raw_matches
]
final_match = _FINAL_RE.search(raw)
final_answer = final_match.group("body").strip() if final_match else None
# Final Answer eger ayni cikti icinde Action'dan SONRA geliyorsa, ReAct
# mantiginda Action'i once isleriz. Bu yuzden eger actions varsa
# final_answer'i bu turda yoksayariz; agent.py loop'u bir sonraki turda
# alir. Aksi durumda final_answer dogru kalir.
if actions and final_answer is not None:
# Eger final_answer son Action'dan sonra geliyorsa, bos olmadigi
# surece bilgi olarak tutalim ama is_final dondermeyiz.
all_action_ends = [
m.end() for m in _ACTION_BLOCK_RE.finditer(raw)
] + [
m.end() for m in _ACTION_FUNCTION_RE.finditer(raw)
]
last_action_end = max(all_action_ends) if all_action_ends else 0
final_pos = raw.rfind("Final Answer")
if final_pos < last_action_end:
final_answer = None
return ParsedResponse(
thought=thought,
actions=actions,
final_answer=final_answer,
raw=raw,
)
# -----------------------------------------------------------------------------
# Dispatcher
# -----------------------------------------------------------------------------
ToolFn = Callable[[dict[str, str]], str]
class UnknownToolError(RuntimeError):
pass
class ToolDispatcher:
"""ToolCall -> Observation string. SafetyChecker araligi burada."""
def __init__(
self,
kali: KaliSSH,
safety: SafetyChecker,
*,
logger: Optional["SessionLogger"] = None,
):
self.kali = kali
self.safety = safety
self.logger = logger
self._tools: dict[str, ToolFn] = {
"run_cmd": self._tool_run_cmd,
}
@property
def available_tools(self) -> tuple[str, ...]:
return tuple(sorted(self._tools.keys()))
def register(self, name: str, fn: ToolFn) -> None:
self._tools[name] = fn
def dispatch(self, call: ToolCall) -> str:
if self.logger:
self.logger.tool_call(call.tool, call.args)
fn = self._tools.get(call.tool)
if fn is None:
msg = (
f"BLOCKED: bilinmeyen tool '{call.tool}'. "
f"Kullanilabilir: {', '.join(self.available_tools)}"
)
if self.logger:
self.logger.error("unknown_tool", tool=call.tool)
return msg
try:
return fn(call.args)
except Exception as exc: # noqa: BLE001
if self.logger:
self.logger.error("tool_exception", tool=call.tool, detail=str(exc))
return f"ERROR: tool '{call.tool}' istisna firlatti: {exc}"
# --- tools ----------------------------------------------------------
def _tool_run_cmd(self, args: dict[str, str]) -> str:
cmd = (args.get("cmd") or "").strip()
if not cmd:
return "BLOCKED: bos komut."
verdict: Verdict = self.safety.check(cmd)
if verdict.blocked:
# Modele net mesaj donelim ki yeniden dene-mesin / siklasmasin.
return verdict.to_user_message()
try:
result = self.kali.run(cmd)
except SSHConnectionError as exc:
return f"ERROR: SSH/komut hatasi: {exc}"
body = result.combined_output()
prefix = ""
if verdict.severity == "warn":
prefix = verdict.to_user_message() + "\n"
return f"{prefix}exit={result.exit_code} ({result.duration_s:.1f}s)\n{body}"
# -----------------------------------------------------------------------------
# Yardimci: observation'i model girdisine donustur
# -----------------------------------------------------------------------------
def format_observation(obs: str) -> str:
"""Modele Observation: olarak verilecek bicimi olustur.
`Observation:` Modelfile'da stop token oldugu icin model bunu yazmaya
calismaz; bizim eklememiz gerekir. Sonuna bos satir koymak modelin
devam etme niyetini desteklemek icin.
"""
return f"Observation: {obs.rstrip()}\n"
# -----------------------------------------------------------------------------
# Self-test
# -----------------------------------------------------------------------------
def _self_test() -> int:
samples = [
# 0: tek action
(
"tek_action",
"Thought: portlari taramaliyim.\n"
"Action: run_cmd\n"
"Action Input: nmap -sV 10.0.0.1\n",
{"actions": 1, "final": False, "tool": "run_cmd",
"cmd": "nmap -sV 10.0.0.1"},
),
# 1: kod bloklu input
(
"kod_bloklu",
"Action: run_cmd\n"
"Action Input: ```bash\nnmap -sV 10.0.0.1\n```\n",
{"actions": 1, "cmd": "nmap -sV 10.0.0.1"},
),
# 2: tirnakli input
(
"tirnakli",
'Action: run_cmd\nAction Input: "whoami"\n',
{"actions": 1, "cmd": "whoami"},
),
# 3: final answer
(
"final",
"Thought: bitti.\nFinal Answer: 22/ssh, 80/http acik.\n",
{"actions": 0, "final": True},
),
# 4: coklu action
(
"coklu_action",
"Action: run_cmd\nAction Input: whoami\n"
"Action: run_cmd\nAction Input: hostname\n",
{"actions": 2},
),
# 5: action sonrasi final (agent loop bir sonraki tura biraktirir)
(
"action_sonra_final",
"Action: run_cmd\nAction Input: ls\nFinal Answer: tamam\n",
{"actions": 1, "final": False}, # is_final False olmali
),
# 6: fonksiyon-cagri stili (model bu sekilde de yazabilir)
(
"fonksiyon_cagri",
"Thought: scan.\nAction: run_cmd(nmap -p- 1.2.3.4 | grep open)\n",
{"actions": 1, "tool": "run_cmd",
"cmd": "nmap -p- 1.2.3.4 | grep open"},
),
# 7: fonksiyon-cagri + final (action calismali, final_answer yoksayilir)
(
"fonksiyon_sonra_final",
"Action: run_cmd(whoami)\nFinal Answer: ben kullaniciyim\n",
{"actions": 1, "final": False, "cmd": "whoami"},
),
# 8: ic parantezli komut (process substitution)
(
"ic_paren",
"Action: run_cmd(echo $(date) > /tmp/x)\n",
{"actions": 1, "cmd": "echo $(date) > /tmp/x"},
),
]
failed = 0
for name, text, expected in samples:
p = parse_react(text)
ok = True
if "actions" in expected and len(p.actions) != expected["actions"]:
ok = False
if "tool" in expected and (not p.actions or p.actions[0].tool != expected["tool"]):
ok = False
if "cmd" in expected and (not p.actions or p.actions[0].args.get("cmd") != expected["cmd"]):
ok = False
if "final" in expected and bool(p.is_final) != bool(expected["final"]):
ok = False
marker = "OK " if ok else "FAIL"
print(f"[{marker}] {name:<20} actions={len(p.actions)} "
f"final={p.is_final} thought={'Y' if p.thought else 'N'}")
if not ok:
failed += 1
print(f" beklenen: {expected}")
print(f" bulunan : actions={[a.short() for a in p.actions]} "
f"final={p.final_answer!r}")
print()
print(f"{len(samples) - failed}/{len(samples)} parse testi gecti.")
return 0 if failed == 0 else 1
if __name__ == "__main__":
import sys
sys.exit(_self_test())