Графы финансовых преступлений с участием нескольких агентов с использованием LangChain, LangGraph и LangSmith

Проблемы на практике

Анализ финансовых преступлений — это не одно единое задание. Для рассмотрения того или иного случая может потребоваться углубленная проверка данных клиентов, анализ санкций, изучение негативных упоминаний в СМИ, выявление закономерностей в транзакциях и составление отчета для следователей; каждый из этих этапов требует разных инструментов, сроков выполнения и способов преодоления ошибок. Единая система, пытающаяся справиться со всем самостоятельно, становится непрозрачной, трудной для тестирования и рискованной при сбоях отдельных компонентов.

Почему множество специализированных агентов лучше одной гигантской системы

Специализированные агенты позволяют сосредоточиться на конкретных задачах и инструментах. Руководитель может распределять работу между агентами. Сбои остаются локальными. Человеческий контроль может блокировать операции с высоким риском. Возможность отслеживания позволяет определять затраты и ошибки для каждой роли, а не для всей системы в целом.

Три уровня: LangChain, LangGraph, LangSmith

LangChain предоставляет модели, инструменты и механизмы обработки сообщений. LangGraph отвечает за оркестрацию с учетом состояния — общее состояние дела, связи между элементами, сохранение данных и обработка прерываний. LangSmith занимается отслеживанием действий, работой с наборами данных и оценкой результатов. Агент, реализованный с использованием простого цикла while, может вызывать инструменты, но редко обеспечивает долговременное хранение информации о деле, возможность параллельных проверок, повторное использование подграфов или первоклассную поддержку HITL.

Общее состояние дела

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

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from operator import add

class CaseState(TypedDict):
    case_id: str
    customer_id: str
    transactions: list[dict]
    messages: Annotated[list, add_messages]
    kyc_flags: Annotated[list[str], add]
    sanctions_hits: Annotated[list[dict], add]
    risk_score: float
    narrative: str
    requires_human_review: bool

Один узел агента с инструментами

Подключите модель к доменным инструментам (поиск санкций, проверка списка под контролем) и позвольте узлу инструментов выполнять соответствующие вызовы.

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

@tool
def screen_ofac(name: str, dob: str | None = None) -> dict:
    """Screen an individual or entity name against the UK list."""
    # In production: call a real screening API (e.g. UK100, Refinitiv)
    return {"name": name, "match": False, "score": 0.02}

@tool
def screen_un_sanctions(name: str) -> dict:
    """Screen a name against the UN Consolidated Sanctions List."""
    return {"name": name, "match": False, "score": 0.0}
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
sanctions_llm = llm.bind_tools([screen_ofac, screen_un_sanctions])

def sanctions_agent(state: CaseState) -> dict:
    response = sanctions_llm.invoke(state["messages"])
    return {"messages": [response]}

Связывание узлов графа

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

from langgraph.prebuilt import ToolNode, tools_condition

graph = StateGraph(CaseState)
graph.add_node("kyc_agent", kyc_agent)
graph.add_node("sanctions_agent", sanctions_agent)
graph.add_node("sanctions_tools", ToolNode([screen_ofac, screen_un_sanctions]))
graph.add_node("tx_monitoring_agent", tx_monitoring_agent)
graph.add_node("narrative_agent", narrative_agent)
graph.set_entry_point("kyc_agent")
graph.add_edge("kyc_agent", "sanctions_agent")
# Conditional edge: loop back to tools if the LLM requested a tool call
graph.add_conditional_edges(
    "sanctions_agent",
    tools_condition,
    {"tools": "sanctions_tools", END: "tx_monitoring_agent"},
)
graph.add_edge("sanctions_tools", "sanctions_agent")
graph.add_edge("tx_monitoring_agent", "narrative_agent")
graph.add_edge("narrative_agent", END)
app = graph.compile()

Тестируйте узлы с фиктивным состоянием перед формированием полной графики:

def test_sanctions_agent_flags_high_score_match():
    fake_state = {
        "messages": [HumanMessage(content="Screen: Igor Petrov, DOB 1978-03-11")],
    }
    result = sanctions_agent(fake_state)
    assert "tool_calls" in result["messages"][-1].additional_kwargs

Маршрутизация контроллера

Класс запроса контроллера определяет следующего специалиста — KYC, санкции, негативные сообщения в СМИ — или завершается кратким резюме.

from langchain_core.messages import SystemMessage

SUPERVISOR_PROMPT = """You are a financial crime case supervisor.
Route to exactly one of: kyc_agent, sanctions_agent, tx_monitoring_agent,
adverse_media_agent, narrative_agent, or FINISH if the case is fully investigated."""

