Рабочий процесс ИИ

AI Workflow — это система, которая упорядоченно объединяет множество вызовов AI-моделей, использование инструментов и шаги обработки данных в автоматизированный конвейер.

Один вызов LLM может ответить на вопросы, но реальные задачи часто требуют: поиска в интернете → извлечения информации → анализа → написания отчёта → отправки электронного письма. AI Workflow заключается в связывании этих шагов в цепочку, чтобы AI мог автоматически выполнить всю задачу, а не просто ответить одним предложением.

Наглядная аналогия

Представьте конвейер:

ModeАналогияОписание
Один вызов LLMКак ремесленникДайте ему кусок железа, и он вернёт вам меч
Рабочий процесс ИИКак целый конвейерСырьё поступает, автоматически проходит через плавку → ковку → закалку → полировку → упаковку, и выходят готовые изделия

Каждый шаг может быть AI-моделью, функцией кода, внешним API или узлом проверки человеком.

Эволюция от вопросов и ответов к действиям

Three levels of AI capability Level 1: Single Q&A Q & A User input → LLM → Output Example: "Help me write a poem" Limitation: single-step only Cannot complete complex tasks → Level 2: Chain — chained calls Chain Step A → Step B → Step C Example: Translate → Summarize → Polish Limitation: fixed flow Cannot make dynamic decisions → Level 3: AI Workflow With Agent Agent Perceive → Plan → Tool calling → Reflect → Iterate in a loop Example: Research competitors and generate a report Dynamic decision-making, can complete open-ended tasks AI Workflow is the main paradigm for deploying AI applications today, and it is also the implementation foundation of AI Agents.

Почему нужен AI Workflow

Ограничения одиночных вызовов

Один вызов LLM может сделать очень мало:

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

Пять ключевых проблем, решаемых AI Workflow

Five core problems solved by AI Workflow Task decomposition Break down complex tasks into multiple small steps Each step focuses on one thing Write report → Research+ Outline + Writing + Review Improve accuracy Reduce per-step pressure Tool integration Call search, databases Code executor, API and other external capabilities Search engines, calculators Databases, email services Break through knowledge boundaries Connect to the real world Iterative reflection AI can examine itself After output, when issues are found Retry and correct Generate code → Run → Error → Fix Automatic error correction Quality is more guaranteed Parallel processing Multiple subtasks simultaneously Execute without queuing Waiting for the previous step to complete Simultaneously analyze finance technology, market — three dimensions Greatly improve efficiency Save runtime Observability The input and output of each step Can all be recorded Monitoring and debugging Logs, tracing Error localization Production-grade reliability Facilitates troubleshooting

Основные компоненты

Полный AI Workflow состоит из следующих ключевых элементов.

AI Workflow core architecture components AI Workflow Orchestration engine LLM/Model GPT-4、Claude Gemini, local models The brain of the Workflow Tools Search, code execution Databases, external APIs Interact with the outside Memory Short-term: conversation history Long-term: vector database Cross-session persistence State Task progress, intermediate results, context passing Like a relay baton Human intervention Review, feedback Correction, authorization Pause for high-risk operations Routing/Condition Decide based on conditions Which execution path to take Branching, looping, jumping Input/Output Text, images, files Structured data All components are orchestrated and coordinated by the orchestration engine (LangChain / LlamaIndex / Dify, etc.)

Подробное объяснение каждого компонента

LLM (большая языковая модель)— «мозг» Workflow, отвечает за рассуждения, генерацию и принятие решений. Часто используются: GPT-4o, Claude 3.5, Gemini 1.5, локальная Llama 3.

Инструменты— интерфейсы, позволяющие ИИ взаимодействовать с внешним миром, включая: поисковые системы (Tavily, Serper, Bing), исполнители кода (Python REPL, sandbox-среды), запросы к базам данных (SQL, vector DB), внешние API (погода, акции, email, календарь), файловые операции (чтение/запись, парсинг PDF/Excel).

Память— краткосрочная память хранит историю диалога текущей сессии (хранится в промпте); долгосрочная память обеспечивает межсессионное постоянное хранение через векторную базу данных + RAG; рабочая память поддерживает промежуточное состояние во время выполнения задачи.

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

