Привет, Хабр! Меня зовут Владимир и я стал немного более GPU-rich. А это значит, что пора сдуть пыль со старого проекта)

Некоторое время назад захотелось мне немного повайбкодить. Но злобные корпорации не захотели брать мои рубли, поэтому я решил

Имеющаяся в то время в наличии 5080 на 16 Гб не дала мне развернуть адекватные модели, а загоняться с постоянным высвобождением памяти и перезагрузкой моделей мне было не охота, поэтому проект был заброшен. Но недавно я стал счастливым обладателем 5090 на 32 Гб, а это значит, что настало время новой попытки!

Статьи будет две - в первой мы создадим необходимую инфраструктуру, напишем простого агента, а также добавим нашему агенту MCP-инструменты. Во второй усложним логику агента и контейнеризируем его.

Первым делом установим зависимости

uv init
uv add langchain==1.3.2 langchain-mcp-adapters==0.2.2 langchain-openai==1.2.2 langfuse==4.7.1 langgraph==1.2.2 loguru==0.7.3 pydantic-settings==2.14.1

Сразу накидаем примерную структуру проекта - так как это агент, да ещё и с инструментами, создадим соответствующие папки. Плюс папку для настроек:

├── agent/
│   └── __init__.py
├── core/
│   ├── __init__.py
│   ├── logger.py
│   └── settings.py
├── tools/
│   └── __init__.py
├── .env
└── main.py

В предыдущей статье про трассировку агентов на LangGraph я постарался достаточно подробно описать свой подход к настройкам проекта (переменным окружения), а также базовые вещи по разработке LangGraph агента и использованию LangFuse для трассировки и хранения промптов, поэтому на этих частях кода подробно останавливаться не буду.

Для настройки проекта нам необходимы:

• настройки запуска агента (хост, порт и путь к рабочей директории, в которой агент будет помогать)

• настройки для LLM (имя, url, API-ключ)

• настройки для логирования

Реализуем core\settings.py для парсинга настроек. Сначала напишем модель для настроек самого сервиса:

class ServiceConfig(BaseModel):
    host: str
    port: int
    workspace: str
    loglevel: str
    log_to_file: bool

Следующий класс - модель настроек LLM:

class LLMConfig(BaseModel):
    host: str
    port: Annotated[int | None, Field(default=None, ge=1, le=65535)]
    model_name: str
    api_key: SecretStr
    temperature: Annotated[float, Field(default=0.0, ge=0.0, le=1.1)]

    @property
    def url(self) -> str:
        if self.port is not None:
            return f'http://{self.host}:{self.port}/v1'
        return f'https://{self.host}/v1'

    @cached_property
    def llm(self) -> ChatOpenAI:
        return ChatOpenAI(
            base_url=self.url,
            api_key=self.api_key,
            model=self.model_name,
            temperature=self.temperature,
        )

Здесь всё как в прошлой статье - port=None для определения локальное развёртывание модели или облачное, что позволит сформировать правильный URL для передачи в конструктор класса ChatOpenAI.

Так как агент у нас кодер, то будем использовать две разных модели - одну для общения, вторую - для кодинга. Добавим ещё один “фиктивный” класс для инкапсуляции настроек LLM и оформим общий класс настроек:

class LLMsTypes(BaseModel):
    chat: LLMConfig
    coder: LLMConfig

class Settings(BaseSettings):
    service: ServiceConfig = Field(validation_alias='AGENT')
    llm: LLMsTypes
    langfuse: LangfuseConfig

    model_config = SettingsConfigDict(
        env_file=Path(__file__).resolve().parent.parent / '.env',
        env_file_encoding='utf-8',
        extra='ignore',
        case_sensitive=False,
        env_nested_delimiter='__',
    )

    def model_post_init(self, __context) -> None:
        _ = self.langfuse.client
        _ = self.llm.chat.llm
        _ = self.llm.coder.llm

@lru_cache
def get_settings() -> Settings:
    try:
        return Settings()
    except Exception as e:
        print(f'Ошибка создания Settings: {e}')
        raise

