Ця стаття — практичний посібник для розробників зі створення автономних ШІ-агентів на Python. Ми не будемо повторювати теорію про те, що таке LangChain та LangGraph. Натомість ми зосередимося на коді, архітектурі та вирішенні реальних завдань.
Мета: Створити з нуля два проєкти:
- Агент-класифікатор: Багатокроковий агент з керованим станом, але без зовнішніх інструментів.
- Агент-асистент: Повноцінний агент з доступом до файлової системи та веб-пошуку через протокол MCP, побудований на циклічній логіці.
Ми розглянемо найкращі практики: керування конфігурацією, вибір моделей та обробку помилок для створення надійних систем.
Коротко про концепції: Агент та міст MCP
Перш ніж перейти до коду, зафіксуємо два поняття:
- ШІ-агент: Програма, побудована навколо циклу «міркування-дія». Вона отримує завдання, за допомогою LLM вирішує, що робити далі (наприклад, викликати інструмент), виконує дію і повторює цикл, доки завдання не буде виконано.
- MCP (Model Context Protocol): Стандарт, що слугує мостом між логікою агента та зовнішніми інструментами. Він дозволяє агенту уніфіковано працювати з файлами, API чи пошуком, не переймаючись деталями їх реалізації.
Частина 1: Налаштування надійного середовища
Крок 1: Віртуальне середовище та залежності
Створіть та активуйте віртуальне середовище. Потім створіть файл requirements.txt:
# Основні фреймворки
langchain
langgraph
# Адаптери для моделей
langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # Для Ollama
# Інструменти та протоколи
langchain-mcp-adapters
mcp
ollama
# Допоміжні утиліти
python-dotenv
tenacity # для надійної обробки помилокВстановіть залежності:
pip install -r requirements.txt```
#### Крок 2: Конфігурація API-ключів
Створіть файл `.env` для зберігання ваших ключів:OPENAI_API_KEY=”sk-…”
GOOGLE_API_KEY=”AIzaSy…”
MISTRAL_API_KEY=”…”
BRAVE_API_KEY=”…” # для інструменту веб-пошуку через MCP
#### Крок 3: Патерн «Фабрика моделей»
Щоб гнучко перемикатися між хмарними та локальними моделями, не змінюючи код агента, використаємо патерн «фабрика».python
llm_factory.py
import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama
load_dotenv()
class ModelProvider(Enum):
OPENAI = “openai”
GEMINI = “gemini”
MISTRAL_API = “mistral_api”
OLLAMA = “ollama”
def get_llm(provider: ModelProvider, model_name: str = None):
“””Фабрика для створення екземплярів LLM.”””
if provider == ModelProvider.OPENAI:
return ChatOpenAI(model=model_name or “gpt-4o-mini”, temperature=0)
elif provider == ModelProvider.GEMINI:
return ChatGoogleGenerativeAI(model=model_name or “gemini-1.5-flash”, temperature=0)
elif provider == ModelProvider.MISTRAL_API:
return ChatMistralAI(model=model_name or “mistral-large-latest”, temperature=0)
elif provider == ModelProvider.OLLAMA:
# Переконайтеся, що у вас запущена Ollama з потрібною моделлю
# docker exec -it ollama ollama pull mistral
return ChatOllama(model=model_name or “mistral”, temperature=0)
raise ValueError(f”Невідомий провайдер моделі: {provider}”)
Приклад використання
if name == “main“:
# local_llm = get_llm(ModelProvider.OLLAMA)
openai_llm = get_llm(ModelProvider.OPENAI)
response = openai_llm.invoke(“Поясни концепцію RAG у трьох реченнях.”)
print(response.content)“`
Частина 2: Проєкт 1 — Агент для класифікації вакансій
Цей агент демонструє, як використовувати LangGraph для створення лінійного графа з керованим станом. Він прийматиме опис вакансії та послідовно класифікуватиме його за трьома параметрами.
Крок 1: Визначення стану
Стан — це «пам’ять» нашого графа, що передається від одного вузла до іншого.
# vacancy_classifier.py
from typing import TypedDict, Dict
class ClassificationState(TypedDict):
"""Стан для агента-класифікатора."""
description: str # Вихідний текст
job_type: str # Тип роботи (проєктна/постійна)
category: str # Професія
search_type: str # Мета (пошук роботи/виконавця)
classification_log: list # Журнал налагодженняКрок 2: Реалізація вузлів графа
Кожен вузол — це функція, яка приймає стан, виконує свою частину роботи та повертає оновлений стан.
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider
class VacancyClassifierAgent:
def __init__(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name="gpt-4o-mini")
async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
"""Вузол 1: Визначає тип роботи."""
prompt = ChatPromptTemplate.from_messages([
("system", "Визнач тип роботи. Відповідь має бути 'проєктна' або 'постійна'."),
("human", "Текст вакансії:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["job_type"] = result.content.strip()
state["classification_log"].append("Визначено тип роботи.")
return state
async def _classify_category(self, state: ClassificationState) -> ClassificationState:
"""Вузол 2: Визначає категорію професії."""
# Категорії можна завантажити з файлу або бази даних
categories = ["Python-розробник", "Дизайнер", "Маркетолог", "3D-аніматор"]
prompt = ChatPromptTemplate.from_messages([
("system", f"Вибери найбільш відповідну категорію зі списку: {', '.join(categories)}."),
("human", "Текст вакансії:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["category"] = result.content.strip()
state["classification_log"].append("Визначено категорію.")
return state
async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
"""Вузол 3: Визначає мету пошуку."""
prompt = ChatPromptTemplate.from_messages([
("system", "Визнач мету автора. Відповідь має бути 'пошук роботи' або 'пошук виконавця'."),
("human", "Текст вакансії:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["search_type"] = result.content.strip()
state["classification_log"].append("Визначено мету пошуку.")
return stateКрок 3: Збирання та запуск графа
Збираємо вузли в єдиний робочий процес.
# ... продовження класу VacancyClassifierAgent ...
from langgraph.graph import StateGraph, END
def build_graph(self):
"""Збирає граф станів."""
workflow = StateGraph(ClassificationState)
workflow.add_node("job_type_classifier", self._classify_job_type)
workflow.add_node("category_classifier", self._classify_category)
workflow.add_node("search_type_classifier", self._classify_search_type)
workflow.set_entry_point("job_type_classifier")
workflow.add_edge("job_type_classifier", "category_classifier")
workflow.add_edge("category_classifier", "search_type_classifier")
workflow.add_edge("search_type_classifier", END)
return workflow.compile()
async def main():
agent = VacancyClassifierAgent()
graph = agent.build_graph()
description = "Шукаємо досвідченого Python-розробника в команду на фултайм для роботи над проєктом у сфері фінтех."
initial_state = ClassificationState(
description=description,
job_type="", category="", search_type="",
classification_log=[]
)
final_state = await graph.ainvoke(initial_state)
print("--- Результат класифікації ---")
print(json.dumps(final_state, indent=2, ensure_ascii=False))
if __name__ == "__main__":
asyncio.run(main())Частина 3: Проєкт 2 — Агент-асистент з інструментами (MCP)
Цей агент демонструє циклічну логіку, де він може багаторазово звертатися до інструментів для вирішення завдання.
Крок 1: Керування конфігурацією
Для агентів, що взаємодіють із зовнішнім світом, важлива надійна конфігурація.
# mcp_agent_config.py
from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider
@dataclass
class AgentConfig:
workdir: str = "./agent_workdir"
model_provider: ModelProvider = ModelProvider.OLLAMA
def __post_init__(self):
"""Валідація після ініціалізації."""
os.makedirs(self.workdir, exist_ok=True)Крок 2: Визначення стану для діалогу
Стан тепер буде зберігати історію повідомлень.
# mcp_agent.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]Крок 3: Реалізація циклічного графа
Граф складатиметься з двох основних вузлів та умовного переходу, який створює цикл «міркування-дія».
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig
class MCPAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None
async def setup_tools(self):
"""Ініціалізація інструментів через MCP."""
mcp_config = {
"filesystem": {
"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
"transport": "stdio"
},
# Додайте brave-search, якщо є ключ BRAVE_API_KEY
}
mcp_client = MultiServerMCPClient(mcp_config)
self.tools = await mcp_client.get_tools()
self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])
# Прив'язуємо інструменти до моделі
self.llm = self.llm.bind_tools(self.tools)
def _should_continue(self, state: AgentState):
"""Умовний перехід: вирішуємо, чи потрібно викликати інструмент."""
last_message = state['messages'][-1]
if not last_message.tool_calls:
return "end"
return "continue"
def _call_model(self, state: AgentState):
"""Вузол 1: Виклик LLM для прийняття рішення."""
response = self.llm.invoke(state['messages'])
return {"messages": [response]}
def _call_tool(self, state: AgentState):
"""Вузол 2: Виконання виклику інструменту."""
last_message = state['messages'][-1]
tool_call = last_message.tool_calls[0]
action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
response = self.tool_executor.invoke(action)
return {"messages": [response]}
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("agent", self._call_model)
workflow.add_node("action", self._call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
self._should_continue,
{"continue": "action", "end": END}
)
workflow.add_edge("action", "agent")
return workflow.compile()Крок 4: Запуск та взаємодія
# ... продовження mcp_agent.py ...
import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
"""Запускає завдання з обробкою помилок."""
return await graph.ainvoke({"messages": [HumanMessage(content=task)]})
async def main():
config = AgentConfig(model_provider=ModelProvider.OPENAI) # або OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()
task = "Створи файл 'hello.txt' у робочій директорії та запиши в нього 'Привіт, світ!'."
result = await run_agent_task(graph, task)
print("\n--- Фінальна відповідь агента ---")
print(result['messages'][-1].content)
if __name__ == "__main__":
asyncio.run(main())Тут ми додали декоратор tenacity для надійності — якщо виклик агента впаде через тимчасову мережеву помилку, він автоматично повториться.
Висновок
Ми створили два типи агентів, використовуючи сучасні практики:
- Лінійний граф чудово підходить для завдань з чіткою послідовністю кроків, таких як ETL-процеси або багатоетапний аналіз.
- Циклічний граф є основою для створення інтерактивних асистентів та автономних агентів, здатних вирішувати складні завдання за допомогою інструментів.
Представлені архітектурні патерни — фабрика моделей, керування конфігурацією, розділення логіки на вузли та використання графів станів — є фундаментом для побудови масштабованих та надійних ШІ-систем.