AI Workflow

AI Workflow is a system that orderly combines multiple AI model calls, tool usage, and data processing steps into an automated pipeline.

A single LLM call can answer questions, but real-world tasks often require: searching the web → extracting information → analyzing → writing a report → sending an email. AI Workflow is about chaining these steps together so AI can automatically complete the entire task, rather than just answering a single sentence.

An Intuitive Analogy

Imagine an assembly line:

Mode	Analogy	Description
Single LLM call	Like a craftsman	Give him a piece of iron, and he returns you a sword
AI Workflow	Like an entire assembly line	Raw materials go in, automatically pass through smelting → forging → quenching → polishing → packaging, and finished products come out

Each step can be an AI model, a code function, an external API, or a human review node.

Evolution from Q&A to Action

Why AI Workflow is Needed

Limitations of Single Calls

A single LLM call can accomplish very little:

Limited context window: cannot read a whole book at once
Cannot access real-time information: training data has a cutoff date
Cannot perform actions: cannot actually send emails, write and run code
Cannot self-validate: cannot realize and correct errors after generating them
Complex tasks are error-prone: doing too much in one step leads to quality degradation

Five Core Problems Solved by AI Workflow

Core Components

A complete AI Workflow consists of the following core elements.

Detailed Explanation of Each Component

LLM (Large Language Model)- The "brain" of the Workflow, responsible for reasoning, generation, and decision-making. Commonly used: GPT-4o, Claude 3.5, Gemini 1.5, local Llama 3.

Tools- Interfaces that allow AI to interact with the external world, including: search engines (Tavily, Serper, Bing), code executors (Python REPL, sandbox environments), database queries (SQL, vector DB), external APIs (weather, stocks, email, calendar), file operations (read/write, parse PDF/Excel).

Memory- Short-term memory stores the conversation history of the current session (stored in the prompt); long-term memory achieves cross-session persistent storage via vector database + RAG; working memory maintains intermediate state during task execution.

State- The information carrier passed between steps in a task, like a relay baton; each step can read the previous step's result and write new results.

Router / Condition (Router)- Dynamically determines the next step based on the previous step's output, enabling complex flow control such as branching, looping, and jumping.

Human in the Loop- Pauses at key nodes to wait for human confirmation, suitable for high-risk operations (e.g., deleting data, sending emails, financial operations).

Six Common Workflow Patterns

The following are the six most common design patterns in AI Workflow, from simple to complex, suitable for different scenarios.

Pattern 1: Sequential Chain

The most basic pattern, where steps A → B → C execute linearly, and the output of the previous step is the input of the next.

Dimension	Description
Applicable scenarios	Document processing pipelines, content generation, data transformation
Advantages	Simple, predictable
Disadvantages	Rigid, cannot adjust dynamically based on content

Pattern 2: Conditional Routing

Dynamically select different subsequent paths based on the output content of a certain step.

Dimension	Description
Applicable scenarios	Intelligent customer service, multi-functional assistants, question classification and processing
Advantages	Flexible, high resource utilization
Disadvantages	Routing logic requires careful design; classification errors affect the entire process

Pattern 3: Parallel Execution

Multiple subtasks run simultaneously, and results are aggregated at the end.

Dimension	Description
Applicable scenarios	Multi-dimensional analysis, batch processing, independent subtasks
Advantages	Significantly improved speed
Disadvantages	Requires handling concurrency control and result merging logic

Pattern 4: ReAct Loop (Reason + Act)

The AI first reasons to decide what to do, then acts to call tools, and continues reasoning based on the results, looping until the task is complete. This is the core pattern of AI Agents.

Dimension	Description
Applicable scenarios	AI Agent, complex task execution, open-ended problem solving
Advantages	Dynamic and flexible, can handle unknown situations
Disadvantages	The number of loops is not controllable, and a maximum step count needs to be set to prevent an infinite loop.

Pattern 5: Plan & Execute

First let the LLM form a complete plan, then execute step by step according to the plan. The difference from ReAct is 'think clearly first, then act'.

Pattern 6: Multi-Agent Collaboration

Multiple specialized Agents work collaboratively, and each Agent has its own role and toolset.

Dimension	Description
Applicable scenarios	Complex software development, research assistance, enterprise automation
Advantages	Dedicated specialization, higher quality, easy to scale
Disadvantages	High system complexity; inter-Agent communication needs careful design

Comparison of Mainstream Frameworks and Tools

The following is a comprehensive comparison of the most mainstream AI Workflow frameworks to help you choose based on your own situation.

Framework Selection Decision Tree

Based on your specific situation, follow the decision tree below to choose the appropriate framework:

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

Quick Start: Python Code Examples

The following examples progress from the simplest sequential chain to complex multi-Agent collaboration, demonstrating how to implement AI Workflows step by step.