def supervisor(state: CaseState) -> dict:
    routing_llm = llm.with_structured_output(RouteDecision)
    decision = routing_llm.invoke(
        [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    )
    return {"next": decision.next_agent}

def route_next(state: CaseState) -> str:
    return state["next"] if state["next"] != "FINISH" else END
graph.add_node("supervisor", supervisor)
for agent_name in ["kyc_agent", "sanctions_agent", "tx_monitoring_agent", "adverse_media_agent"]:
    graph.add_edge(agent_name, "supervisor")
graph.add_conditional_edges("supervisor", route_next)

Параллельное распространение

Независимые проверки могут выполняться после KYC без необходимости последовательной обработки всего:

graph.add_edge("kyc_agent", "sanctions_agent")
graph.add_edge("kyc_agent", "adverse_media_agent")
graph.add_edge("kyc_agent", "tx_monitoring_agent")

# Fan-in: all three must complete before scoring
graph.add_edge("sanctions_agent", "risk_scoring_agent")
graph.add_edge("adverse_media_agent", "risk_scoring_agent")
graph.add_edge("tx_monitoring_agent", "risk_scoring_agent")

Участие человека

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

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

def compliance_review(state: CaseState) -> dict:
    if state["risk_score"] > 0.7:
        decision = interrupt({
            "narrative": state["narrative"],
            "risk_score": state["risk_score"],
            "action_needed": "approve_or_reject_sar",
        })
        return {"requires_human_review": False, "human_decision": decision}
    return {"requires_human_review": False}

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "case-2026-00931"}}
result = app.invoke({"case_id": "case-2026-00931", ...}, config=config)

# Later, once a human reviews in your case management UI:
final = app.invoke(Command(resume={"approved": True, "reviewer": "j.smith"}), config=config)

for chunk in app.stream(
    {"case_id": "case-2026-00931", ...},
    config=config,
    stream_mode="updates",
):
    for node_name, node_output in chunk.items():
        print(f"[{node_name}] emitted: {node_output}")
        # push node_output to a websocket / case UI here

Подграфы для повторного использования

Сформируйте граф для решения задач KYC как отдельный элемент и включите его в качестве узла, чтобы другие типы дел могли использовать ту же структуру.

kyc_subgraph = StateGraph(CaseState)
kyc_subgraph.add_node("resolve_entity", resolve_entity_node)
kyc_subgraph.add_node("verify_documents", verify_documents_node)
kyc_subgraph.add_node("pep_check", pep_check_node)
kyc_subgraph.set_entry_point("resolve_entity")
kyc_subgraph.add_edge("resolve_entity", "verify_documents")
kyc_subgraph.add_edge("verify_documents", "pep_check")
kyc_subgraph.add_edge("pep_check", END)
compiled_kyc = kyc_subgraph.compile()
parent_graph = StateGraph(CaseState)
parent_graph.add_node("kyc_process", compiled_kyc)  # subgraph as a node

Наблюдаемость с помощью LangSmith

Включите отслеживание и помечайте запуски с идентификаторами дел и ролями агентов.

import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls__..."
os.environ["LANGSMITH_PROJECT"] = "financial-crime-agents"

result = app.invoke(
    {"case_id": "case-2026-00931", ...},
    config={
        "configurable": {"thread_id": "case-2026-00931"},
        "metadata": {"case_id": "case-2026-00931", "customer_tier": "high_risk"},
        "tags": ["sar_pipeline", "prod"],
    },
)

Итерации на основе оценки

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

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()
def correctness_evaluator(run, example) -> dict:
    predicted_risk = run.outputs.get("risk_score", 0)
    actual_risk = example.outputs.get("risk_score", 0)
    return {
        "key": "risk_score_accuracy",
        "score": 1.0 if abs(predicted_risk - actual_risk) < 0.15 else 0.0,
    }
results = evaluate(
    lambda inputs: app.invoke(inputs, config={"configurable": {"thread_id": "eval"}}),
    data="aml-labeled-cases-2026-h1",
    evaluators=[correctness_evaluator],
    experiment_prefix="sar-pipeline-v3",
)

from langsmith.evaluation import LangChainStringEvaluator

narrative_judge = LangChainStringEvaluator(
    "criteria",
    config={
        "criteria": {
            "cites_evidence": "Does the narrative cite specific transaction IDs and dates?",
            "regulatory_tone": "Is the language neutral and factual, avoiding speculation?",
        },
        "llm": ChatAnthropic(model="claude-sonnet-4-6", temperature=0),
    },
)
results = evaluate(
    lambda inputs: app.invoke(inputs, config=config),
    data="aml-labeled-cases-2026-h1",
    evaluators=[correctness_evaluator, narrative_judge],
    experiment_prefix="sar-narrative-v2",
)

