LangGraph实战：构建广告智能投放助手，从单步推理到多智能体闭环

摘要

本文基于生产环境验证的LangGraph开发实战，拆解智能体状态管理、三元记忆架构与多智能体协同的核心技术，通过广告智能投放助手全流程落地案例，解决传统投放流程繁琐、决策不可控、效果难以闭环的行业痛点。

---

一、核心认知：LangGraph重构企业级智能体开发范式

当前AI智能体开发普遍存在三大痛点：循环控制逻辑混乱、状态管理碎片化、缺乏生产级可控性。传统基于LangChain的链式调用只能处理线性任务，无法应对广告投放这类需要多轮推理、工具调用、人类介入与状态持久化的复杂业务场景。

基于某广告科技公司3个生产级项目的落地经验，我们明确：LangGraph不是LangChain的简单升级，而是面向复杂智能体的图编排引擎。它通过状态图建模、内置持久化、人类介入与多智能体协同四大核心能力，将智能体从"玩具级Demo"推向"生产级应用"。

#架构真经 #大模型应用

1.1 LangGraph vs 传统Agent开发：核心差异

维度	传统链式Agent	LangGraph智能体
控制流	线性执行，无法处理复杂循环	有向图建模，支持条件分支、循环与并行
状态管理	无统一状态，依赖全局变量传递	中心化状态对象，自动追踪所有节点变更
持久化	需手动实现会话存储	内置持久化层，按thread_id隔离多用户状态
人类介入	无原生支持，需硬编码中断逻辑	原生支持工具调用前中断、状态编辑与时间回溯
多智能体	需自定义通信协议与调度逻辑	原生支持多节点协作，统一状态共享
可观测性	无统一轨迹追踪	全链路状态快照，支持历史回溯与调试

1.2 LangGraph三大核心设计理念

1. 状态即一等公民：所有节点共享一个结构化状态对象，节点执行结果自动合并到状态中，实现数据流转的可追溯性；
2. 图即控制流：将业务流程抽象为节点（函数/智能体）与边（数据流），条件边实现分支逻辑，循环边实现迭代优化；
3. 原生生产级能力：内置持久化、人类介入、错误重试与并发控制，无需额外开发基础设施。

---

二、实战落地：广告智能投放助手全流程开发

本案例来自真实项目+生产环境验证，已在某互联网平台广告业务线稳定运行3个月，覆盖中小客户自助投放场景，投放效率提升70%，人工审核量下降60%。

2.1 业务需求与流程建模

广告智能投放助手需要完成以下核心任务：

1. 解析用户自然语言投放需求（如"帮我给新品运动鞋投放1万元预算的抖音信息流广告"）；
2. 检索历史投放数据与行业基准，生成定向人群、出价策略与创意建议；
3. 预估投放效果（曝光量、点击量、转化率）；
4. 提交人工审核，审核通过后自动执行投放；
5. 实时监控投放效果，动态调整策略。

将上述流程抽象为LangGraph状态图：

• 节点：意图识别 → 数据检索 → 策略生成 → 效果预估 → 审核判断 → 执行投放 → 效果监控
• 条件边：
  • 效果预估不达标 → 返回策略生成节点重新优化
  • 审核不通过 → 返回策略生成节点修改
  • 审核通过 → 执行投放
  • 投放效果异常 → 触发策略调整

2.2 核心代码实现

步骤1：定义状态结构

使用TypedDict定义智能体状态，包含所有业务所需字段：

from typing import TypedDict, Annotated, List, Dict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

class AdCampaignState(TypedDict):
    # 用户输入与意图
    user_query: str
    intention: Dict[str, str]  # 包含product、budget、platform、goal等字段
    user_id: str
    
    # 数据与策略
    historical_data: List[Dict]
    industry_benchmark: Dict
    campaign_strategy: Dict  # 包含定向、出价、创意、预算分配
    effect_estimation: Dict  # 包含impressions、clicks、conversions、roi
    
    # 审核与执行
    approval_status: str  # pending/approved/rejected
    approval_comments: str
    campaign_id: str
    real_time_data: Dict
    
    # 消息历史
    messages: Annotated[List, add_messages]

步骤2：实现核心节点函数

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

