Создание ИИ-агентов с использованием FastAPI, LangGraph и принципов чистой архитектуры

В этом руководстве пошагово показан путь от сырья до готовой системы для проекта «Как я структурирую ИИ-агентов для производства с использованием FastAPI, LangGraph и принципов чистой архитектуры». Основное внимание уделяется практическим шагам, четким проверкам и коду, который можно просто добавить в репозиторий без необходимости догадываться о его назначении. Для получения обзора определите входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь код.

@router.post("/suppliers")
async def create_supplier(request: SupplierRequest):
    policies = opensearch.search(
            index="supplier-policies",
            query=request.description,
        )
        response = bedrock.converse(
            modelId=MODEL_ID,
            messages=build_messages(request, policies),
        )
        supplier = Supplier(
            name=request.name,
            tax_id=request.tax_id,
        )
        db.add(supplier)
        db.commit()
        sqs.send_message(
            QueueUrl=SUPPLIER_QUEUE,
            MessageBody=serialize(supplier),
        )
        return {"status": "created"}

FastAPI — это интерфейс, а не само приложение

Поскольку FastAPI представляет собой интерфейс, а не само приложение, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо одновременно задокументировать успешный и аварийный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом доработки позже. Используйте ограничения и индексы перед массовым вводом данных. Обеспечение уникальности по бизнес-ключам позволяет превращать последующие объединения в предсказуемые операции добавления или замены данных, а не в ситуации с дубликатами.

@router.post("/suppliers")
async def create_supplier(
    request: CreateSupplierRequest,
    use_case: CreateSupplierUseCase = Depends(
        get_create_supplier_use_case
    ),
):
    command = CreateSupplierCommand(
        name=request.name,
        tax_id=request.tax_id,
        country=request.country,
    )

    result = await use_case.execute(command)
    return CreateSupplierResponse.from_result(result)

HTTP Request
     ↓
   FastAPI
     ↓
 Application
     ↓
   Domain
     ↓
    Ports
     ↓
  Adapters

Приложение взаимодействует с функциями, а не с технологиями

Поскольку приложение взаимодействует с функциями, а не с технологиями, необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную ответственность, а не на запутанную цепочку операций. Применяйте ограничения и индексы перед массовым вводом данных. Уникальность по бизнес-ключам превращает последующие объединения в предсказуемые операции добавления или замены данных, а не в ситуацию дубликатов.

class CreateSupplierUseCase:
    def __init__(
        self,
        repository: SupplierRepository,
        policy_service: SupplierPolicyService,
        event_publisher: EventPublisher,
    ):
        self.repository = repository
        self.policy_service = policy_service
        self.event_publisher = event_publisher
    async def execute(
        self,
        command: CreateSupplierCommand,
    ) -> Supplier:
        existing = await self.repository.find_by_tax_id(
            command.tax_id
        )
        if existing:
            raise SupplierAlreadyExists(command.tax_id)
        policy = await self.policy_service.evaluate(
            country=command.country
        )
        supplier = Supplier.create(
            name=command.name,
            tax_id=command.tax_id,
            country=command.country,
            requires_approval=policy.requires_approval,
        )
        await self.repository.add(supplier)
        await self.event_publisher.publish(
            SupplierCreated(supplier.id)
        )
        return supplier

Порты определяют границы

Для портов необходимо сначала определить границы, ввести параметры входных данных, указать ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия создаваемым объектам, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Используйте ограничения и индексы перед массовым вводом данных. Уникальность бизнес-ключей позволяет превращать последующие объединения в предсказуемые операции добавления или обновления данных, а не в ситуации с дубликатами. Для портов необходимо сначала определить границы, ввести параметры входных данных, указать ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь код.

ph.

from typing import Protocol

class SupplierRepository(Protocol):
    async def find_by_tax_id(
        self,
        tax_id: str,
    ) -> Supplier | None:
        ...
    async def add(
        self,
        supplier: Supplier,
    ) -> None:
        ...

class PolicyRetriever(Protocol):
    async def retrieve(
        self,
        query: str,
    ) -> list[PolicyDocument]:
        ...

class LLMProvider(Protocol):
    async def reason(
        self,
        context: AgentContext,
    ) -> AgentDecision:
        ...