Ограничения и обработка сбоев

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

from langgraph.pregel import RetryPolicy

graph.add_node(
    "sanctions_tools",
    ToolNode([screen_ofac, screen_un_sanctions]),
    retry=RetryPolicy(max_attempts=3, backoff_factor=2.0),
)

result = app.invoke(initial_state, config={**config, "recursion_limit": 25})

Проблемы в производственных схемах AML

Скрытые общие изменяемые глобальные переменные, автоматическое создание документов без надзора, отсутствие связи между оценками и доказательствами, а также системы надзора, которые постоянно задают один и тот же вопрос агенту. Необходимо ограничивать циклы, требовать указания источников в состоянии системы и сохранять действия по созданию документов в рамках HITL.

Типовая структура

Практическая топология: надзорщик в центре, специализированные агенты с циклами инструментов, необязательные параллельные ветви, узел объединения/описания, шлюз прерывания, затем экспорт — с использованием повторяемых подграфов там, где единицы дублируются.

             ┌─────────────┐
             │  Supervisor │◄───────────────-─┐
             └──────┬──────┘                  │
     ┌───────────────┼───────────────┐        │
     ▼               ▼               ▼        │
┌─────────┐   ┌──────────────┐  ┌───────────┐ │
│   KYC   │   │ Sanctions    │  │ TX Monitor│ │
│Subgraph │   │ Agent + Tools│  │  Agent    │ │
└────┬────┘   └──────┬───────┘  └─────┬─────┘ │
     └───────────────┴────────────────┘       │
                       ▼                      │
              ┌─────────────────┐             │
              │ Risk Scoring    │─────────────┘
              └────────┬────────┘
                       ▼
              ┌─────────────────┐        ┌──────────────┐
              │ Narrative Agent │───────►│ Human Review │
              └─────────────────┘        │ (interrupt)  │
                                         └──────┬───────┘
                                                 ▼
                                          SAR Filed / Closed

Стоимость и задержка

Распространение данных потребляет токены; кэшируются стабильные профили KYC; для маршрутизации используются более мелкие модели, а для описания событий — более крупные; обновления интерфейса передаются в реальном времени, чтобы следователи могли видеть прогресс до окончательной записи.

Системы борьбы с финансовыми преступлениями с участием нескольких агентов работают эффективно, когда состояние системы явно определено, специалисты поддаются тестированию, оркестрация действий видима, а люди сохраняют участие в принятии решений по необратимым действиям. LangGraph обеспечивает план управления; LangChain — инструменты; LangSmith — цикл обратной связи, который сохраняет точность графа по мере изменения типологий.

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

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

Факторы человека имеют большое значение. Если прерывания возникают слишком часто, исследователи игнорируют их; если слишком редко, риск остается нераскрытым. Настройте пороги с помощью анализа выборок. В отображаемом промежуточном состоянии должна подчеркиваться неопределенность, а не только уверенные выводы, чтобы эксперты знали, где искать информацию.

Устойчивость к атакам является частью надежности: результаты работы инструментов могут быть искажены или содержать вредоносный контент (например, отравленные тексты в неблагоприятных СМИ). Узлы верификации должны очищать данные и указывать на источники. Никогда не разрешайте одному только скопированному абзацу приводить к принятию решения без дополнительной проверки.

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

Эти практики превращают архитектуру справочной модели из умного диаграмматического представления в функциональную систему. Приведённые выше примеры кода являются лишь основой; именно модель управления обеспечивает доверие к агентам, занимающимся борьбой с финансовыми преступлениями.

Создание специализированных запросов без дублирования

Каждый специализированный агент должен получать только ту часть информации о состоянии дела, которая ему необходима. Загружение всей информации об расследовании в каждый запрос тратит ресурсы и приводит к утечке ненужных персональных данных в инструменты, которые их не требуют. Необходимо определить функции отображения: sanctions_view(state), kyc_view(state) и протестировать их с помощью единичных тестов. Руководители видят краткие сводки маршрутизации; авторы запросов — пакеты доказательств.

Клиенты общих инструментов должны находиться в библиотеке с механизмами повторных попыток и отслеживания показателей. Агентам не следует самостоятельно настраивать параметры времени ожидания HTTP-запросов по-разному. Необходимо централизовать учёт учетных данных поставщиков списков и механизмов защиты от сбоев, чтобы сбои в работе системы санкций влияли одинаково на всю сеть.

Цепочка происхождения данных и хранилища признаков

Признаки транзакций, используемые агентами для поиска закономерностей, должны поступать из управляемых конвейеров обработки данных, а не из импровизированных SQL-запросов внутри инструментов. Необходимо версионировать определения признаков. Когда аналитик оспаривает значение показателя, необходимо восстановить исходные данные. Графы, которые не могут объяснить значение показателя, не пройдут проверку на риски модели.

