src/agent_server/chat/summary.py - master - man4j/agent-server

src/agent_server/chat/summary.py
253 строки8 KB
Vladimir
fixes
28 апр 2026, 19:57
28 апр 2026, 19:57f04f1da
import logging

from openai import OpenAI

from agent_server.chat.policy import ChatBehaviorPolicy
from agent_server.chat.types import ChatMessage
from agent_server.llm.agent import summarize_messages
from agent_server.message_window import compute_trimmed_history, trim_history_by_budget
from agent_server.config.agents import AgentDefinition
from agent_server.config.profiles import LlmProfile


CHAT_SUMMARY_MESSAGE_NAME = "chat_summary"
SUMMARY_REFRESH_MAX_ITERATIONS = 5
logger = logging.getLogger(__name__)
BASE_CHAT_SUMMARY_PROMPT = (
    "Ты сжимаешь историю текущего чата перед подрезанием контекста. "
    "Верни только краткую, плотную Markdown-сводку на русском языке. "
    "Сохраняй только информацию, которая поможет корректно продолжить диалог: "
    "ключевые факты, принятые решения, устойчивые предпочтения, важные изменения состояния, "
    "ограничения, незавершенные вопросы и значимые договоренности. "
    "Убирай повторы, воду, мелкие реплики и одноразовые детали. "
    "Не пиши вводных фраз и комментариев о том, что это summary."
)

def load_agent_summary_prompt(agent: AgentDefinition) -> str:
    path = getattr(agent, "summary_prompt_file", None)
    if not path:
        return ""

    try:
        with open(path, encoding="utf-8") as f:
            return f.read().strip()
    except Exception:
        return ""


def build_chat_summary_system_prompt(agent: AgentDefinition) -> str:
    agent_prompt = load_agent_summary_prompt(agent)
    if agent_prompt:
        return BASE_CHAT_SUMMARY_PROMPT + "\n\n" + agent_prompt
    return BASE_CHAT_SUMMARY_PROMPT


def find_chat_summary_index(messages: list[ChatMessage]) -> int | None:
    for i, message in enumerate(messages):
        if message["role"] == "system" and message.get("name") == CHAT_SUMMARY_MESSAGE_NAME:
            return i
    return None


def extract_chat_summary(messages: list[ChatMessage]) -> str:
    idx = find_chat_summary_index(messages)
    if idx is None:
        return ""
    return str(messages[idx].get("content") or "").strip()


def build_chat_summary_message(summary: str) -> ChatMessage:
    return {
        "role": "system",
        "name": CHAT_SUMMARY_MESSAGE_NAME,
        "content": summary.strip(),
    }


def upsert_chat_summary(messages: list[ChatMessage], summary: str) -> None:
    summary = (summary or "").strip()
    idx = find_chat_summary_index(messages)

    if not summary:
        if idx is not None:
            messages.pop(idx)
        return

    new_message = build_chat_summary_message(summary)

    if idx is not None:
        messages[idx] = new_message
        return

    insert_at = 0
    while insert_at < len(messages) and messages[insert_at]["role"] == "system":
        insert_at += 1

    if insert_at > 0:
        insert_at = 1

    messages.insert(insert_at, new_message)


def is_meaningful_for_summary(message: ChatMessage) -> bool:
    role = message.get("role")
    if role == "system":
        return False

    if role == "tool":
        return bool((message.get("content") or "").strip())

    content = (message.get("content") or "").strip()
    reasoning = (message.get("reasoning_content") or "").strip()
    return bool(content or reasoning or message.get("tool_calls"))


def render_messages_for_summary(messages: list[ChatMessage]) -> str:
    parts: list[str] = []

    for message in messages:
        role = message.get("role") or "unknown"
        name = message.get("name")
        header = role
        if name:
            header += f" ({name})"

        content = (message.get("content") or "").strip()
        if message.get("tool_calls"):
            content += ("\n\n" if content else "") + f"tool_calls: {message.get('tool_calls')}"
        if role == "tool" and message.get("tool_call_id"):
            content += ("\n\n" if content else "") + f"tool_call_id: {message.get('tool_call_id')}"

        if not content:
            continue

        parts.append(f"[{header}]\n{content}")

    return "\n\n".join(parts).strip()