Маршрутизатор / условие (Router)— динамически определяет следующий шаг на основе вывода предыдущего шага, обеспечивая сложное управление потоком, такое как ветвление, циклы и переходы.

Человек в цикле— приостанавливает работу в ключевых узлах для ожидания подтверждения от человека, подходит для высокорисковых операций (например, удаление данных, отправка email, финансовые операции).


Шесть распространённых паттернов Workflow

Ниже приведены шесть самых распространённых паттернов проектирования в AI Workflow, от простого к сложному, подходящих для разных сценариев.

Паттерн 1: Последовательная цепочка

Самый базовый паттерн, где шаги A → B → C выполняются линейно, а вывод предыдущего шага является входом для следующего.

Pattern 1: Sequential Chain Input User's original English article Step 1 Translate English → Chinese GPT-4o Step 2 Summarize Extract core points Within 100 characters Step 3 Polish Optimize expression Add title Output Refined Chinese summary Including title and key points Each step's output automatically becomes the input of the next step, flowing linearly with clear logic
ИзмерениеОписание
Применимые сценарииКонвейеры обработки документов, генерация контента, преобразование данных
ПреимуществаПросто, предсказуемо
НедостаткиЖёстко, нельзя динамически адаптироваться под содержимое

Паттерн 2: Условная маршрутизация

Динамически выбирает разные последующие пути на основе выходного содержимого определённого шага.

Pattern 2: Conditional Routing User question "How to use Python to sort a list?" Intent classification Router Code problem → Code generation Agent Run code to verify Return result + explanation Knowledge problem → RAG knowledge base retrieval Retrieve relevant documents Generate an answer based on the documents Real-time problem → Online search Search for the latest information Synthesize into an answer The routing node dispatches tasks to the most appropriate processing branch based on the question type
ИзмерениеОписание
Применимые сценарииИнтеллектуальная служба поддержки, многофункциональные ассистенты, классификация и обработка вопросов
ПреимуществаГибко, высокая утилизация ресурсов
НедостаткиЛогика маршрутизации требует тщательного проектирования; ошибки классификации влияют на весь процесс

Паттерн 3: Параллельное выполнение

Несколько подзадач выполняются одновременно, а результаты агрегируются в конце.

ИзмерениеОписание
Применимые сценарииМногомерный анализ, пакетная обработка, независимые подзадачи
ПреимуществаЗначительно повышается скорость
НедостаткиТребуется обработка управления параллелизмом и логики объединения результатов

Паттерн 4: Цикл ReAct (Reason + Act)

ИИ сначала рассуждает, чтобы решить, что делать, затем действует, вызывая инструменты, и продолжает рассуждать на основе результатов, циклически повторяя до завершения задачи. Это основной паттерн AI-агентов.

Pattern 4: ReAct Loop (Reason → Act → Observe → Repeat) Task Query Beijing today's weather and recommend outfits Thought (thinking) I need to first obtain Beijing real-time weather Should call the weather API Action (acting) Call weather tool get_weather( city="Beijing") Observation Return: sunny, temperature 18°C, light breeze (tool execution result) Loop continues reasoning Thought (Round 2) The weather is known, now we can directly provide outfit suggestions Final Answer Beijing is sunny today, 18°C; it is recommended to wear a light jacket Long pants + sneakers; you may bring a thin jacket as backup. (Task complete, exit loop) In each round, 'Thought → Action → Observation' continuously accumulates information until the LLM believes it can make a final answer.
ИзмерениеОписание
Применимые сценарииAI-агент, выполнение сложных задач, решение открытых проблем
ПреимуществаДинамично и гибко, может справляться с неизвестными ситуациями
НедостаткиКоличество циклов не контролируется, и нужно задать максимальное число шагов, чтобы предотвратить бесконечный цикл.

Паттерн 5: Plan & Execute

Сначала дайте LLM сформировать полный план, затем выполняйте шаг за шагом согласно плану. Отличие от ReAct — «сначала всё продумать, затем действовать».