Параллельность и согласованность

Ветви, расходящиеся в разные направления, должны объединяться детерминированно. Необходимо определить функции сведения данных для списковых полей и правила «последний внесший изменения» для скалярных значений. Тестируйте одновременные обновления. Если как негативные медиаотчеты, так и санкции записывают значения в поле risk_flags, используйте функции сведения данных с аннотациями, чтобы значения объединялись, а не перезаписывались.

Интеграция с рабочим процессом человека

Системы управления делами (похожие на Jira, пользовательские CRM) должны обрабатывать прерывания. Отправляйте данные прерываний в виде задач; по их завершении вызывайте API для возобновления работы с результатом выполнения задачи. Не создавайте отдельный веб-сайт для утверждения, если сотрудники уже работают в существующей консоли — интегрируйте функции утверждения прямо там.

Тестирование с использованием «красной команды» графа

Симулируйте отравление инструментов, вставку команд в записи клиентов и несоответствия при определении сущностей. Убедитесь, что граф отклоняет запросы на сохранение при слабых доказательствах. Записывайте эти симуляции в качестве тестовых случаев. Инструменты борьбы с финансовой преступностью могут использоваться в двух целях; злоумышленники попытаются принудительно получить разрешения.

Управление затратами

Тег LangSmith (или аналогичный) использует параметры case_type, agent и model. Установлен бюджет на каждый тип дела. Подается оповещение при обнаружении аномалий. Для явно однозначных случаев согласования/отклонения предпочтительны детерминистические правила, а нейросети используются для работы с неоднозначными ситуациями. Гибридные системы превосходят чисто нейросетевые решения как по стоимости, так и по возможностям аудита.

Штат и ответственность

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

Благодаря таким подходам паттерн многоагентной системы LangGraph превращается в операционную систему для проведения расследований, а не в демонстрационный инструмент, работающий только в одном определенном случае.

От пилотного проекта к внедрению в основную деятельность

Пилотные проекты часто основаны на ограниченном количестве синтетических случаев. Для внедрения в рабочих процессах необходимо адаптировать типологии, обеспечить поддержку языков для описания ситуаций и интегрировать систему с существующими механизмами отслеживания времени выполнения задач и SLA. Планируйте поэтапное внедрение по сегментам: сначала продукты с низким уровнем риска, затем более рискованные. Сохраняйте контрольную группу случаев, рассматриваемых вручную, чтобы оценить эффективность изменений. Следует четко донести, что использование моделей не означает отсутствия человеческого контроля. Обучение сотрудников должно включать знания по интерпретации графиков, способам переопределения параметров и методам сообщения о некорректных данных инструментов. Технологии без такой человеческой составляющей работают хуже, даже если схема архитектуры безупречна.

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

Чек-лист перед запуском в производственных условиях

Схема системы проверена отделом соблюдения стандартов; инструменты защищены механизмами аутентификации и таймаутов; руководитель оценил прошлые случаи как успешные; путь HITL протестирован полностью от начала до конца; панели управления LangSmith (или аналогичных инструментов) работают в реальном времени; установлены лимиты на рекурсию; документирован механизм отключения системы; настроены уведомления о превышении бюджета; согласована система версионирования подграфов; протестированы инструменты параллельного объединения данных; требуются письменные обоснования всех действий; опубликованы руководства для сотрудников, работающих в режиме дежурства. Обнаружение финансовых преступлений с использованием нескольких агентов зависит скорее от этих простых механизмов, чем от изощренных запросов. Отметьте их выполнение, затем постепенно увеличивайте объем обрабатываемых данных, еженедельно проверяя примеры ложно положительных результатов в течение первого квартала.

Сотрудничество с следователями в качестве пользователей продукта

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

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

Публикуйте ежемесячный отчет о качестве, оценку которого помогают проводить аналитики. Совместная ответственность обеспечивает согласованность многоагентной системы с основной задачей — борьбой с финансовыми преступлениями, а не только с показателями демонстрационной версии. Если делать это регулярно, многоагентная архитектура LangGraph завоюет доверие единственным способом, который имеет значение в работе по борьбе с финансовыми преступлениями — путем выдержки реальных проверок, работы с настоящими следователями и реальными аудитами, а не просто красивым внешним видом на схемах. Сначала реализуйте простые механизмы контроля; сложные алгоритмы маршрутизации должны появиться только после того, как базовые механизмы контроля сработают надежно. Следователи сразу заметят разницу. Всегда проектируйте с учетом такой тщательной проверки. Менее этого здесь недостаточно.