settings = Settings()
configure_logging(level=settings.service.loglevel, create_file=settings.service.log_to_file)

Настройки логирования опишем в файле core\logger.py:

def configure_logging(level: str = 'INFO', create_file: bool = False):
    logger.remove()

    logger.add(
        sys.stdout,
        format=('<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{module}:{line}</cyan> - {message}'),
        level=level,
        colorize=True)
    if create_file:
        logger.add(
            'app.log',
            format='{time} | {level} | {module}:{line} - {message}',
            level=level, rotation='10 MB', retention='7 days', encoding='utf-8')

В настройках мы сначала очищаем стандартные настройки и создаем свои стильные, модные, молодёжные с выводом имени файла модуля и номера строки, из которой этот лог пишется. Параметр colorize=True раскрасит наш лог.

При необходимости, лог можно дополнительно сохранить в файл. Формат вывода не содержит тегов цвета, т.к. в текстовом файле они скорее всего превратятся в мусор. Дополнительно ограничим размер файла 10 мегабайтами (loguru закроет старый и создаст новый файл при достижении этого объема) и установил время жизни логов в 7 дней (loguru будет автоматически удалять файлы логов). Параметр encoding='utf-8' нужен, чтобы великий и могучий не превратился в кракозябры.

Подготовительные работы завершены, переходим к созданию агента.

Начнем реализовывать наш агент. Нам понадобится:

• LLM - 2 шт

• Управляемое состояние (он же State) - 1 шт

• Узел-агент - 1 шт

• Инструмент с кодером - 1 шт

• Какие-либо полезные инструменты - ХЗ шт

• Узел маршрутизации - 1 шт

• Класс сборки графа - 1 шт

Начнем с инициализации LLM. Для этого необходимо заиметь на нашей GPU две модели. Я выбрал ai-sage/GigaChat3-10B-A1.8B для реализации агента и Qwen/Qwen2.5-Coder-7B-Instruct-AWQ для кодера. Размер моделей как раз позволит разместить их на 32 Гб памяти (с учетом резерва места для кеша и контекста). С выбором инференс-движка вопросов также не возникло - для серьёзных моделей использовать Ollama - это моветон, а поднять ai-sage/GigaChat3-10B-A1.8B в vLLM на архитектуре Blackwell у меня не вышло. Поэтому наш выбор - SGLang (сделаем вид, что я просто забыл про llama.cpp). Поднимать модели будем через docker compose:

  sglang-chat:
    image: lmsysorg/sglang@sha256:7515a1e626d22a8582ad4478d3d68b294d2b484c78a023f8b61c6867454bbf4a
    volumes:
      - ./nlp_models/sglang_cache/huggingface:/root/.cache/huggingface
    env_file:
      - .env
    shm_size: 32g
    ipc: host
    ports:
      - "30001:30000"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]
    networks:
      - app_network
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:30000/health_generate"]
      interval: 30s
      timeout: 10s
      retries: 20
      start_period: 120s
    command: >
      sglang serve
      --model-path ИМЯ_МОДЕЛИ
      --host 0.0.0.0
      --port 30000
      --mem-fraction-static 0.45
      --max-total-tokens 128000
      --cuda-graph-max-bs 16
      --kv-cache-dtype fp8_e4m3

Что тут есть:

• образ с конкретным хешем (в какой-то момент решил для себя, что так правильнее чем использовать теги)

• монтирование папки для модели. При первом запуске модель загрузится в папку /root/.cache/huggingface контейнера, поэтому её надо куда-то прокинуть. Я сделал это в локальную папку.

• env_file, в котором надо обязательно указать свой HF_TOKEN - sglang грузит модели с HuggingFace, и без токена может резаться скорость

• ну и собственно параметры запуска модели. mem-fraction-static резервирует 45% GPU памяти (итого свободным останется около 10%), max-total-tokens ограничит максимум токенов в кеше, kv-cache-dtype заставит хранить KV кеш в 8 бит формате.

Таких секций надо две (под две модели), с разными портами. Ну и настройки под них:

LLM__CHAT__HOST=localhost
LLM__CHAT__PORT=30001
LLM__CHAT__MODEL_NAME=ai-sage/GigaChat3-10B-A1.8B
LLM__CHAT__API_KEY=fake_key
LLM__CHAT__TEMPERATURE=0.2

LLM__CODER__HOST=localhost
LLM__CODER__PORT=30002
LLM__CODER__MODEL_NAME=Qwen/Qwen2.5-Coder-7B-Instruct-AWQ
LLM__CODER__API_KEY=fake_key
LLM__CODER__TEMPERATURE=0.0

SGLang и другие OpenAI-совместимые эндпоинты часто требуют наличия заголовка API-ключа, даже если аутентификация отключена, поэтому передаем любое непустое значение (у меня fake_key)

Перейдём к State. Усложнять его не будем и воспользуемся стандартным узлом MessagesState из LangGraph. Всё, что делает этот узел - хранит историю сообщений. Но для порядка всё-таки создадим файл agent\state.py и пропишем в нём свой класс, унаследованный от MessagesState - во второй части обязательно надо будет полей добавить.

class AgentState(MessagesState):
    pass

Следующим по плану идёт узел-агент. Но так как он часть класса для построения графа, то пока пропустим его и перейдём к самому интересному - MCP инструментам.

Для начала небольшая справка (хотя кого я обманываю, все кто хотел уже про MCP знают, а кто не хотел не будет это читать)

MCP (Model Context Protocol) — это открытый протокол от Anthropic, который позволяет создавать универсальные “инструменты” (например, для поиска, вычислений, работы с БД) и подключать их к любой LLM. MCP отделяет логику инструмента от кода агента. Это позволяет написать инструмент один раз, запустить его на MCP-сервере, и он становится доступен любому клиенту, поддерживающему протокол. При этом агент перестаёт быть жёстко связанным с конкретными функциями — он просто обращается к нужному MCP-серверу, получает список инструментов и использует их по мере необходимости.

В прошлой статье я описывал применение инструментов в LangChain/Graph - это декоратор @tool и наследование от BaseTool. Но MCP это прям новый уровень. И для обеспечения возможности работы LangChain и LangGraph с MCP-инструментами была разработана библиотека langchain-mcp-adapters. Она позволяет “превращать” инструменты с любого MCP-сервера в обычные LangChain-инструменты, понятные агенту. Работает, правда, пока не без изъянов (у меня падает при попытке сделать astream_events с графа), но возможно это не баги, а фичи.

Для начала создадим инструмент из нашего кодера. Воспользуемся инструкцией. Импортируем FastMCP (библиотека сама поставится при установке langchain-mcp-adapters) и создадим экземпляр. В качестве аргумента передаём имя нашего будущего сервера:

from mcp.server.fastmcp import FastMCP
mcp = FastMCP('coder')

Теперь код инструмента. Фактически, это будет узел графа, но завёрнутый в декоратор @mcp.tool(). Грузим из LangFuse промпт, подставляем контекст и задание на кодировку, вызываем модельку-кодер. Обязательно добавляем докстринг, так как это единственная информация об инструменте для нашего агента.

@mcp.tool()
def generate_code(task: str, code_context: str = '') -> str:
    """Генерирует код по заданию пользователя.
    Args:
        task: Краткое описание задачи (например, "напиши функцию factorial с кэшированием")
        code_context: Опциональный контекст — содержимое других файлов для учёта стиля/зависимостей
    Returns:
        Сгенерированный код
    """
    prompt = (
        settings.langfuse.client
        .get_prompt(name='mcp_agent_coder_prompt')
        .compile(
            code_context=code_context.strip() or 'Нет дополнительного контекста. Пиши самодостаточный код.',
            task=task)
    )
    response = settings.llm.coder.llm.invoke(prompt)

    content = response.content
    return content.strip()

Примечание - в уже упомянутой статье я приводил подробное описание, как локально развернуть LangFuse и хранить в нём промпты, поэтому не буду повторяться.

Теперь надо реализовать запуск нашего инструмента:

if __name__ == '__main__':
    mcp.run(transport='stdio')

Всё. transport='stdio' означает, что у нас локальный сервер со стандартным потоком ввода-вывода. Сохраняем наш инструмент в tools/coder_mcp.py.

Перейдем к остальным инструментам. Сделаем следующий джентльменский набор:

• filesystem - инструменты для работы с файловой системой

• git - инструменты для работы с Git

• context7 - инструменты для получения актуальной документации

Для запуска серверов нам потребуется предварительно установить npx (установить пакет Node.js) и uvx для гит-инструментов (уже должен быть, если установлен uv). Для работы context7 необходимо зарегистрироваться на их сайте и получить API-ключ (это бесплатно).

Перед тем, как инициализировать сервер с инструментами, необходимо написать пару служебных утилит для контроля параметров. Проблема в том, что сервер будет падать с ошибкой, если:

1. Директория, переданная параметром к filesystem не существует или вообще не директория

2. В директории, переданной в git, нет репозитория

Дополнительно настроим локализацию через переменные окружения для сервера. Сначала как раз они:

def create_common_env() -> dict:
    common_env = {'PYTHONUTF8': '1', 'NODE_NO_WARNINGS': '1'}
    return {**common_env, 'LC_ALL': 'en_US.UTF-8'} if os.name != 'nt' else common_env

PYTHONUTF8='1' заставляет Python использовать UTF-8 для ввода/вывода, что нужно для корректной работы с великим и могучим, NODE_NO_WARNINGS='1' отключит лишние предупреждения от Node.js. Ну и для Unix-систем дополнительно принудительно ставим английскую локализацию в UTF-8 (честно не помню почему, но явно что-то не работало).

Дальше метод проверки workspace. Проверяем, что это существующая директория и возвращаем абсолютный путь к ней:

def check_workspace(workspace: Path | str) -> Path:
    try:
        workspace = Path(workspace)
        if not workspace.exists():
            raise FileNotFoundError(f'Директория не создана: {workspace}')
        if not workspace.is_dir():
            raise NotADirectoryError(f'Передана не директория: {workspace}')
    except Exception as e:
        logger.error(f'Ошибка валидации "Workspace": {e}')
        raise
    logger.info(f'Установлена рабочая директория "{str(workspace)}"')
    return workspace.absolute()

Проверку существования git-репозитория сделаем просто по наличию каталога .git в workspace:

def is_git_repo_init(workspace: Path | str) -> bool:
    git_dir = workspace / '.git'
    if not git_dir.exists():
        logger.warning(f'Директория не является git-репозиторием: {workspace}')
        logger.warning('Git-инструменты будут отключены')
        return False
    else:
        logger.success('Git-репозиторий найден')
        return True

Сохраним утилиты в файл tools/utils.py и перейдем непосредственно к конфигурации инструментов:

async def init_tools():
    common_env = create_common_env()
    workspace = check_workspace(settings.service.workspace)

    mcp_config = {
        'filesystem': {
            'command': 'npx',
            'args': ['-y', '@modelcontextprotocol/server-filesystem', str(workspace)],
            'env': common_env,
            'cwd': str(workspace),
            'transport': 'stdio',
        },
        'context7': {
            'command': 'npx',
            'args': ['-y', '@upstash/context7-mcp'],
            'env': {**common_env, 'CONTEXT7_API_KEY': settings.service.context7_api_key},
            'transport': 'stdio',
        },
        'coder': {
            'command': sys.executable,
            'args': [str(Path(__file__).parent / 'coder_mcp.py')],
            'env': {**common_env, 'PYTHONPATH': str(Path(__file__).parent.parent)},
            'transport': 'stdio',
        }
    }

Все настройки (ну кроме кастомного coder) можно найти в документации. А в кодер мы дополнительно передаём PYTHONPATH, чтобы он мог найти настройки проекта с классом LLM. Теперь git. Его добавляем только при наличии репозитория:

async def init_tools():
    # код выше ...
    if is_git_repo_init(workspace):
        mcp_config['git'] = {
            'command': 'uvx',
            'args': ['mcp-server-git', '--repository', str(workspace)],
            'env': {**common_env},
            'cwd': str(workspace),
            'transport': 'stdio',
        }

Ну и поднимаем MCP-сервер. Для этого используем класс langchain_mcp_adapters.client.MultiServerMCPClient:

from langchain_mcp_adapters.client import MultiServerMCPClient

async def init_tools():
    # код выше ...
    try:
        mcp_client = MultiServerMCPClient(mcp_config)
        tools = await mcp_client.get_tools()
    except Exception as e:
        logger.error(f'Ошибка создания MultiServerMCPClient: {e}')
        raise
    return tools

Всё. Инструменты готовы к работе. Переходим к агенту.

В этой части статьи агент будет из одного узла, чисто чтобы научиться работать с инструментами. Начнем с конструктора.

class MCPAgent:
    def __init__(self):
        self.llm: ChatOpenAI = settings.llm.chat.llm
        self.llm_with_tools = None
        self.tools: list[BaseTool] | None = None
        self.graph = None
        self.lf_handler = CallbackHandler(public_key=settings.langfuse.public_key)

    async def init_graph(self):
        self.tools = await init_tools()
        self.llm_with_tools = self.llm.bind_tools(self.tools, parallel_tool_calls=False)
        self.graph = self._compile_graph()

Получаем из настроек объект LLM, инициализируем сервер с инструментами, биндим инструменты к нашей языковой модели. Заодно создаём экземпляр LangFuse CallbackHandler для обеспечения трассировки агента.

Теперь узел. Формируем запрос к модели из системного промпта и истории сообщений. Далее вызовем модель с инструментами:

class MCPAgent:
    def agent_node(self, state: MessagesState) -> dict[str, Any]:
        prompt = settings.langfuse.client.get_prompt(name='mcp_agent_prompt').compile()
        prompt_with_history = ChatPromptTemplate.from_messages([
            ('system', prompt),
            MessagesPlaceholder(variable_name='messages'),
        ])
        chain = prompt_with_history | self.llm_with_tools
        response = chain.invoke({'messages': state['messages']})
        return {'messages': [response]}

Собираем граф. Для маршрутизации будем использовать langgraph.prebuilt.tools_condition:

class MCPAgent:
    def _compile_graph(self):
        workflow = StateGraph(MessagesState)
        workflow.add_node('agent', self.agent_node)
        workflow.add_node('tools', ToolNode(tools=self.tools))

        workflow.set_entry_point('agent')
        workflow.add_conditional_edges(
            'agent', tools_condition,
            {'tools': 'tools', '__end__': '__end__'})
        workflow.add_edge('tools', 'agent')

        graph = workflow.compile()
        return graph

Дополним класс методом для инференса. Как писал в начале, у меня не получилось стримить токены при использовании MCP, поэтому будем использовать простой invoke:

class MCPAgent:
    async def run(self, input_messages: dict[str, Any]) -> list[BaseMessage]:
        if self.graph is None:
            raise RuntimeError(
                'Агент не инициализирован. Запустите `initialize()`.')
        return (
            await self.graph.ainvoke(input_messages, config={'callbacks': [self.lf_handler]})
        )['messages']

На этом можно было бы и остановиться, сделав консольный чатик через while(True): или просто каждый раз вызывая python main.py. Но хочется чего-то большего. Поэтому сделаем ленивый Web-интерфейс с помощью библиотеки gradio. Если кто не знаком, то

Gradio — это пакет Python с открытым исходным кодом, который позволяет быстро создать демонстрационную версию или веб-приложение для вашей модели машинного обучения, API или любой произвольной функции на Python. С помощью встроенных функций публикации Gradio вы можете всего за несколько секунд поделиться ссылкой на свою демонстрационную версию или веб-приложение. Никаких навыков работы с JavaScript, CSS или веб-хостингом не требуется! (машинный перевод с README репозитория)