Mode 5: Plan & Execute (plan first, then execute) Task objective Research competitors and write an analysis report Planner Planning 1 Collect competitor information 2 Extract key features 3 Compare and analyze 4 Write the report Execute 1 Search engine Call x3 Execute 2+3 LLM extraction Comparative analysis Execute 4 → Output Generate Markdown Competitive analysis report The Planner uses only one LLM call to formulate a global plan, and subsequent steps are strictly executed according to the plan.

Паттерн 6: Совместная работа нескольких агентов

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

Mode 6: Multi-Agent Collaboration (Multi-Agent) Coordinator Agent Receives user tasks, assigns them to specialized Agents, and aggregates results. Researcher Agent Tools: search engine, web scraping Responsibilities: collect information, organize facts Output: structured information summary ↑ Report to coordinator Analyst Agent Tools: calculator, Python Responsibilities: data analysis, chart generation Output: analysis conclusions ↑ Report to coordinator Writer Agent Tools: document templates Responsibilities: writing and polishing reports Output: final document ↑ Report to coordinator Reviewer Agent Tools: fact-checking tools Responsibilities: verify accuracy Output: review comments ↑ Report to coordinator Final result → User Each Agent specializes in one skill; the coordinator manages the overall situation. Division of labor and cooperation is far more efficient than a single Agent.
ИзмерениеОписание
Применимые сценарииСложная разработка ПО, помощь в исследованиях, корпоративная автоматизация
ПреимуществаУзкая специализация, более высокое качество, легко масштабировать
НедостаткиВысокая сложность системы; коммуникация между агентами требует тщательного проектирования

Сравнение основных фреймворков и инструментов

Ниже приведено комплексное сравнение самых распространённых фреймворков AI Workflow, чтобы помочь вам выбрать с учётом вашей ситуации.

AI Workflow mainstream framework comprehensive comparison Framework Positioning Core features Suitable for Learning curve Open source LangChain Python/JS Code-first framework Most complete ecosystem Full coverage of Chain, Agent, RAG Largest number of tool integrations (200+) LangSmith observability platform Developers with Python basics Need extensive custom integration Medium √ LangGraph Created by LangChain Graph-based Agent framework Stateful workflows Use "graphs" to define complex workflows Built-in state management, resumable checkpoints Supports Human-in-the-Loop Scenarios requiring complex process control Production-grade Agent applications Difficult √ LlamaIndex Python RAG-specialized framework Strongest data processing Optimal data ingestion, indexing, and retrieval Supports 80+ data source connectors Built-in advanced RAG strategies Knowledge base, document Q&A scenarios Requires integrating large amounts of unstructured data Medium √ Dify Open-source LLMOps Visual low-code Full-stack AI platform Drag-and-drop workflow building, no coding required Built-in application management, API publishing Supports self-hosting and cloud Users with non-technical backgrounds From rapid prototyping to production deployment Lowest √ n8n Automation platform General-purpose workflow automation Includes AI nodes 400+ service integrations, visual orchestration Hybrid of AI nodes + traditional automation Self-hostable, data stays in-country Requires connecting to various SaaS systems Business process automation scenarios Relatively low √ CrewAI Python Multi-Agent collaboration Role-playing framework Defines Agents by 'role' and 'task' Built-in delegation and supervision mechanisms Minimalist API, quick to learn Multi-Agent collaboration scenarios Beginners with basic Python knowledge Relatively low √ Recommendations for beginners: with coding background → LangChain / CrewAI; non-technical background → Dify / n8n

Дерево решений для выбора фреймворка

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

你的情况是什么?
│
├─── 没有编程基础,想用可视化工具搭建
│    ├─── 主要是 AI 应用(问答、生成)→ Dify(首选)
│    └─── 需要连接 Slack/邮件等 SaaS 系统 → n8n
│
├─── 有 Python 基础,代码优先
│    ├─── 做知识库 / RAG 系统 → LlamaIndex
│    ├─── 做多 Agent 协作,想快速上手 → CrewAI
│    ├─── 需要复杂有状态流程控制 → LangGraph
│    └─── 通用场景,想要最大生态 → LangChain
│
└─── 已有明确场景,生产级要求
     ├─── 高并发、精细控制 → LangGraph + LangSmith
     └─── 企业部署、私有化 → Dify 自托管