LangChain Sequential Chain

The most basic Workflow pattern, chaining multiple LLM call steps with the pipe operator |.

Install dependency packages:

pip install langchain langchain-openai

Example

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")

# ─── Define three steps ────────────────────────────────────────────
# Step 1: Translate the article into Chinese
translate_prompt = ChatPromptTemplate.from_template(
Translate the following English article into Chinese, preserving the original meaning:\n\n{article}"
)

# Step 2: Extract summary
summarize_prompt = ChatPromptTemplate.from_template(
Please distill the following article into 3 key points, one sentence each:\n\n{translated}"
)

# Step 3: Generate title
title_prompt = ChatPromptTemplate.from_template(
Based on the following summary, generate an attractive Chinese title (within 15 characters):\n\n{summary}"
)

parser = StrOutputParser()

# ─── Chain with | operator into a pipeline ─────────────────────────────────
chain = (
{"translated": translate_prompt | llm | parser}
| {"summary": summarize_prompt | llm | parser,
"translated": lambda x: x["translated"]}
| title_prompt | llm | parser
)

# ─── Run ────────────────────────────────────────────────────
article = """
Artificial intelligence is transforming how we work and live.
From automating repetitive tasks to assisting in creative work,
AI tools are becoming indispensable in modern workflows...
"""

result = chain.invoke({"article": article})
print(result)

AI 正在重塑现代工作流：从自动化到创意辅助

Tool-Calling Agent (ReAct Mode)

ReAct is the core mode of an AI Agent, allowing the AI to cycle between thinking and acting until the task is completed.

Example

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

# ─── Define tools ────────────────────────────────────────────────
@tool
def get_weather(city: str) -> str:
"""Get the current weather information for a specified city"""
# Replace with a real weather API in production
mock_data = {
"Beijing": "Sunny, 22°C, light breeze",
"Shanghai": "Cloudy, 26°C, humidity 75%",
"Guangzhou": "Light rain, 30°C, bring an umbrella",
}
return mock_data.get(city, f"No weather data available for {city}")

@tool
def search_web(query: str) -> str:
"""Search for information on the web and return a summary of relevant content"""
# Integrate Tavily / Serper API in production
return f"Search results for '{query}': This is a simulated search result..."

@tool
def calculate(expression: str) -> str:
"""Calculate a mathematical expression, e.g., '2 + 3 * 4'"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"Calculation error: {e}"

@tool
def get_date() -> str:
"""Get today's date"""
return datetime.date.today().strftime("%Y-%m-%d")

# ─── Create Agent ───────────────────────────────────────────────
tools = [get_weather, search_web, calculate, get_date]
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Use the standard ReAct prompt from LangChain Hub
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # Print each reasoning step for debugging convenience
max_iterations=8, # Prevent infinite loops
handle_parsing_errors=True
)

# ─── Run ────────────────────────────────────────────────────
result = agent_executor.invoke({
"input": "What's today's date? How's the weather in Beijing? If you walk 5km outdoors, you burn about 300 calories, "
"Running the same distance burns about 1.6 times that of walking. Please calculate the calories burned by running."
})

print(result["output"])
# The agent will automatically decide: first call get_date → get_weather("Beijing") → calculate("300*1.6")
# Finally, combine all the information to give a complete answer

LangGraph Stateful Workflow

LangGraph defines complex workflows using graphs, where each node is a function and edges define transition logic.

Example

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

# ─── Define state structure (data passed between nodes) ────────────────────────
class ResearchState(TypedDict):
topic: str # Research topic
research_notes: str # Research notes
draft: str # Draft
review_feedback: str # Review feedback
final_report: str # Final report
revision_count: Annotated[int, operator.add] # Number of revisions (cumulative)

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

# ─── Define node functions ─────────────────────────────────────────────
def research_node(state: ResearchState) -> dict:
"""Node 1: Research phase"""
response = llm.invoke(
f"Please conduct a brief research on the following topic and list 5 key points: {state['topic']}"
)
return {"research_notes": response.content}

def write_node(state: ResearchState) -> dict:
"""Node 2: Write draft"""
prompt = f"""
Topic: {state['topic']}
Research notes: {state['research_notes']}
{'Previous review feedback: ' + state.get('review_feedback', '') if state.get('review_feedback') else ''}

Please write a 300-word draft of an analysis report based on the above content.
"""
response = llm.invoke(prompt)
return {"draft": response.content, "revision_count": 1}

def review_node(state: ResearchState) -> dict:
"""Node 3: Review draft"""
response = llm.invoke(
f"Review the following report. If quality meets the standard, reply 'APPROVED'; otherwise, provide specific revision suggestions:\n\n{state['draft']}"
)
return {"review_feedback": response.content}