Реализуем класс MCPCodingAgentApp в файле agent/agent_app.py. Для начала напишем методы для создания объекта графа и для сборки визуальной части. Мы используем компонент gr.ChatInterface, который “из коробки” предоставляет готовое окно чата с историей сообщений и примерами запросов.

class MCPCodingAgentApp:
    def __init__(self):
        self.demo = None
        self.agent = None

    async def init_agent(self):
        self.agent = MCPAgent()
        await self.agent.init_graph()
        logger.success('MCP агент инициализирован')

    def build_interface(self):
        with gr.Blocks(title='MCP Coding Agent', fill_height=True) as self.demo:
            gr.Markdown('# MCP Coding Agent')
            gr.Markdown('Помощник разработчика с доступом к файлам, Git и документации')
            chatbot = gr.ChatInterface(
                fn=self.respond,
                title='Чат с агентом',
                examples=[
                    'Напиши функцию вычисления факториала на Python',
                    'прочитай и выведи содержимое файла test.py',
                    'Как использовать requests в Python?',
                ],
            )

gr.Blocks - это основной контейнер для компонентов. В него мы добавляем две строчки текста (заголовок и простое описание) и чат gr.ChatInterface. Единственный обязательный параметр для gr.ChatInterface - это функция-обработчик новых сообщений. Остальное - просто для красоты. Теперь напишем сам обработчик. Согласно документации, это должна быть функция, принимающая два значения: новое сообщение типа str и история сообщений в OpenAI стиле (list словарей вида {'role': , 'content': }):

class MCPCodingAgentApp:
    # код выше
    async def respond(self, message: str, history: list[dict[str, str]]) -> str:
        if self.agent is None:
            logger.error('Агент не инициализирован')
            return 'Агент не инициализирован'
    
        messages = []
        for msg in history:
            if msg['role'] == 'user':
                messages.append(HumanMessage(content=msg['content']))
            elif msg['role'] == 'assistant':
                messages.append(AIMessage(content=msg['content']))
        messages.append(HumanMessage(content=message))
    
        result = await self.agent.run({'messages': messages})
        return result[-1].content

Проверяем, что агент инициализирован, приводим сообщения к LangChain классам и передаём всё это богатство для инференса в агент. Осталось связать всё воедино и запустить сервер.

class MCPCodingAgentApp:
    # код выше
    async def initialize_and_launch(self):
        await self.init_agent()
        self.build_interface()
        self.demo.launch(
            server_name=settings.service.host,
            server_port=settings.service.port,
            show_error=True,
        )

    def run(self):
        asyncio.run(self.initialize_and_launch())

В main файле создадим экземпляр GUI-класса и вызовем метод run():

if __name__ == "__main__":
    try:
        app = MCPCodingAgentApp()
        app.run()
    except KeyboardInterrupt:
        logger.info('Выполнение прервано пользователем')
    except Exception as e:
        logger.error(e)

Если всё сделано правильно (в том числе запущен докер с моделями), то после вызова python main.py увидим такую красоту:

Веб интерфейс агента

Попробуем поработать с агентом. Для этого создадим тестовую директорию и инициализируем в ней Git-репозиторий. На этом момента фантазия уже закончилась, поэтому запрос максимально простой:

Пример общения

Ну и результат выполнения: файл с примитивным кодом - 1 шт, коммит (даже с нормальным текстом) - 1 шт.

Результат работы

Ну и трейсы можно глянуть, на всякий случай:

Трейсы

На каждом шаге агент вызывает корректный инструмент. Поздравляю, мы получили агента-выпускника ПТУ (синьоры ПТУ-шники, без обид) - при корректном и компактном ТЗ вполне себе работает.

Что удалось реализовать в этой части:

• Базовую базу агента - создали структуру, настроили конфигурацию с валидацией

• Локальный инференс - развернули две модели через SGLang в Docker

• MCP-инструменты - написали свой инструмент, подключили несколько стандартных

• Написали ядро агента

Код проекта доступен тут (там в репозитории отдельная веточка для этой части статьи).

Вторая часть в которой показана реализация узлов тут

Третья часть со сборкой графа тут