AI Prompt链式调用：把复杂任务拆成一步步

Prompt Chain：从单步到多步的进化

实战：自动化产品上架系统

from langgraph.graph import StateGraph
from typing import TypedDict, Optional

class ProductState(TypedDict):
    raw_input: str
    extracted_info: dict
    title: str
    description: str
    seo_keywords: list
    final_output: str

# 步骤1：提取核心信息
def extract_info(state: ProductState) -> dict:
    prompt = f"""从以下产品描述中提取结构化信息：
产品名称、品牌、规格、价格、主要卖点（最多3个）。

原文：{state['raw_input']}

请以JSON格式输出。"""
    result = llm.invoke(prompt)
    return {"extracted_info": json.loads(result)}

# 步骤2：生成标题
def generate_title(state: ProductState) -> dict:
    info = state['extracted_info']
    prompt = f"""根据以下信息生成3个产品标题备选：
名称：{info['name']}
品牌：{info['brand']}
规格：{info['spec']}
卖点：{', '.join(info['key_points'])}
要求：标题不超过30字，包含核心卖点，有吸引力。"""
    return {"title": llm.invoke(prompt)}

# 步骤3：写详情描述
def generate_description(state: ProductState) -> dict:
    prompt = f"""写一段产品详情描述（200-300字）：
标题：{state['title']}
品牌：{state['extracted_info']['brand']}
规格：{state['extracted_info']['spec']}
价格：{state['extracted_info']['price']}
要求：描述产品优势、适用场景、使用方式。"""
    return {"description": llm.invoke(prompt)}

# 步骤4：提取SEO关键词
def extract_seo_keywords(state: ProductState) -> dict:
    prompt = f"""从以下内容中提取5个SEO关键词：

标题：{state['title']}
描述：{state['description']}

关键词需包含2个长尾词（3-5字）、2个核心词、1个场景词。"""
    return {"seo_keywords": llm.invoke(prompt).split('\n')[:5]}

# 步骤5：组装最终输出
def assemble(state: ProductState) -> dict:
    output = f"""【产品标题】
{state['title']}

【产品详情】
{state['description']}

【SEO关键词】
{', '.join(state['seo_keywords'])}"""
    return {"final_output": output}

# 构建Chain
workflow = StateGraph(ProductState)
workflow.add_node("extract", extract_info)
workflow.add_node("title", generate_title)
workflow.add_node("desc", generate_description)
workflow.add_node("seo", extract_seo_keywords)
workflow.add_node("assemble", assemble)
workflow.set_entry_point("extract")
workflow.add_edge("extract", "title")
workflow.add_edge("title", "desc")
workflow.add_edge("desc", "seo")
workflow.add_edge("seo", "assemble")
chain = workflow.compile()

单步Prompt vs 链式调用的对比

实施步骤

第一步：分析复杂任务，画出流程图——从输入到输出的每一步是什么，哪步可能出错。

第二步：用StateGraph或简单的函数链（函数A→函数B→函数C）实现第一步的原型。

第三步：每步输出到控制台检查质量，单独优化质量不达标的步骤。

第四步：添加中间结果缓存——如果某步的输入和之前一样，直接返回缓存结果，节省token。

第五步：生产环境中，对每步做超时监控。某步超过阈值则跳过该步（取上一步结果直接往下传），保证整体不崩溃。