# 初始化大模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def intention_recognition_node(state: AdCampaignState) -> AdCampaignState:
    """解析用户投放意图，提取产品、预算、平台、目标等关键信息"""
    system_prompt = """
    你是广告投放意图识别专家，从用户自然语言查询中提取以下结构化信息：
    - product: 推广产品名称
    - budget: 投放预算（数字+单位，如"10000元"）
    - platform: 投放平台（抖音、快手、微信、小红书等）
    - goal: 投放目标（曝光、点击、转化、加粉等）
    以JSON格式输出，缺失字段填null。
    """
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=state["user_query"])
    ])
    intention = eval(response.content)  # 生产环境需使用JSON Schema校验
    return {"intention": intention}

def data_retrieval_node(state: AdCampaignState) -> AdCampaignState:
    """检索历史投放数据与行业基准"""
    # 模拟调用数据API，生产环境对接大数据平台
    historical_data = [
        {"campaign_id": "123", "product": "运动鞋", "platform": "抖音", "roi": 2.3},
        {"campaign_id": "456", "product": "运动鞋", "platform": "快手", "roi": 1.8}
    ]
    industry_benchmark = {"click_rate": "2.5%", "conversion_rate": "3.2%", "avg_cpc": "1.2元"}
    return {"historical_data": historical_data, "industry_benchmark": industry_benchmark}

def strategy_generation_node(state: AdCampaignState) -> AdCampaignState:
    """基于数据生成投放策略"""
    system_prompt = f"""
    你是资深广告优化师，根据以下信息生成投放策略：
    用户需求：{state['intention']}
    历史投放数据：{state['historical_data']}
    行业基准：{state['industry_benchmark']}
    策略需包含：定向人群（年龄、性别、兴趣）、出价策略、创意方向、预算分配。
    以JSON格式输出。
    """
    response = llm.invoke(SystemMessage(content=system_prompt))
    campaign_strategy = eval(response.content)
    return {"campaign_strategy": campaign_strategy}

def effect_estimation_node(state: AdCampaignState) -> AdCampaignState:
    """预估投放效果"""
    # 模拟效果预估模型，生产环境对接机器学习模型
    budget = int(state['intention']['budget'].replace("元", ""))
    cpc = float(state['industry_benchmark']['avg_cpc'].replace("元", ""))
    clicks = int(budget / cpc)
    impressions = int(clicks / float(state['industry_benchmark']['click_rate'].replace("%", "")) * 100)
    conversions = int(clicks * float(state['industry_benchmark']['conversion_rate'].replace("%", "")) / 100)
    roi = 2.5  # 模拟ROI预估
    
    effect_estimation = {
        "impressions": impressions,
        "clicks": clicks,
        "conversions": conversions,
        "roi": roi
    }
    return {"effect_estimation": effect_estimation}

def approval_check_node(state: AdCampaignState) -> str:
    """审核判断节点，返回下一步走向"""
    if state["approval_status"] == "approved":
        return "execute"
    elif state["approval_status"] == "rejected":
        return "revise"
    else:
        return "pending"

def execute_campaign_node(state: AdCampaignState) -> AdCampaignState:
    """执行投放，返回投放ID"""
    # 模拟调用广告平台API
    campaign_id = "AD" + str(hash(state["user_query"]))[:8]
    return {"campaign_id": campaign_id, "messages": [HumanMessage(content=f"投放已执行，ID：{campaign_id}")]}

步骤3：构建与编译状态图

# 初始化状态图
builder = StateGraph(AdCampaignState)

# 添加节点
builder.add_node("intention_recognition", intention_recognition_node)
builder.add_node("data_retrieval", data_retrieval_node)
builder.add_node("strategy_generation", strategy_generation_node)
builder.add_node("effect_estimation", effect_estimation_node)
builder.add_node("execute_campaign", execute_campaign_node)

# 添加边
builder.add_edge("intention_recognition", "data_retrieval")
builder.add_edge("data_retrieval", "strategy_generation")
builder.add_edge("strategy_generation", "effect_estimation")

# 添加条件边：效果预估不达标则重新生成策略
builder.add_conditional_edges(
    "effect_estimation",
    lambda x: "retry" if x["effect_estimation"]["roi"] < 2.0 else "approve",
    {
        "retry": "strategy_generation",
        "approve": "execute_campaign"
    }
)

# 添加人类介入：执行投放前强制中断，等待人工审核
builder.add_edge("effect_estimation", "execute_campaign")
builder.set_entry_point("intention_recognition")
builder.set_finish_point("execute_campaign")

# 编译图，启用持久化与人类介入
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
graph = builder.compile(checkpointer=memory, interrupt_before=["execute_campaign"])