class EventPublisher(Protocol):
    async def publish(
        self,
        event: DomainEvent,
    ) -> None:
        ...

SupplierRepository
PolicyRetriever
LLMProvider
EventPublisher

PostgreSQL
Amazon OpenSearch
Amazon Bedrock
AWS SQS

Какое место занимает Amazon Bedrock

При определении того, какое место занимает Amazon Bedrock, необходимо сначала определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не последующими улучшениями. После первой партии необходимо проверить примеры данных из окружения. Инструменты типа Cypher позволяют выявить смещение меток и отсутствие свойств дешевле, чем тесты общения от начала до конца.

class BedrockLLMProvider(LLMProvider):
    def __init__(self, client, model_id: str):
        self.client = client
        self.model_id = model_id
    async def reason(
        self,
        context: AgentContext,
    ) -> AgentDecision:
        response = self.client.converse(
            modelId=self.model_id,
            messages=build_messages(context),
        )
        return map_bedrock_response(response)

Application
     ↓
LLMProvider
     ↑
BedrockLLMProvider

Какое место занимает RAG

Чтобы определить место применения RAG, необходимо заранее описать входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага он должен указывать на конкретную проблему, а не на сложную структуру обработки данных. После первой партии данных следует проверить примеры из окружающей области. Инструменты типа Cypher позволяют выявить смещение меток и отсутствие свойств дешевле, чем тесты общения от начала до конца.

FastAPI
   ↓
OpenSearch
   ↓
LLM

class OpenSearchPolicyRetriever(PolicyRetriever):
    def __init__(
        self,
        opensearch_client,
        embedding_provider,
    ):
        self.client = opensearch_client
        self.embedding_provider = embedding_provider
    async def retrieve(
        self,
        query: str,
    ) -> list[PolicyDocument]:
        vector = await self.embedding_provider.embed(query)
        results = self.client.search(
            index="supplier-policies",
            body=build_vector_query(vector),
        )
        return map_documents(results)

RAG обеспечивает контекст. Он не принимает решений.

Для RAG предоставляется контекст, но он не принимает решений. Перед изменением кода необходимо определить входные данные, ответственного за выполнение шага и критерии завершения. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Укажите названия элементов, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Проверьте пример группы данных после первой партии. Инструменты типа Cypher позволяют выявить смещение меток и отсутствие свойств дешевле, чем тесты в формате полного обмена сообщениями. Для RAG предоставляется контекст, но он не принимает решений. Перед изменением кода необходимо определить входные данные, ответственного за выполнение шага и критерии завершения. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, доступном для операторов.

Можно провести аудит, не читая весь граф.

RAG
 ↓
LLM
 ↓
"Looks fine"
 ↓
Create Supplier

RAG
 ↓
Relevant Context
 ↓
Agent Reasoning
 ↓
Application
 ↓
Domain Rules
 ↓
Decision

Какое место занимает LangGraph

Чтобы определить, какое место занимает LangGraph, необходимо сначала задать входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не последующими улучшениями. Необходимо создавать узлы и связи модели для тех вопросов, которые вы собираетесь задавать, а не для каждого существительного в документе. Редкие, типизированные ребра лучше, чем плотные загадочные графы.

Understand Request
       ↓
Retrieve Policies
       ↓
Evaluate Information
       ↓
Need More Data?
   ↙          ↘
 Yes          No
  ↓            ↓
Ask User   Continue
               ↓
       Approval Required?
          ↙          ↘
        Yes           No
         ↓             ↓
   Human Approval   Continue
          ↘          ↙
        Request Action

graph = StateGraph(AgentState)
graph.add_node(
    "understand_intent",
    understand_intent,
)
graph.add_node(
    "retrieve_policies",
    retrieve_policies,
)
graph.add_node(
    "evaluate",
    evaluate_request,
)
graph.add_node(
    "human_approval",
    request_human_approval,
)
graph.add_node(
    "request_action",
    request_application_action,
)

graph.add_edge(
    "understand_intent",
    "retrieve_policies",
)
graph.add_edge(
    "retrieve_policies",
    "evaluate",
)
graph.add_conditional_edges(
    "evaluate",
    determine_next_step,
    {
        "approval": "human_approval",
        "execute": "request_action",
    },
)

LangGraph не должен стать новым монолитом