Быстрый старт: примеры кода на Python

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

Последовательная цепочка LangChain

Самый базовый шаблон Workflow, соединяющий несколько шагов вызова LLM с помощью оператора конвейера |.

Установите пакеты зависимостей:

pip install langchain langchain-openai

Пример

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", api_key="your-api-key")

# ─── Определить три шага ────────────────────────────────────────────
# Шаг 1: Перевести статью на китайский
translate_prompt = ChatPromptTemplate.from_template(
    Переведите следующую английскую статью на китайский, сохраняя исходный смысл:\n\n{article}"
)

# Шаг 2: Извлечь сводку
summarize_prompt = ChatPromptTemplate.from_template(
    Пожалуйста, выделите из следующей статьи 3 ключевых пункта, по одному предложению каждый:\n\n{translated}"
)

# Шаг 3: Сгенерировать заголовок
title_prompt = ChatPromptTemplate.from_template(
    На основе следующей сводки сгенерируйте привлекательный китайский заголовок (не более 15 символов):\n\n{summary}"
)

parser = StrOutputParser()

# ─── Построение цепочки-конвейера с помощью оператора | ─────────────────────────────────
chain = (
    {"translated": translate_prompt | llm | parser}
    | {"summary": summarize_prompt | llm | parser,
       "translated": lambda x: x["translated"]}
    | title_prompt | llm | parser
)

# ─── Запуск ────────────────────────────────────────────────────
article = """
Искусственный интеллект меняет то, как мы работаем и живём.
От автоматизации повторяющихся задач до помощи в творческой работе,
Инструменты ИИ становятся незаменимыми в современных рабочих процессах...
"""


result = chain.invoke({"article": article})
print(result)
AI 正在重塑现代工作流:从自动化到创意辅助

Агент, вызывающий инструменты (режим ReAct)

ReAct — это основной режим AI-агента, позволяющий ИИ циклически переходить между размышлением и действием до завершения задачи.

Пример

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain import hub
import requests, datetime

# ─── Определить инструменты ────────────────────────────────────────────────
@tool
def get_weather(city: str) -> str:
    """Получить текущую информацию о погоде для указанного города"""
    # Замените на реальный API погоды в продакшене
    mock_data = {
        "Пекин": "Солнечно, 22°C, легкий ветер",
        "Шанхай": "Облачно, 26°C, влажность 75%",
        "Гуанчжоу": "Небольшой дождь, 30°C, возьмите зонт",
    }
    return mock_data.get(city, f"Нет данных о погоде для {city}")

@tool
def search_web(query: str) -> str:
    """Поиск информации в интернете и возврат краткого содержания релевантного контента"""
    # Интегрировать Tavily / Serper API в продакшене
    return f"Результаты поиска для '{query}': Это имитация результата поиска..."

@tool
def calculate(expression: str) -> str:
    """Вычислить математическое выражение, например, '2 + 3 * 4'"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Ошибка вычисления: {e}"

@tool
def get_date() -> str:
    """Получить сегодняшнюю дату"""
    return datetime.date.today().strftime("%Y-%m-%d")

# ─── Создание агента ───────────────────────────────────────────────
tools = [get_weather, search_web, calculate, get_date]
llm   = ChatOpenAI(model="gpt-4o", temperature=0)

# Использовать стандартный ReAct-промпт из LangChain Hub
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,      # Печатать каждый шаг рассуждения для удобства отладки
    max_iterations=8,  # Предотвратить бесконечные циклы
    handle_parsing_errors=True
)

# ─── Запуск ────────────────────────────────────────────────────
result = agent_executor.invoke({
    "input": "Какая сегодня дата? Какая погода в Пекине? Если вы пройдёте 5km на улице, вы сожжёте около 300 калорий, "
             "Бег на ту же дистанцию сжигает примерно в 1.6 раза больше, чем ходьба. Пожалуйста, рассчитайте количество калорий, сожжённых при беге."
})

print(result["output"])
# Агент автоматически решит: сначала вызвать get_date → get_weather("Beijing") → calculate("300*1.6")
# Наконец, объедините всю информацию, чтобы дать полный ответ

Рабочий процесс с состоянием в LangGraph

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

Пример

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator

# ─── Определение структуры состояния (данные, передаваемые между узлами) ────────────────────────
class ResearchState(TypedDict):
    topic: str                             # Тема исследования
    research_notes: str                    # Заметки по исследованию
    draft: str                             # Черновик
    review_feedback: str                   # Отзыв о проверке
    final_report: str                      # Итоговый отчёт
    revision_count: Annotated[int, operator.add]   # Количество правок (накопительно)

llm = ChatOpenAI(model="gpt-4o")

# ─── Определение функций узлов ─────────────────────────────────────────────
def research_node(state: ResearchState) -> dict:
    """Узел 1: Этап исследования"""
    response = llm.invoke(
        f"Пожалуйста, проведите краткое исследование по следующей теме и перечислите 5 ключевых пунктов: {state['topic']}"
    )
    return {"research_notes": response.content}

def write_node(state: ResearchState) -> dict:
    """Узел 2: Написание черновика"""
    prompt = f"""
