Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 706021cd2f | |||
| c2fd21b3f9 |
+17
-4
@@ -79,6 +79,7 @@ class AgentRequest:
|
||||
class FilterRequest:
|
||||
user_text: str
|
||||
candidate_text: str
|
||||
password: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -113,11 +114,14 @@ class PromptAgent:
|
||||
self.model = model
|
||||
self.prompt_suffix = prompt_suffix
|
||||
|
||||
def reply(self, request: AgentRequest) -> str:
|
||||
system_prompt = BASE_SYSTEM_PROMPT.format(password=request.password)
|
||||
def _build_system_prompt(self, password: str) -> str:
|
||||
system_prompt = BASE_SYSTEM_PROMPT.format(password=password)
|
||||
if self.prompt_suffix:
|
||||
system_prompt += self.prompt_suffix
|
||||
return system_prompt
|
||||
|
||||
def reply(self, request: AgentRequest) -> str:
|
||||
system_prompt = self._build_system_prompt(request.password)
|
||||
response = self.model.invoke(
|
||||
[
|
||||
SystemMessage(content=system_prompt),
|
||||
@@ -139,6 +143,7 @@ class FilterCheck:
|
||||
HumanMessage(
|
||||
content=(
|
||||
f"Тип проверки: {self.check_kind}\n"
|
||||
f"Секрет для проверки:\n{request.password}\n\n"
|
||||
f"Сообщение пользователя:\n{request.user_text}\n\n"
|
||||
f"Проверяемый текст:\n{request.candidate_text}"
|
||||
)
|
||||
@@ -163,7 +168,11 @@ class LevelExecutor:
|
||||
request_filter: FilterDecision | None = None
|
||||
for check in self.pipeline.input_checks:
|
||||
decision = check.check(
|
||||
FilterRequest(user_text=user_text, candidate_text=user_text)
|
||||
FilterRequest(
|
||||
user_text=user_text,
|
||||
candidate_text=user_text,
|
||||
password=password,
|
||||
)
|
||||
)
|
||||
if decision.triggered:
|
||||
request_filter = decision
|
||||
@@ -180,7 +189,11 @@ class LevelExecutor:
|
||||
response_filter: FilterDecision | None = None
|
||||
for check in self.pipeline.output_checks:
|
||||
decision = check.check(
|
||||
FilterRequest(user_text=user_text, candidate_text=reply)
|
||||
FilterRequest(
|
||||
user_text=user_text,
|
||||
candidate_text=reply,
|
||||
password=password,
|
||||
)
|
||||
)
|
||||
if decision.triggered:
|
||||
response_filter = decision
|
||||
|
||||
+6
-1
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import logging.config
|
||||
import tomllib
|
||||
@@ -79,7 +80,11 @@ def setup_logging(settings: LoggingSettings) -> None:
|
||||
},
|
||||
"json": {
|
||||
"()": structlog.stdlib.ProcessorFormatter,
|
||||
"processor": structlog.processors.JSONRenderer(),
|
||||
"processor": structlog.processors.JSONRenderer(
|
||||
serializer=lambda obj, **kwargs: json.dumps(
|
||||
obj, ensure_ascii=False, **kwargs
|
||||
)
|
||||
),
|
||||
"foreign_pre_chain": shared_processors,
|
||||
},
|
||||
},
|
||||
|
||||
@@ -73,6 +73,7 @@ def test_level1_uses_simple_agent_without_filters() -> None:
|
||||
assert result.filter_request is None
|
||||
assert result.filter_response is None
|
||||
assert len(chat_model.invocations) == 1
|
||||
assert "Твой пароль apple" in chat_model.invocations[0][0].content
|
||||
assert filter_model.invocations == []
|
||||
|
||||
|
||||
@@ -100,6 +101,7 @@ def test_level2_blocks_response_when_output_check_triggers() -> None:
|
||||
assert result.filter_response.triggered is True
|
||||
assert len(chat_model.invocations) == 1
|
||||
assert len(filter_model.invocations) == 1
|
||||
assert "Секрет для проверки:\napple" in filter_model.invocations[0][1].content
|
||||
|
||||
|
||||
def test_level3_blocks_on_input_before_agent_call() -> None:
|
||||
@@ -155,7 +157,9 @@ def test_level4_uses_astronomy_agent_with_both_checks() -> None:
|
||||
assert ASTRONOMY_PROMPT_SUFFIX in system_prompt
|
||||
assert len(filter_model.invocations) == 2
|
||||
assert "Тип проверки: user_request" in filter_model.invocations[0][1].content
|
||||
assert "Секрет для проверки:\napple" in filter_model.invocations[0][1].content
|
||||
assert "Тип проверки: assistant_response" in filter_model.invocations[1][1].content
|
||||
assert "Секрет для проверки:\napple" in filter_model.invocations[1][1].content
|
||||
|
||||
|
||||
def test_hard_mode_rotates_session_and_uses_new_session_for_current_request() -> None:
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from config import AppSettings, as_public_dict
|
||||
import json
|
||||
import logging
|
||||
|
||||
from config import AppSettings, LoggingSettings, as_public_dict, setup_logging
|
||||
|
||||
|
||||
def test_app_settings_load_from_toml_and_masks_api_key(tmp_path) -> None:
|
||||
@@ -30,3 +33,14 @@ password_words = [
|
||||
assert settings.llm.model == "test-model"
|
||||
assert settings.game.hard_mode_rotation_interval == 3
|
||||
assert public_data["llm"]["api_key"] == "***"
|
||||
|
||||
|
||||
def test_file_logs_write_unicode_without_ascii_escaping(tmp_path) -> None:
|
||||
setup_logging(LoggingSettings(logs_dir=tmp_path, app_log_name="app.log"))
|
||||
|
||||
logging.getLogger().info("Привет")
|
||||
|
||||
log_text = (tmp_path / "app.log").read_text(encoding="utf-8")
|
||||
|
||||
assert "\\u041f" not in log_text
|
||||
assert json.loads(log_text)["event"] == "Привет"
|
||||
|
||||
Reference in New Issue
Block a user