src/agent.py - master - sasha.martian/DevAgent

src/agent.py
116 строк4 KB
Galushkin Alexander
Step
06 авг 2026, 13:02
06 авг 2026, 13:0290d584c
"""ReAct agent factory and turn helper."""

from __future__ import annotations

from collections.abc import Callable
from typing import Any

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph.state import CompiledStateGraph
from langgraph.prebuilt import create_react_agent

from src.config import Settings, create_gigachat_client
from src.logger import setup_logging
from src.tools import get_mvp_tools

DEFAULT_THREAD_ID = "cli-session"
EventCallback = Callable[[str], None]

SYSTEM_PROMPT = (
    "Ты DevAgent — помощник разработчика. "
    "Используй инструменты для чтения файлов, просмотра каталогов и выполнения команд. "
    "Отвечай по делу на русском языке."
)


def create_agent(
    settings: Settings,
    *,
    confirm: Callable[[str], bool],
) -> CompiledStateGraph:
    """Build a ReAct agent with GigaChat, MVP tools, and MemorySaver."""
    model = create_gigachat_client(settings)
    tools = get_mvp_tools(confirm)
    checkpointer = MemorySaver()
    return create_react_agent(
        model,
        tools,
        prompt=SYSTEM_PROMPT,
        checkpointer=checkpointer,
    )


def _format_tool_call(name: str, args: Any) -> str:
    if isinstance(args, dict):
        parts = ", ".join(f"{k}={v!r}" for k, v in args.items())
        return f"→ {name}({parts})"
    return f"→ {name}({args!r})"


def _emit_message_events(message: BaseMessage, on_event: EventCallback | None) -> str | None:
    """Print progress for a message; return AI text content if present."""
    if isinstance(message, AIMessage):
        tool_calls = getattr(message, "tool_calls", None) or []
        for call in tool_calls:
            name = call.get("name", "tool") if isinstance(call, dict) else getattr(call, "name", "tool")
            args = call.get("args", {}) if isinstance(call, dict) else getattr(call, "args", {})
            if on_event:
                on_event(_format_tool_call(name, args))
        content = message.content
        if isinstance(content, list):
            text_parts = [
                block.get("text", "") if isinstance(block, dict) else str(block)
                for block in content
            ]
            text = "".join(text_parts).strip()
        else:
            text = str(content).strip() if content else ""
        return text or None

    if isinstance(message, ToolMessage) and on_event:
        name = getattr(message, "name", None) or "tool"
        preview = str(message.content)
        if len(preview) > 200:
            preview = preview[:200] + "…"
        on_event(f"← {name}: {preview}")
    return None


def run_agent_turn(
    agent: CompiledStateGraph,
    user_message: str,
    *,
    thread_id: str = DEFAULT_THREAD_ID,
    on_event: EventCallback | None = None,
) -> str:
    """Run one user turn; stream progress via on_event; return final assistant text."""
    logger = setup_logging()
    config = {"configurable": {"thread_id": thread_id}}
    final_text = ""

    try:
        for update in agent.stream(
            {"messages": [HumanMessage(content=user_message)]},
            config,
            stream_mode="updates",
        ):
            if not isinstance(update, dict):
                continue
            for _node, payload in update.items():
                if not isinstance(payload, dict):
                    continue
                messages = payload.get("messages")
                if messages is None:
                    continue
                if isinstance(messages, BaseMessage):
                    messages = [messages]
                for message in messages:
                    text = _emit_message_events(message, on_event)
                    if text:
                        final_text = text
    except Exception as exc:  # noqa: BLE001 — surface API/tool failures to CLI
        logger.exception("Ошибка хода агента: %s", exc)
        raise

    return final_text or "(пустой ответ модели)"