Привет, Хабр! Меня зовут Владимир и я стал немного более GPU-rich. А это значит, что пора сдуть пыль со старого проекта)
Некоторое время назад захотелось мне немного повайбкодить. Но злобные корпорации не захотели брать мои рубли, поэтому я решил
Имеющаяся в то время в наличии 5080 на 16 Гб не дала мне развернуть адекватные модели, а загоняться с постоянным высвобождением памяти и перезагрузкой моделей мне было не охота, поэтому проект был заброшен. Но недавно я стал счастливым обладателем 5090 на 32 Гб, а это значит, что настало время новой попытки!
Статьи будет две — в первой мы создадим необходимую инфраструктуру, напишем простого агента, а также добавим нашему агенту MCP-инструменты. Во второй усложним логику агента и контейнеризируем его.
Подготовительное
Первым делом установим зависимости
uv inituv 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: LLMConfigclass 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_cachedef get_settings() -> Settings: try: return Settings() except Exception as e: print(f'Ошибка создания Settings: {e}') raisesettings = 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=localhostLLM__CHAT__PORT=30001LLM__CHAT__MODEL_NAME=ai-sage/GigaChat3-10B-A1.8BLLM__CHAT__API_KEY=fake_keyLLM__CHAT__TEMPERATURE=0.2LLM__CODER__HOST=localhostLLM__CODER__PORT=30002LLM__CODER__MODEL_NAME=Qwen/Qwen2.5-Coder-7B-Instruct-AWQLLM__CODER__API_KEY=fake_keyLLM__CODER__TEMPERATURE=0.0
SGLang и другие OpenAI-совместимые эндпоинты часто требуют наличия заголовка API-ключа, даже если аутентификация отключена, поэтому передаем любое непустое значение (у меня fake_key)
Перейдём к State. Усложнять его не будем и воспользуемся стандартным узлом MessagesState из LangGraph. Всё, что делает этот узел — хранит историю сообщений. Но для порядка всё-таки создадим файл agent\state.py и пропишем в нём свой класс, унаследованный от MessagesState — во второй части обязательно надо будет полей добавить.
class AgentState(MessagesState): pass
Следующим по плану идёт узел-агент. Но так как он часть класса для построения графа, то пока пропустим его и перейдём к самому интересному — MCP инструментам.
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 FastMCPmcp = 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-ключ (это бесплатно).
Перед тем, как инициализировать сервер с инструментами, необходимо написать пару служебных утилит для контроля параметров. Проблема в том, что сервер будет падать с ошибкой, если:
-
Директория, переданная параметром к
filesystemне существует или вообще не директория -
В директории, переданной в
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 MultiServerMCPClientasync 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-инструменты — написали свой инструмент, подключили несколько стандартных
-
Написали ядро агента
Код проекта доступен тут (там в репозитории отдельная веточка для этой части статьи).
ссылка на оригинал статьи https://habr.com/ru/articles/1043348/