Тема: {state['topic']}
Заметки по исследованию: {state['research_notes']}
{'Отзыв предыдущей проверки: ' + state.get('review_feedback', '') if state.get('review_feedback') else ''}

Пожалуйста, напишите черновик аналитического отчёта объёмом 300 слов на основе вышеуказанного содержания.
    """

    response = llm.invoke(prompt)
    return {"draft": response.content, "revision_count": 1}

def review_node(state: ResearchState) -> dict:
    """Узел 3: Проверка черновика"""
    response = llm.invoke(
        f"Проверьте следующий отчёт. Если качество соответствует стандарту, ответьте 'APPROVED'; в противном случае дайте конкретные предложения по доработке:\n\n{state['draft']}"
    )
    return {"review_feedback": response.content}

def finalize_node(state: ResearchState) -> dict:
    """Узел 4: Завершение"""
    return {"final_report": state["draft"]}

# ─── Функция маршрутизации: определить, какой путь выбрать после проверки ─────────────────────────────
def should_revise(state: ResearchState) -> str:
    if "APPROVED" in state["review_feedback"]:
        return "finalize"           # → Финализировать
    elif state["revision_count"] >= 3:
        return "finalize"           # → Превышено 3 ревизии, принудительное завершение
    else:
        return "revise"             # → Вернуться к узлу написания для доработки

# ─── Построение графа рабочего процесса ─────────────────────────────────────────────
workflow = StateGraph(ResearchState)

# Добавить узлы
workflow.add_node("research",  research_node)
workflow.add_node("write",     write_node)
workflow.add_node("review",    review_node)
workflow.add_node("finalize",  finalize_node)

# Задать точку входа
workflow.set_entry_point("research")

# Добавить рёбра (определить логику переходов)
workflow.add_edge("research", "write")       # Исследование → Написание
workflow.add_edge("write",    "review")      # Написание → Проверка

# Условное ребро: выбор пути на основе результата проверки
workflow.add_conditional_edges(
    "review",
    should_revise,
    {
        "revise":   "write",    # Требуется доработка → Возврат к написанию
        "finalize": "finalize"  # Одобрено → Финализация
    }
)

workflow.add_edge("finalize", END)

# Скомпилировать и запустить
app = workflow.compile()

result = app.invoke({"topic": "Влияние генеративного ИИ на индустрию разработки программного обеспечения", "revision_count": 0})
print(result["final_report"])

CrewAI Многоагентное сотрудничество

CrewAI позволяет нескольким агентам сотрудничать как команда, определяя роли и задачи.

Пример

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# ─── Определить агента (роль) ────────────────────────────────────────
researcher = Agent(
    role=Исследователь рынка,
    goal=Собирать и систематизировать всеобъемлющую, точную рыночную информацию по целевой теме.,
    backstory=Вы — опытный аналитик рынка, умеющий извлекать ключевые выводы из огромных объёмов информации.,
    llm=llm,
    verbose=True
)

analyst = Agent(
    role=Аналитик данных,
    goal=На основе информации, предоставленной исследователем, проведите углубленный анализ и сформулируйте ценные выводы.,
    backstory=Вы отлично используете данные для коммуникации и умеете выявлять тенденции и возможности, скрытые в информации.,
    llm=llm,
    verbose=True
)

writer = Agent(
    role=Эксперт по написанию отчетов,
    goal=Оформите аналитические выводы в виде четкого, профессионального и убедительного отчета.,
    backstory=У вас обширный опыт в написании деловых текстов, и вы умеете объяснять сложный анализ простым языком.,
    llm=llm,
    verbose=True
)

# ─── Определение задачи ────────────────────────────────────────
research_task = Task(
    description=Исследуйте текущее состояние рынка автомобилей на новых источниках энергии в Китае, включая основные бренды, долю рынка и тенденции роста.,
    expected_output=Отчёт о маркетинговом исследовании, содержащий 5 ключевых точек данных, включая цифры и конкретные факты.,
    agent=researcher
)

analysis_task = Task(
    description=На основе исследовательского отчёта проанализируйте возможности и риски на ближайшие 3 года и предоставьте рекомендации по инвестиционному рейтингу.,
    expected_output=Таблица SWOT-анализа + инвестиционный рейтинг (настоятельно рекомендуется/рекомендуется/нейтрально/осторожно) + обоснование,
    agent=analyst,
    context=[research_task]    Результат зависимой исследовательской задачи
)

writing_task = Task(
    description=Объедините исследование и анализ в профессиональный инвестиционный брифинг объёмом 500 слов с чётким форматированием.,
    expected_output=Брифинг, состоящий из четырёх разделов: резюме для руководства, состояние рынка, возможности и риски, и инвестиционные рекомендации.,
    agent=writer,
    context=[research_task, analysis_task]
)

# ─── Соберите команду и выполните ───────────────────────────────────────────
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,    # Последовательное выполнение (также можно изменить на иерархическое)
    verbose=True
)

result = crew.kickoff()
print(result)

Человек в цикле (Узел ручной проверки)

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

Пример

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class EmailState(TypedDict):
    recipient: str
    content: str
    approved: bool

def draft_email(state: EmailState) -> dict:
    Составить электронное письмо.
    content = fУважаемый {state['recipient']},\n\nЭто содержимое письма, составленное ИИ...\n\nС уважением
    return {"content": content}

def send_email(state: EmailState) -> dict:
    Отправить письмо (операция с высоким риском, требует ручной проверки перед выполнением)
    print(fПисьмо отправлено {state['recipient']})
    return {}

# Маршрутизация: решить, отправлять ли, на основе результата ручной проверки.
def check_approval(state: EmailState) -> str:
    return "send" if state.get("approved") else END

workflow = StateGraph(EmailState)
workflow.add_node("draft", draft_email)
workflow.add_node("send", send_email)
workflow.set_entry_point("draft")

# Пауза после завершения черновика, ожидание проверки человеком (interrupt_after)
workflow.add_conditional_edges("draft", check_approval, {"send": "send", END: END})
workflow.add_edge("send", END)

# Использовать MemorySaver для поддержки прерывания и возобновления
memory = MemorySaver()
app = workflow.compile(
    checkpointer=memory,
    interrupt_after=["draft"]      # Пауза после узла draft
)

config = {"configurable": {"thread_id": "email-001"}}

# Первый запуск: пауза после выполнения до draft
state = app.invoke({"recipient": "Клиент A", "approved": False}, config)
print("Черновик создан, ожидание проверки:")
print(state["content"])

# После ручной проверки обновите статус и продолжите выполнение
user_input = input("\nОдобрить отправку? (y/n): ")
if user_input.lower() == "y":
    app.update_state(config, {"approved": True})
    app.invoke(None, config)   # Продолжить с точки останова
    print("Письмо отправлено")
else:
    print("Отправка отменена")

Промпт-инжиниринг: ключевые техники в рабочем процессе

В рабочем процессе ИИ качество промпта напрямую влияет на качество вывода каждого узла.

Пример

# Нечёткое определение роли
# system_prompt = "Ты — ИИ-ассистент, помоги мне проанализировать этот документ"

# Чёткое определение роли (рекомендуется)
system_prompt = """
Ты — профессиональный финансовый аналитик, сосредоточенный на выявлении сигналов риска в финансовых отчётах.
Твоя задача: извлечь все ключевые данные, относящиеся к обязательствам, денежному потоку и прибыльности, из следующего отчёта.
Формат вывода: JSON, содержащий risk_level (high/medium/low) и key_findings (список).
Примечание: выводите только JSON, не добавляйте никакого пояснительного текста.
"""

Пример

# Передать неструктурированный текст
# state["previous_output"] = "Результат анализа: эта компания выглядит хорошо, есть некоторые риски..."

# Передать структурированные данные (рекомендуется)
state["analysis_result"] = {
    "risk_level": "medium",
    "key_findings": ["Коэффициент долговой нагрузки 45%, средний по отрасли 38%", "Денежный поток положительный, Q3 снизился на 12% по сравнению с предыдущим кварталом"],
    "recommendation": "Удерживать с осторожностью"
}

Пример

# Принудительный вывод JSON в промпте
prompt = """
Пожалуйста, проанализируйте тональность следующего текста.
Выводите строго в следующем формате JSON, не добавляйте никакого другого содержимого:
{
  "sentiment": "positive" | "negative" | "neutral",
  "confidence": 0.0-1.0,
"reason": "краткое объяснение"
}