Чтобы LangGraph не превратился в новый монолит, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на запутанную цепочку операций. Необходимо моделировать узлы и связи только для тех вопросов, которые вы собираетесь задать, а не для каждого существительного в документе. Редкие, типизированные связи лучше, чем плотные, непонятные графы.

def evaluate_node(state):
    if state.amount > 100_000:
        state.requires_approval = True
    if state.country == "BR":
        ...
    if state.supplier_type == "CRITICAL":
        ...

decision = supplier_policy.evaluate(
    supplier=supplier,
    context=context,
)

LangGraph → orchestrates
Domain → decides

Где применяется MCP

Чтобы определить, где подходит MCP, необходимо заранее указать входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Моделируйте узлы и связи только для тех вопросов, которые вы собираетесь задать, а не для каждого существительного в документе. Редкие, типизированные связи лучше, чем плотные непонятные графы. Чтобы определить, где подходит MCP, необходимо заранее указать входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь граф.

LLM
 ↓
PostgreSQL

LLM
 ↓
SAP

Agent
  ↓
MCP Tool
  ↓
Application API
  ↓
Use Case
  ↓
Domain
  ↓
Infrastructure

create_supplier

execute_sql

PostgreSQL также является адаптером

В случае использования PostgreSQL в качестве адаптера необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние. Необходимо одновременно задокументировать успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не дополнительными улучшениями, внесенными позже. Перед массовой загрузкой следует использовать ограничения и индексы. Обеспечение уникальности по бизнес-ключам позволяет превращать последующие объединения данных в предсказуемые операции добавления или замены записей вместо возникновения дубликатов.

class SupplierRepository(Protocol):
    async def add(
        self,
        supplier: Supplier,
    ) -> None:
        ...

class PostgresSupplierRepository(
    SupplierRepository
):
    def __init__(self, session):
        self.session = session
    async def add(
        self,
        supplier: Supplier,
    ) -> None:
        entity = SupplierModel.from_domain(
            supplier
        )
        self.session.add(entity)

FastAPI      → doesn't know PostgreSQL exists

Application  → doesn't know PostgreSQL existsDomain       → doesn't know PostgreSQL existsInfrastructure → does

SAP должен находиться за другой границей

Для SAP, который должен находиться за дополнительным барьером, необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную ответственность, а не на запутанную цепочку операций. Применяйте ограничения и индексы перед массовым вводом данных. Уникальность бизнес-ключей позволяет превращать последующие объединения в предсказуемые операции добавления или замены данных, а не в ситуацию дубликатов.

CreateSupplierUseCase
        ↓
Persist State
        ↓
SupplierCreated
        ↓
AWS SQS
        ↓
Integration Worker
        ↓
SAP

Сборка архитектуры

Для формирования архитектуры необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия результатам работы, определите критерии успеха и не допускайте молчаливого частичного завершения задачи. Используйте ограничения и индексы перед массовым вводом данных. Уникальность по бизнес-ключам превращает последующие объединения в предсказуемые операции добавления или замены данных, а не в ситуацию дублирования. Для формирования архитектуры необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять без специальных настроек.

Перевести весь граф.

USER
                           │
                           ▼
                       FastAPI
                           │
                           ▼
                       LangGraph
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
             RAG          LLM          MCP
              │            │            │
              ▼            ▼            │
         OpenSearch     Bedrock         │
                                        ▼
                                  Application
                                        │
                                        ▼
                                     Domain
                                        │
                              ┌─────────┴─────────┐
                              │                   │
                              ▼                   ▼
                         PostgreSQL            AWS SQS
                                                  │
                                                  ▼
                                               Worker
                                                  │
                                                  ▼
                                                 SAP

Два мира

Для проекта «Два мира» необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо одновременно задокументировать успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. После первой партии необходимо проверить примерный набор элементов соседства. Инструменты типа Cypher позволяют выявить смещение меток и отсутствие свойств дешевле, чем тесты общения от начала до конца.

PROBABILISTICLLM
RAG
Natural Language
Intent Understanding
Agent Reasoning
Tool Selection

DETERMINISTIC
Authorization
Business Rules
Transactions
Persistence
Idempotency
Integration
Auditability

Практический тест архитектуры