async def _extend_chat_summary(
    *,
    client: OpenAI,
    profile: LlmProfile,
    agent: AgentDefinition,
    existing_summary: str,
    dropped_messages: list[ChatMessage],
) -> str:
    dropped_block = render_messages_for_summary(dropped_messages)
    if not dropped_block:
        return existing_summary

    summary_request_messages = [
        {
            "role": "system",
            "content": build_chat_summary_system_prompt(agent),
        },
        {
            "role": "user",
            "content": (
                "Текущая сводка:\n"
                f"{existing_summary or '(пусто)'}\n\n"
                "Фрагмент истории, который будет удален:\n"
                f"{dropped_block}\n\n"
                "Обнови сводку с учетом этого фрагмента."
            ),
        },
    ]

    new_summary = await summarize_messages(
        client=client,
        profile=profile,
        messages=summary_request_messages,
    )
    return new_summary or existing_summary


async def refresh_chat_summary_if_needed(
    messages: list[ChatMessage],
    *,
    client: OpenAI,
    profile: LlmProfile,
    agent: AgentDefinition,
    policy: ChatBehaviorPolicy,
    char_per_token: float,
) -> None:
    if not policy.chat_summarization_enabled:
        idx = find_chat_summary_index(messages)
        if idx is not None:
            messages.pop(idx)
        return

    working_messages = list(messages)
    working_summary = extract_chat_summary(messages)

    for _ in range(SUMMARY_REFRESH_MAX_ITERATIONS):
        trimmed_messages, dropped_messages = compute_trimmed_history(
            working_messages,
            char_per_token,
            policy.context_size,
            response_headroom_tokens=policy.response_headroom_tokens,
            recent_history_ratio=policy.recent_history_ratio,
        )

        meaningful_dropped = [
            message for message in dropped_messages if is_meaningful_for_summary(message)
        ]
        if not meaningful_dropped:
            messages[:] = trimmed_messages
            return

        working_summary = await _extend_chat_summary(
            client=client,
            profile=profile,
            agent=agent,
            existing_summary=working_summary,
            dropped_messages=meaningful_dropped,
        )
        working_messages = list(trimmed_messages)
        upsert_chat_summary(working_messages, working_summary)

    final_trimmed_messages, final_dropped_messages = compute_trimmed_history(
        working_messages,
        char_per_token,
        policy.context_size,
        response_headroom_tokens=policy.response_headroom_tokens,
        recent_history_ratio=policy.recent_history_ratio,
    )
    meaningful_final_dropped = [
        message for message in final_dropped_messages if is_meaningful_for_summary(message)
    ]
    if meaningful_final_dropped:
        logger.warning(
            "Chat summary refresh did not converge after %s iterations; "
            "dropping %s additional meaningful messages",
            SUMMARY_REFRESH_MAX_ITERATIONS,
            len(meaningful_final_dropped),
        )
    messages[:] = final_trimmed_messages


async def prepare_messages_for_next_llm_round(
    messages: list[ChatMessage],
    *,
    client: OpenAI,
    profile: LlmProfile,
    agent: AgentDefinition,
    policy: ChatBehaviorPolicy,
    char_per_token: float,
) -> None:
    await refresh_chat_summary_if_needed(
        messages,
        client=client,
        profile=profile,
        agent=agent,
        policy=policy,
        char_per_token=char_per_token,
    )
    trim_history_by_budget(
        messages,
        char_per_token,
        policy.context_size,
        response_headroom_tokens=policy.response_headroom_tokens,
        recent_history_ratio=policy.recent_history_ratio,
    )