Текст: {text}
"""

Обработка ошибок и повторные попытки

Пример

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

@retry(
    stop=stop_after_attempt(3),                         # Повторять до 3 раз
    wait=wait_exponential(multiplier=1, min=2, max=10)  # Ожидание экспоненциальной задержки
)
def call_llm_with_retry(prompt: str) -> str:
    try:
        response = llm.invoke(prompt)
        return response.content
    except Exception as e:
        logging.error(f"Ошибка вызова LLM: {e}")
        raise

def safe_parse_json(text: str) -> dict:
    """Безопасно разобрать JSON-вывод от LLM"""
    import json, re
    # Извлечь часть JSON (LLM иногда добавляет дополнительный пояснительный текст)
    json_match = re.search(r'\{.*\}', text, re.DOTALL)
    if json_match:
        try:
            return json.loads(json_match.group())
        except json.JSONDecodeError:
            pass
    return {"error": Ошибка разбора, "raw": text}

Контроль затрат

Пример

# ─── Выбор моделей в зависимости от важности задачи ────────────────────────────────
def get_llm(task_type: str):
    if task_type == "classification":   # Для задач классификации достаточно лёгкой модели
        return ChatOpenAI(model="gpt-4o-mini")
    elif task_type == "generation":     # Для задач генерации используйте среднюю модель
        return ChatOpenAI(model="gpt-4o-mini")
    elif task_type == "reasoning":      # Для сложных рассуждений используйте мощную модель
        return ChatOpenAI(model="gpt-4o")

# ─── Кэширование повторных запросов ──────────────────────────────────────────
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache
set_llm_cache(InMemoryCache())   # Возвращать кэшированный результат напрямую для идентичного ввода; без дублирования оплаты

# ─── Разбить текст на части, чтобы избежать чрезмерной длины ──────────────────────────────────────
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000,       # 2000 символов на фрагмент
    chunk_overlap=200      # Перекрытие 200 символов для сохранения связности контекста
)
chunks = splitter.split_text(long_document)

Типичные сценарии применения

AI Workflow широко применяется во многих областях. Ниже приведены шесть наиболее показательных сценариев.

AI Workflow Typical Application Scenarios AI Programming Assistant User describes requirements → Understand requirements → Generate code → Run automatically → Detect errors → Auto-fix → Generate documentation and test cases Representative: GitHub Copilot Workspace Claude Code、Cursor Automated Research Reports Receive research topic → Break down into sub-problems → Gather information in parallel → Extract Key information → Comprehensive analysis → Generate a complete report with citations Representative: Perplexity Deep Research OpenAI Deep Research Intelligent Customer Service System Identify user intent → Query order database → Retrieve policy documents → Generate personalized reply → Complex issues escalated to human agents Representative: AI customer service of major e-commerce platforms Bank Intelligent Customer Service Data Analysis Automation Upload Excel/CSV → Understand data structure → Automatically select analysis method → Execute Python → Generate charts + natural language interpretation Representative: ChatGPT Advanced Data Analysis Content Marketing Pipeline Input topic and audience → Research hot topics → Generate outline → Write full text → Optimize SEO → Generate image descriptions → Multi-platform adaptation Examples: Various types of AI writing and marketing automation tools Medical assisted diagnosis Patient describes symptoms → Structured Medical history → Retrieve medical literature → Analyze imaging reports → Generate Reference suggestions → Doctor confirmation Note: High-risk scenarios must retain human confirmation Human-in-the-Loop

Лучшие практики и распространённые ошибки

Распространённые ошибки и решения

Ниже перечислены наиболее распространённые проблемы, с которыми сталкиваются новички при использовании AI Workflow.

ОшибкаСимптомРешение
Каскад галлюцинацийПредыдущий вывод ИИ неверен, передаётся как факт следующему шагу, усиливая ошибку.Добавьте шаги проверки в ключевых узлах; используйте инструменты для числовой информации.
Бесконечный циклReAct Agent многократно вызывает инструменты, не зная, когда остановиться.Установите max_iterations; задайте LLM явные условия завершения.
Взрыв контекстаПо мере увеличения числа шагов текст, передаваемый LLM, становится длиннее и превышает окно.Передавайте только необходимые поля на каждом шаге; сжимайте историческую информацию с помощью резюме.
Злоупотребление инструментамиАгент может отвечать напрямую, но продолжает вызывать инструменты.Оптимизируйте описания инструментов; чётко указывайте LLM, когда инструменты не нужны.
Ошибка парсинга JSONФормат вывода LLM нестабилен, что приводит к сбою программы.Добавьте safe_parse_json; используйте LangChain OutputParser.
Превышение затратИспользование токенов не оценивается, и месячный счёт превышает ожидания.Сначала протестируйте с gpt-4o-mini; используйте LangSmith для мониторинга использования.
Конфликт параллелизмаНесколько агентов, одновременно записывающих в одно и то же состояние, приводят к повреждению данных.Используйте встроенное управление состоянием LangGraph; избегайте совместного использования изменяемого состояния.

Резюме и путь обучения

Повторение базовых знаний

AI Workflow Core Knowledge Points Core Elements LLM, tools memory, state routing, human Intervention: six major components Form a complete system Six major patterns Sequential chain Conditional routing Parallel execution ReAct loop Plan&Execute Multi-Agent Mainstream frameworks LangChain LangGraph LlamaIndex CrewAI Dify / n8n Select as needed Key code Chain: chained invocation Tool Definition StateGraph state graph Agent + Executor Human-in-Loop interrupt_after Key pitfalls Hallucination cascade Infinite Loop Context explosion. JSON parsing failed Cost overrun Prevention is better than cure application scenarios AI coding assistant Research report Smart customer service Data analysis Content production Medical assistance

Рекомендуемый путь обучения

Этап 1: Понимание основ (1 неделя)

  1. Понимать вызовы LLM API, уметь выполнять запросы с помощью OpenAI SDK
  2. Пройти пример последовательной цепочки LangChain из этой статьи
  3. Понимать основные принципы промпт-инжиниринга

Этап 2: Инструменты и агенты (2 недели)

  1. Научиться определять инструменты с помощью декоратора @tool
  2. Пройти пример ReAct Agent, изучить подробные логи, чтобы понять логику цикла
  3. Использовать Dify или n8n для создания визуального прототипа рабочего процесса

Этап 3: Сложный рабочий процесс (3 недели)

  1. Изучить графы состояний LangGraph, реализовать рабочий процесс с условным ветвлением
  2. Используйте CrewAI для реализации задачи совместной работы двух агентов
  3. Интегрируйте реальные инструменты (Tavily Search, Python REPL)
  4. Используйте LangSmith для наблюдения и отладки вашего рабочего процесса

Этап 4: Готовность к продакшену (постоянно)

  1. Реализуйте повторные попытки при ошибках и разбор структурированного вывода
  2. Мониторинг и оптимизация затрат (многоуровневое использование моделей)
  3. Разверните в продакшене, настройте мониторинг и оповещения

Справочные ресурсы

Другие расширения