Для практического тестирования архитектуры необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную проблему, а не на сложную структуру обработки данных. Проверьте примерный набор данных после первой партии. Инструменты вроде Cypher позволяют выявить смещение меток и отсутствие свойств дешевле, чем тесты общения от начала до конца.

async def test_supplier_requires_approval():
    repository = FakeSupplierRepository()
    policies = FakePolicyService(
        requires_approval=True
    )
    events = FakeEventPublisher()
    use_case = CreateSupplierUseCase(
        repository=repository,
        policy_service=policies,
        event_publisher=events,
    )
    supplier = await use_case.execute(
        CreateSupplierCommand(
            name="ACME",
            tax_id="123",
            country="BR",
        )
    )
    assert supplier.requires_approval is True

Изменение модели должно быть скучным

Чтобы изменение модели было скучным процессом, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия результатам работы, определите критерии успеха и не допускайте молчаливого частичного завершения задачи. Проверьте примерный набор данных после первой партии обработки. Инструменты типа Cypher позволяют выявить смещение меток и отсутствие свойств дешевле, чем тесты в формате полного обмена сообщениями. Чтобы изменение модели было скучным процессом, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять без необходимости чтения всего кода.

Всю графику.

LLMProvider
             ▲
             │
   ┌─────────┴─────────┐
   │                   │
BedrockProvider   AnotherProvider

FastAPI
Application
Domain
Business Rules

Чистая архитектура — это не структура папок

В подходе «Чистая архитектура — это не структура папок» необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный пути выполнения. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не последующими улучшениями. Моделируйте узлы и связи для тех вопросов, которые вы собираетесь задать, а не для каждого существительного в документе. Редкие, типизированные связи лучше, чем плотные, непонятные графы.

src/
├── api/
├── application/
├── domain/
├── agents/
├── ports/
└── infrastructure/

Искусственный интеллект делает эти границы ещё более важными

Поскольку искусственный интеллект делает эти границы более важными, необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. Когда шаг терпит неудачу, причина должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Определяйте узлы и связи модели для тех вопросов, которые вы собираетесь задать, а не для каждого существительного в документе. Редкие, типизированные связи лучше, чем плотные, непонятные графы.

Архитектура в одном предложении

Для формулировки архитектуры в одном предложении необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения. Моделируйте узлы и связи для тех вопросов, которые вы собираетесь задать, а не для каждого существительного в документе. Редкие, типизированные связи лучше, чем плотные загадочные графы. Для формулировки архитектуры в одном предложении необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, которое операторы могут проверять без необходимости читать весь код.

весь граф.

В разделе «Заключение» необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Документируйте как успешный путь выполнения, так и пути восстановления одновременно. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. Используйте ограничения и индексы перед массовым вводом данных. Уникальность по бизнес-ключам превращает последующие объединения в предсказуемые операции добавления или замены данных, а не в ситуацию дубликатов.

Prompt
  ↓
 LLM
  ↓
 Tool
  ↓
Database

• Структура бизнес-команды в области ИИ: координация агентов с использованием LangGraph и FastAPI — Подробный обзор платформы с открытым исходным кодом для работы с несколькими агентами: как координируются агенты-сооснователи, менеджеры и специалисты, как они обмениваются информацией, приостанавливают работу в ожидании одобрения и предоставляют отчеты.
• Координация агентов-исследователей и программистов с использованием LangGraph — Создание многопоточного приложения на LangGraph с использованием инструментов вроде soul prompts, моделей Ollama, функций поиска Tavily, инструментов передачи заданий и точек контроля Postgres для организации рабочих процессов под руководством координатора.

Безопасность агентов и тестирование на прочность для архитектур с агентами — Модели угроз для инструментов, корпуса данных для тестирования на прочность, политики автоматического отключения при сбоях и механизмы контроля, способные выдержать воздействие в реальных условиях.

Агенты для возврата средств: LangGraph выжил там, где провалились CrewAI и AutoGen — Одни и те же инструменты и политики в трех разных фреймворках — только явное описание состояния, идемпотентность и точки контроля позволили выжить в условиях хаоса.

После урока LangGraph: жесткие границы, схемы и слои безопасности — Как превратить рабочего SQL-агента в защищенный с использованием типизированных механизмов защиты, обновления схем, оптимизации затрат и многоуровневых проверок.