From 4 Hours to 3 Minutes: Graph Engineering Case Study for E-commerce Customer Service, Selection & Marketing

Problem Context

A small-to-medium e-commerce company selling home goods (~500 SKUs, 200-300 daily customer inquiries, 5-person operations team) faced three disconnected pain points:

Customer Service : 5 operators rotating shifts, average response time 4 hours; 70% of queries were repetitive (returns, logistics, product specs), only 30% needed human judgment.

Product Selection : Weekly competitor analysis and sales review relied on intuition without systematic process, leading to missed opportunities or wrong bets.

Marketing Copy : Each new product required detail-page copy, Xiaohongshu posts, and WeChat promotions; 5 operators wrote independently, causing inconsistent style, variable quality, and high time cost.

They had tried ChatGPT for each task individually but never formed an integrated system.

Solution: Three Interconnected Graphs

The core insight: these three problems are naturally linked—customer feedback informs product selection, selection insights drive marketing copy, and marketing performance feeds back into selection. The solution uses three LangGraph graphs that share data.

Graph 1: Customer Service Routing

State Definition

from typing import TypedDict, Optional
from enum import Enum

class QuestionType(str, Enum):
    FAQ = "faq"      # common questions
    ORDER = "order"  # order-related
    COMPLEX = "complex"  # needs human

class CustomerServiceState(TypedDict):
    # Input
    customer_id: str
    message: str
    order_id: Optional[str]
    # Intermediate
    question_type: Optional[QuestionType]
    retrieved_faq: Optional[str]
    order_info: Optional[dict]
    # Output
    final_reply: Optional[str]
    need_human: bool
    human_summary: Optional[str]  # summary for human handoff
    # Data collection for product selection
    feedback_tags: list[str]

Core Nodes

classify_node : Uses gpt-4o-mini (temperature 0) to classify question type and extract feedback tags (quality, logistics, price, function) for downstream selection analysis.

faq_node : Retrieves top-3 relevant FAQs from vector database.

order_node : Extracts order ID from message if missing, queries order system.

human_node : Generates a ≤50-character summary for human agents, sets need_human=True, returns holding reply.

reply_node : Uses gpt-4o-mini (temperature 0.3) to generate final reply based on FAQ context or order info.

record_node : Persists feedback (customer_id, message, question_type, tags, need_human) to database for selection graph.

Routing & Compilation

def route_by_type(state: CustomerServiceState) -> str:
    qtype = state.get("question_type", "complex")
    if qtype == "faq": return "faq"
    elif qtype == "order": return "order"
    else: return "human"

cs_builder = StateGraph(CustomerServiceState)
cs_builder.add_node("classify", classify_node)
cs_builder.add_node("faq", faq_node)
cs_builder.add_node("order", order_node)
cs_builder.add_node("human", human_node)
cs_builder.add_node("reply", reply_node)
cs_builder.add_node("record", record_node)

cs_builder.set_entry_point("classify")
cs_builder.add_conditional_edges("classify", route_by_type, {
    "faq": "faq", "order": "order", "human": "human"
})
cs_builder.add_edge("faq", "reply")
cs_builder.add_edge("order", "reply")
cs_builder.add_edge("human", "reply")
cs_builder.add_edge("reply", "record")
cs_builder.add_edge("record", END)

from langgraph.checkpoint.sqlite import SqliteSaver
cs_graph = cs_builder.compile(checkpointer=SqliteSaver.from_conn_string("cs.db"))

Results

70% of inquiries fully automated; average response time dropped from 4 hours to 3 minutes. The record_node daily accumulates feedback tags that become input for the selection graph.

Graph 2: Product Selection Analysis

State with Parallel Reducer

from typing import TypedDict, Annotated
import operator

class SelectionState(TypedDict):
    week: str
    category: str
    # Parallel nodes append via reducer
    data_sources: Annotated[list[dict], operator.add]
    analysis: str
    confidence: float  # 0-1
    selection_report: str
    iteration: int  # prevents infinite loops

Parallel Fan-Out Nodes

customer_feedback_node : Queries feedback DB for the week/category, counts high-frequency tags.

competitor_node : Scrapes top-20 competitor products in category.

sales_node : Queries internal sales DB for week/category.

Each returns

{"data_sources": [{"type": "feedback|competitor|sales", "data": ...}]}

; the operator.add reducer merges them into the list.

Aggregation & Analysis with Confidence Loop

aggregate_node : Formats merged data for LLM consumption.

analysis_node : Uses gpt-4o (temperature 0.2) to produce selection recommendations (user-preferred attributes, competitor trends, store strengths/weaknesses, 3-5 concrete suggestions) and a confidence score (0-1).