def finalize_node(state: ResearchState) -> dict:
"""Node 4: Finalize"""
return {"final_report": state["draft"]}

# ─── Routing function: determine which path to take after review ─────────────────────────────
def should_revise(state: ResearchState) -> str:
if "APPROVED" in state["review_feedback"]:
return "finalize" # → Finalize
elif state["revision_count"] >= 3:
return "finalize" # → Exceeds 3 revisions, force termination
else:
return "revise" # → Return to writing node for revision

# ─── Build workflow graph ─────────────────────────────────────────────
workflow = StateGraph(ResearchState)

# Add nodes
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)

# Set entry point
workflow.set_entry_point("research")

# Add edges (define transition logic)
workflow.add_edge("research", "write") # Research → Writing
workflow.add_edge("write", "review") # Writing → Review

# Conditional edge: choose path based on review result
workflow.add_conditional_edges(
"review",
should_revise,
{
"revise": "write", # Needs revision → Return to writing
"finalize": "finalize" # Approved → Finalize
}
)

workflow.add_edge("finalize", END)

# Compile and run
app = workflow.compile()

result = app.invoke({"topic": "The impact of generative AI on the software development industry", "revision_count": 0})
print(result["final_report"])

CrewAI Multi-Agent Collaboration

CrewAI enables multiple Agents to collaborate like a team by defining roles and tasks.

Example

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

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

# ─── Define Agent (Role) ────────────────────────────────────────
researcher = Agent(
role=Market researcher,
goal=Collect and organize comprehensive, accurate market information on the target topic.,
backstory=You are an experienced market analyst skilled at extracting key insights from vast amounts of information.,
llm=llm,
verbose=True
)

analyst = Agent(
role=Data Analyst,
goal=Based on the information provided by the researcher, conduct in-depth analysis and draw valuable conclusions.,
backstory=You excel at using data to communicate, and can uncover trends and opportunities hidden within information.,
llm=llm,
verbose=True
)

writer = Agent(
role=Report writing expert,
goal=Write the analytical conclusions into a clear, professional, and persuasive report.,
backstory=You have extensive business writing experience and can make complex analysis easy to understand.,
llm=llm,
verbose=True
)

# ─── Define Task ────────────────────────────────────────
research_task = Task(
description=Research the current status of China's new energy vehicle market, including major brands, market share, and growth trends.,
expected_output=A market research report containing 5 key data points, including figures and specific facts.,
agent=researcher
)

analysis_task = Task(
description=Based on the research report, analyze the opportunities and risks over the next 3 years and provide investment rating recommendations.,
expected_output=SWOT analysis table + investment rating (strongly recommend/recommend/neutral/cautious) + rationale,
agent=analyst,
context=[research_task] Dependency research task output
)

writing_task = Task(
description=Integrate the research and analysis into a 500-word professional investment briefing with clear formatting.,
expected_output=A briefing comprising four sections: executive summary, market status, opportunities and risks, and investment recommendations.,
agent=writer,
context=[research_task, analysis_task]
)

# ─── Build the team and execute ───────────────────────────────────────────
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential, # Sequential execution (can also be changed to hierarchical)
verbose=True
)

result = crew.kickoff()
print(result)

Human-in-the-Loop (Manual Review Node)

LangGraph supports pausing at critical steps to wait for human confirmation before continuing execution.

Example

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:
Draft an email.
content = fDear {state['recipient']},\n\nThis is the email content drafted by AI...\n\nSincerely
return {"content": content}

def send_email(state: EmailState) -> dict:
Send email (high-risk operation, requires manual review before execution)
print(fEmail has been sent to {state['recipient']})
return {}

# Routing: Decide whether to send based on the manual review result.
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")

# Pause after draft is complete, wait for human review (interrupt_after)
workflow.add_conditional_edges("draft", check_approval, {"send": "send", END: END})
workflow.add_edge("send", END)

# Use MemorySaver to support interruption and resumption
memory = MemorySaver()
app = workflow.compile(
checkpointer=memory,
interrupt_after=["draft"] # Pause after the draft node
)

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

# First run: pause after executing to draft
state = app.invoke({"recipient": "Customer A", "approved": False}, config)
print("Draft generated, waiting for review:")
print(state["content"])

# After manual review, update status and continue execution
user_input = input("\nApprove sending? (y/n): ")
if user_input.lower() == "y":
app.update_state(config, {"approved": True})
app.invoke(None, config) # Continue from breakpoint
print("Email sent")
else:
print("Sending canceled")

Prompt Engineering: Key Techniques in Workflow

In AI Workflow, the quality of the Prompt directly affects the output quality of each node.

Example

# Vague role definition
# system_prompt = "You are an AI assistant, help me analyze this document"

