Створення AI-агентів за допомогою 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
Порти створюють межі
Для Ports необхідно створити межі, визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цю стадію як контракт між вхідними даними та перевіреними результатами. Призначте назви для результатів роботи, визначте критерії успіху та не допускайте беззвучного часткового завершення. Використовуйте обмеження та індекси перед масовим введенням даних. Унікальність бізнес-ключів перетворює подальшу об’єднання даних на прогнозовані операції додавання чи заміни, а не на ситуації з дублікатами. Для Ports необхідно створити межі, визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Зберігайте конфігурацію окремо від коду програми. Файли середовища, сховища секретних даних та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевіряти, не читаючи весь код.
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 probes дозволяють виявити зміни міток та відсутність властивостей дешевше, ніж тести чату від початку до кінця.
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