route_after_review : If confidence ≥ 0.7 or iteration ≥ 2, proceed to output; else loop to supplement_data_node for additional data gathering.

Graph Structure

sel_builder = StateGraph(SelectionState)
sel_builder.add_node("trigger", lambda s: s)
sel_builder.add_node("feedback", customer_feedback_node)
sel_builder.add_node("competitor", competitor_node)
sel_builder.add_node("sales", sales_node)
sel_builder.add_node("aggregate", aggregate_node)
sel_builder.add_node("analysis", analysis_node)
sel_builder.add_node("output", generate_report_node)
sel_builder.add_node("supplement", supplement_data_node)

sel_builder.set_entry_point("trigger")
# Fan-out
sel_builder.add_edge("trigger", "feedback")
sel_builder.add_edge("trigger", "competitor")
sel_builder.add_edge("trigger", "sales")
# Fan-in
sel_builder.add_edge("feedback", "aggregate")
sel_builder.add_edge("competitor", "aggregate")
sel_builder.add_edge("sales", "aggregate")
sel_builder.add_edge("aggregate", "analysis")
sel_builder.add_conditional_edges("analysis", route_after_review, {
    "output": "output", "supplement": "supplement"
})
sel_builder.add_edge("supplement", "analysis")
sel_builder.add_edge("output", END)

sel_graph = sel_builder.compile()

Graph 3: Marketing Copy Generation

State with Parallel Drafts & Review Loop

class MarketingState(TypedDict):
    product_name: str
    product_info: str
    selection_insights: str  # from selection graph
    core_selling_points: str
    drafts: Annotated[list[dict], operator.add]  # parallel platform copies
    review_result: dict
    final_copies: dict
    revision_count: int

Nodes

extract_selling_points : Distills 3-5 core selling points from product info + selection insights.

detail_page_node , xiaohongshu_node , wechat_node : Run in parallel, each generates platform-specific copy (detail page: title, subtitle, bullet points, description; Xiaohongshu: emoji title, 200-char colloquial body, 5-8 tags; WeChat: ≤80 chars with call-to-action). All use gpt-4o-mini.

review_node : Checks all drafts for brand consistency (professional, warm, practical), returns JSON with overall_pass, issues, platforms_to_revise.

route_after_review : If pass or revision_count ≥ 2, go to output; else loop to revise_node.

Graph Structure

mkt_builder = StateGraph(MarketingState)
mkt_builder.add_node("extract", extract_selling_points)
mkt_builder.add_node("detail", detail_page_node)
mkt_builder.add_node("xiaohongshu", xiaohongshu_node)
mkt_builder.add_node("wechat", wechat_node)
mkt_builder.add_node("review", review_node)
mkt_builder.add_node("output", package_output_node)
mkt_builder.add_node("revise", revise_node)

mkt_builder.set_entry_point("extract")
# Fan-out
mkt_builder.add_edge("extract", "detail")
mkt_builder.add_edge("extract", "xiaohongshu")
mkt_builder.add_edge("extract", "wechat")
# Fan-in
mkt_builder.add_edge("detail", "review")
mkt_builder.add_edge("xiaohongshu", "review")
mkt_builder.add_edge("wechat", "review")
mkt_builder.add_conditional_edges("review", route_after_review, {
    "output": "output", "revise": "revise"
})
mkt_builder.add_edge("revise", "review")
mkt_builder.add_edge("output", END)

mkt_graph = mkt_builder.compile()

Inter-Graph Data Flow

def weekly_selection_pipeline(category: str):
    week = get_current_week()
    # 1. Run selection graph
    sel_result = sel_graph.invoke({
        "week": week,
        "category": category,
        "data_sources": [],
        "iteration": 0
    })
    # 2. Pass selection insights to marketing graph for each recommended product
    for product in sel_result["recommended_products"]:
        mkt_result = mkt_graph.invoke({
            "product_name": product["name"],
            "product_info": product["info"],
            "selection_insights": sel_result["selection_report"],  # key link
            "drafts": [],
            "revision_count": 0
        })
        save_marketing_copies(product["id"], mkt_result["final_copies"])
    # Customer service feedback stored daily → selection graph reads weekly → closed loop

Key Takeaways

Three core e-commerce pain points (service, selection, marketing) solved by three interconnected graphs.

Customer service graph: classification + conditional routing + human fallback + feedback collection for downstream use.

Selection graph: parallel fan-out to three data sources + reducer aggregation + confidence-gated iteration loop.

Marketing graph: parallel multi-platform generation + brand-consistency review + bounded revision loop.

Data flow: service feedback → selection insights → marketing copies, forming a business closed loop.

Monthly API cost ~¥800, replacing large volumes of repetitive manual work with clear ROI.