# Clear role definition (recommended)
system_prompt = """
You are a professional financial analyst, focusing on identifying risk signals in financial reports.
Your task: Extract all key data points related to liabilities, cash flow, and profitability from the following report.
Output format: JSON, containing risk_level (high/medium/low) and key_findings (list).
Note: Only output JSON, do not add any explanatory text.
"""

Example

# Pass unstructured text
# state["previous_output"] = "Analysis result: This company looks good, there are some risks..."

# Pass structured data (recommended)
state["analysis_result"] = {
"risk_level": "medium",
"key_findings": ["Debt ratio 45%, industry average 38%", "Cash flow is positive, Q3 down 12% quarter-on-quarter"],
"recommendation": "Hold with caution"
}

Example

# Force JSON output in the prompt
prompt = """
Please analyze the sentiment of the following text.
Strictly output in the following JSON format, do not add any other content:
{
"sentiment": "positive" | "negative" | "neutral",
"confidence": 0.0-1.0,
"reason": "brief explanation"
}

Text: {text}
"""

Error Handling and Retry

Example

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

@retry(
stop=stop_after_attempt(3), # Retry up to 3 times
wait=wait_exponential(multiplier=1, min=2, max=10) # Exponential backoff wait
)
def call_llm_with_retry(prompt: str) -> str:
try:
response = llm.invoke(prompt)
return response.content
except Exception as e:
logging.error(f"LLM call failed: {e}")
raise

def safe_parse_json(text: str) -> dict:
"""Safely parse the JSON output from LLM"""
import json, re
# Extract the JSON part (LLM sometimes adds extra explanatory text)
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
return {"error": Parsing failed, "raw": text}

Cost Control

Example

# ─── Select models based on task importance ────────────────────────────────
def get_llm(task_type: str):
if task_type == "classification": # For classification tasks, a lightweight model is sufficient
return ChatOpenAI(model="gpt-4o-mini")
elif task_type == "generation": # For generation tasks, use a medium model
return ChatOpenAI(model="gpt-4o-mini")
elif task_type == "reasoning": # For complex reasoning, use a powerful model
return ChatOpenAI(model="gpt-4o")

# ─── Cache repeated requests ──────────────────────────────────────────
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache
set_llm_cache(InMemoryCache()) # Return the cached result directly for identical input; no duplicate billing

# ─── Chunk text to avoid excessive length ──────────────────────────────────────
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
chunk_size=2000, # 2000 characters per chunk
chunk_overlap=200 # Overlap 200 characters to maintain context coherence
)
chunks = splitter.split_text(long_document)

Typical Application Scenarios

AI Workflow has been widely adopted across multiple domains. The following are the six most representative scenarios.

Best Practices and Common Pitfalls

Common Pitfalls and Solutions

The following are the most common problems beginners encounter when using AI Workflow.

Pitfall	Symptom	Solution
Hallucination cascade	The previous AI output is incorrect, passed as fact to the next step, amplifying the error.	Add verification steps at key nodes; use tools for numerical information.
Infinite loop	ReAct Agent repeatedly calls tools without knowing when to stop.	Set max_iterations; give the LLM explicit termination conditions.
Context explosion	As steps increase, the text passed to the LLM becomes longer and exceeds the window.	Pass only necessary fields at each step; compress historical information with summaries.
Tool abuse	The Agent can answer directly but keeps calling tools.	Optimize tool descriptions; clearly tell the LLM when tools are not needed.
JSON parse failure	LLM output format is unstable, causing the program to crash.	Add safe_parse_json; use LangChain OutputParser.
Cost overrun	Token usage is not estimated, and the monthly bill exceeds expectations.	First test with gpt-4o-mini; use LangSmith to monitor usage.
Concurrency conflict	Multiple Agents writing to the same state simultaneously cause data corruption.	Use LangGraph's built-in state management; avoid sharing mutable state.

Summary and Learning Path

Core Knowledge Review

Recommended Learning Path

Phase 1: Understanding the Basics (1 Week)

Understand LLM API calls, be able to make requests using the OpenAI SDK
Run through the LangChain sequential chain example in this article
Understand the basic principles of Prompt Engineering

Phase 2: Tools and Agents (2 weeks)

Learn to define tools with the @tool decorator
Run through the ReAct Agent example, observe the verbose logs to understand the loop logic
Use Dify or n8n to build a visual Workflow prototype

Phase 3: Complex Workflow (3 weeks)

Learn LangGraph state graphs, implement a workflow with conditional branching
Use CrewAI to implement a dual-Agent collaboration task
Integrate real tools (Tavily Search, Python REPL)
Use LangSmith to observe and debug your Workflow

Phase 4: Production Ready (Continuous)

Implement error retry and structured output parsing
Cost monitoring and optimization (tiered model usage)
Deploy to production, set up monitoring and alerting

Reference Resources

other extensions