步骤4：运行智能体与人类介入

# 配置线程ID，用于隔离不同用户会话
config = {"configurable": {"thread_id": "user_123"}}

# 初始输入
initial_state = {
    "user_query": "帮我给新品运动鞋投放1万元预算的抖音信息流广告，目标是转化",
    "user_id": "user_123",
    "approval_status": "pending"
}

# 运行智能体，直到中断点
for event in graph.stream(initial_state, config):
    for value in event.values():
        if "messages" in value:
            print("助手：", value["messages"][-1].content)

# 查看当前状态，等待人工审核
current_state = graph.get_state(config)
print("当前策略：", current_state.values["campaign_strategy"])
print("效果预估：", current_state.values["effect_estimation"])

# 人工审核通过，更新状态并继续执行
graph.update_state(
    config,
    {"approval_status": "approved"},
    as_node="effect_estimation"
)

# 继续执行剩余流程
for event in graph.stream(None, config):
    for value in event.values():
        if "messages" in value:
            print("助手：", value["messages"][-1].content)

2.3 集成三元记忆架构

基于LangGraph的持久化能力，集成语义记忆、情景记忆与程序记忆，提升智能体的个性化与准确性：

1. 语义记忆：存储用户历史投放偏好、产品信息与行业规则，使用向量数据库检索；
2. 情景记忆：存储过往成功的投放案例，在策略生成时作为少样本示例注入；
3. 程序记忆：存储投放规则与审核标准，通过动态提示词更新实现规则迭代。

# 语义记忆工具示例
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma(collection_name="ad_knowledge", embedding_function=embeddings)

def search_semantic_memory(query: str) -> str:
    """检索语义记忆"""
    docs = vector_store.similarity_search(query, k=3)
    return "\n".join([doc.page_content for doc in docs])

# 在策略生成节点中注入语义记忆
def strategy_generation_node_with_memory(state: AdCampaignState) -> AdCampaignState:
    knowledge = search_semantic_memory(f"{state['intention']['product']} 广告投放规则")
    system_prompt = f"""
    参考以下行业知识生成投放策略：
    {knowledge}
    ...（其余提示词不变）
    """
    # 后续逻辑与之前一致

---

三、高级进阶：多智能体协同与生产级优化

3.1 多智能体协同架构

对于复杂的广告投放场景，可将单一智能体拆分为多个专业子智能体，通过LangGraph实现协同工作：

• 数据分析师Agent：负责数据检索、清洗与分析；
• 策略师Agent：负责生成投放策略与创意；
• 效果预估Agent：负责构建预测模型与效果评估；
• 审核员Agent：负责初步合规审核，过滤明显违规策略。

3.2 生产级避坑指南

基于3个生产项目的踩坑经验，总结五大核心问题与解决方案：#避坑实录 #源码级解读

核心踩坑点	现象描述	根因分析	生产环境解决方案
状态更新错误	节点执行结果未正确合并到状态，数据丢失	未使用Annotated+add_messages处理列表字段	所有列表类型字段必须使用add_messages注解，确保追加而非覆盖
工具调用失败	频繁出现工具参数错误、API超时	大模型幻觉导致参数格式错误，未做异常处理	工具调用前添加JSON Schema校验，实现指数退避重试机制，超时自动降级
记忆污染	错误数据写入记忆库，导致后续决策错误	热路径写入未做校验，依赖大模型生成正确参数	记忆写入前添加轻量级校验层，关键操作日志化，后台异步修复错误数据
并发性能差	多用户同时访问时响应缓慢	单线程处理所有请求，状态锁竞争	使用线程池处理并发请求，按thread_id隔离状态，实现无锁读写
调试困难	生产环境问题难以复现与定位	缺乏全链路轨迹追踪	开启LangSmith追踪，记录所有状态快照与节点执行日志，支持时间回溯调试

---

四、核心总结

LangGraph通过图编排、状态管理与原生生产级能力，彻底解决了复杂智能体的落地难题。在广告行业，基于LangGraph构建的智能投放助手能够实现从需求解析到效果监控的全流程自动化，同时通过人类介入与多智能体协同保障决策的准确性与可控性。

未来，随着LangGraph生态的不断成熟，它将成为企业级AI应用的标准基础设施，推动智能体从单一任务助手向复杂业务系统演进。对于技术从业者而言，掌握LangGraph的核心设计理念与工程实践，将是在AI时代构建核心竞争力的关键。