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

src/agent_server/chat/thread_metadata.py
70 строк2 KB
Vladimir
initial
18 апр 2026, 19:36
18 апр 2026, 19:36a95f341
from chainlit.types import ThreadDict

from agent_server.chat.types import ChatMessage, ThreadMetadata


def get_thread_id(thread: ThreadDict | None) -> str | None:
    if not thread:
        return None
    thread_id = thread.get("id")
    return str(thread_id) if thread_id is not None else None


def get_thread_metadata(thread: ThreadDict | None) -> ThreadMetadata:
    if not thread:
        return {}
    metadata = thread.get("metadata") or {}
    return metadata if isinstance(metadata, dict) else {}


def get_thread_metadata_system_prompt(thread: ThreadDict | None) -> str | None:
    value = get_thread_metadata(thread).get("system_prompt")
    return value if isinstance(value, str) and value.strip() else None


def get_thread_metadata_profile_id(thread: ThreadDict | None) -> str | None:
    value = get_thread_metadata(thread).get("llm_profile_id")
    return value if isinstance(value, str) and value.strip() else None


def get_thread_metadata_chat_summary(thread: ThreadDict | None) -> str | None:
    value = get_thread_metadata(thread).get("chat_summary")
    return value if isinstance(value, str) and value.strip() else None


def get_thread_metadata_messages(thread: ThreadDict | None) -> list[ChatMessage] | None:
    value = get_thread_metadata(thread).get("messages_v2")

    if not isinstance(value, list):
        return None

    result: list[ChatMessage] = []

    for item in value:
        if not isinstance(item, dict):
            continue

        role = item.get("role")
        if role not in {"system", "user", "assistant", "tool"}:
            continue

        restored: ChatMessage = {
            "role": role,
            "content": item.get("content") or "",
        }

        if "reasoning_content" in item:
            restored["reasoning_content"] = item["reasoning_content"]

        if "tool_call_id" in item:
            restored["tool_call_id"] = item["tool_call_id"]

        if "tool_calls" in item:
            restored["tool_calls"] = item["tool_calls"]

        if "name" in item:
            restored["name"] = item["name"]

        result.append(restored)

    return result