ElebobaandClaude Opus 5.5
65acf3abf1
CI / ruff · mypy · pytest (pull_request) Successful in 39m13s
- prefilter: правила из config/agent_rules.yaml (длина, команды, shell,
секреты и системные файлы, prompt injection) — без LLM
- guard: отдельный вызов, JSON {verdict, reason}; невалидный ответ = attack,
недоступный LLM — «попробуйте позже», а не атака
- planner: только AgentLogQuery (контейнер из белого списка, ≤ 7 д, шаблон
≤ 200 без вложенных квантификаторов, лимит ≤ 500); логи в LLM не идут
- бот: «🤖 Спросить агента» → показывает построенный grep и строки;
атака — вежливый отказ и 🛡 администраторам; квота 30 в сутки
- журнал agent_requests + log_queries (mode=agent)
- при LLM_PROVIDER=mock агент выключен
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
"""Step 3 (SPEC L-13, L-16): the LLM turns the question into an `AgentLogQuery`. It never sees the logs."""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from app.agent.log_query import AgentLogQuery, json_schema
|
|
from app.llm.provider import ChatMessage, JsonSchema, LLMProvider
|
|
|
|
SCHEMA = JsonSchema(name="log_query", schema=json_schema())
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Plan:
|
|
query: AgentLogQuery | None # None — the reply didn't pass validation
|
|
error: str
|
|
tokens: int
|
|
|
|
|
|
def system_prompt(template: str, containers: list[str], default: str, now: datetime) -> str:
|
|
# str.replace, not format: the template contains JSON braces
|
|
return (
|
|
template.replace("{containers}", ", ".join(containers))
|
|
.replace("{default}", default)
|
|
.replace("{now}", now.strftime("%Y-%m-%d %H:%M"))
|
|
)
|
|
|
|
|
|
async def plan(
|
|
llm: LLMProvider, template: str, prompt: str, containers: list[str], default: str, now: datetime
|
|
) -> Plan:
|
|
messages = [ChatMessage("system", system_prompt(template, containers, default, now)), ChatMessage("user", prompt)]
|
|
result = await llm.complete(messages, SCHEMA)
|
|
if result.data is None:
|
|
return Plan(None, "not JSON", result.tokens)
|
|
try:
|
|
query = AgentLogQuery.model_validate(result.data, context={"containers": containers})
|
|
except ValidationError as e:
|
|
return Plan(
|
|
None,
|
|
"; ".join(f"{'.'.join(map(str, err['loc']))}: {err['msg']}" for err in e.errors())[:300],
|
|
result.tokens,
|
|
)
|
|
return Plan(query, "", result.tokens)
|