提示链

提示链设计思维

提示链（Prompt Chaining），也称为流水线（Pipeline）模式，是将复杂任务拆解为一系列更小、更易管理的子问题，每个子问题通过专门设计的提示单独处理，并将前一步的输出作为下一步的输入，形成链式依赖。

核心思想：与其让 LLM 一步到位解决复杂问题，提示链采用分而治之策略，每一步的输出作为下一步的输入至关重要，这种信息传递建立了依赖链，使 LLM 能够在前一步基础上不断完善理解，逐步逼近目标解。

  1. 明确每个步骤的目标：每个提示都应该有清晰、单一的目标，避免在一个提示中处理多个不相关的任务
  2. 合理设计上下文传递：前一个步骤的输出应该包含下一个步骤所需的关键信息，但也要避免传递过多冗余信息
  3. 使用角色定位：为每个步骤分配明确的角色，如"市场分析师"、"技术专家"等，提升专业性
  4. 结构化输出：提升下一步提示词拼接的灵活性

为什么需要提示链？

单一提示的局限性 ⚠️

对于多层次任务，单一复杂提示往往效率低下，模型容易出现以下问题：

  • 忽略部分指令：模型可能只完成部分任务，遗漏关键环节
  • 丢失上下文：在长文本处理中，模型可能无法保持完整的上下文理解
  • 错误累积：一步到位的处理方式，错误会直接体现在最终输出中
  • 上下文窗口不足：复杂任务需要大量上下文，可能超出模型的窗口限制
  • 出现幻觉：模型可能生成看似合理但实际错误的信息

示例：要求模型分析市场调研报告、总结发现、提取数据点并撰写邮件，单一提示可能导致模型只完成部分任务，遗漏关键环节。

提示链的优势 ✅

提示链通过将复杂任务拆解为聚焦的顺序流程，显著提升可靠性和可控性：

  1. 模块化设计：将复杂任务拆分为独立的模块，每个模块负责一个特定功能，更容易理解、调试和维护
  2. 提高准确性：每个步骤的输出都经过精心设计的提示进行约束和优化，减少错误传播的可能性
  3. 增强可解释性：每个步骤都会产生中间输出，这些中间结果可以被检查、验证和调试
  4. 灵活组合：可以根据需求动态调整步骤，添加、删除或重新排序

角色分配：为确保每步任务准确，可为模型分配不同角色，如"市场分析师"、"贸易分析师"、"文档专家"等，让每个步骤都有明确的专业定位。

应用场景与流程构建

1. 信息处理流程 📄

场景：许多任务需对原始信息多次转换，如文档摘要、实体提取、用实体查询数据库、生成报告

提示链：

  1. 从指定URL或文档提取文本内容
  2. 摘要清洗后的文本
  3. 从摘要或原文中提取实体（如姓名、日期、地点）
  4. 用实体查询内部知识库
  5. 生成包含摘要、实体和查询结果的最终报告

流程构建：

from typing import TypedDict, List
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)

class Entity(BaseModel):
    name: str
    date: str
    location: str
    organization: str

class AgentState(TypedDict):
    source: str
    text: str
    summary: str
    entities: List[Entity]
    query_results: str
    final_report: str

# 提示 1：提取文本内容
def extract_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是文档处理专家。请从以下 URL 或文档中提取文本内容：

{source}

输出格式：纯文本内容，去除格式标记。""")
    chain = prompt | llm | StrOutputParser()
    return {"text": chain.invoke({"source": state["source"]})}

# 提示 2：摘要文本
def summarize_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是内容摘要专家。请对以下文本进行摘要，保留关键信息：
    
{text}

输出格式：200字以内的摘要。""")
    chain = prompt | llm | StrOutputParser()
    return {"summary": chain.invoke({"text": state["text"]})}

# 提示 3：提取实体
def extract_entities_node(state: AgentState):
    summary = state.get('summary', '')
    parser = PydanticOutputParser(pydantic_object=Entity)
    format_instructions = parser.get_format_instructions()
    prompt = ChatPromptTemplate.from_template("""你是信息提取专家。请从以下文本中提取实体信息：
    
{summary}

需要提取的实体类型：
- 人名
- 日期
- 地点
- 组织

{format_instructions}""")
    chain = prompt | llm | parser
    entity = chain.invoke({"summary": state["summary"], "format_instructions": format_instructions})
    return {"entities": [entity]}

# 提示 4：查询知识库
def query_kb_node(state: AgentState):
    entities = state.get('entities', [])
    entities_str = "\n".join([f"- {entity.name} ({entity.date}, {entity.location}, {entity.organization})" for entity in entities])
    prompt = ChatPromptTemplate.from_template("""你是知识库查询专家。根据以下实体信息，查询相关知识：

{entities_str}

输出格式：200字以内的查询结果，不要有任何解释。""")
    chain = prompt | llm | StrOutputParser()
    return {"query_results": chain.invoke({"entities_str": entities_str})}

# 提示 5：生成最终报告
def report_node(state: AgentState):
    summary = state.get('summary', '')
    entities = state.get('entities', [])
    entities_str = "\n".join([f"- {entity.name} ({entity.date}, {entity.location}, {entity.organization})" for entity in entities])
    query_results = state.get('query_results', '')
    prompt = ChatPromptTemplate.from_template("""你是报告撰写专家。请基于以下信息生成综合报告：
    
摘要：{summary}
实体：{entities_str}
查询结果：{query_results}

输出格式：结构化的 Markdown 报告，包含标题、摘要、关键发现、结论等部分。""")
    chain = prompt | llm | StrOutputParser()
    return {"final_report": chain.invoke({"summary": summary, "entities_str": entities_str, "query_results": query_results})}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("extract", extract_node)
workflow.add_node("summarize", summarize_node)
workflow.add_node("entities", extract_entities_node)
workflow.add_node("query", query_kb_node)
workflow.add_node("report", report_node)

# 定义连线
workflow.set_entry_point("extract")
workflow.add_edge("extract", "summarize")
workflow.add_edge("summarize", "entities")
workflow.add_edge("entities", "query")
workflow.add_edge("query", "report")
workflow.add_edge("report", END)

# 编译
app = workflow.compile()

# 运行
inputs = {"source": "http://example.com/data"}
result = app.invoke(inputs)
print(result["final_report"])

2. 复杂问答 💬

场景：回答需要多步推理或信息检索的问题

提示链：

  1. 识别用户问题的核心子问题（崩盘原因、政府应对）
  2. 检索第一个子问题的相关信息
  3. 检索第二个子问题的相关信息
  4. 综合所有信息，形成完整答案

流程构建：

from typing import TypedDict, List
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)

class SubQuestion(BaseModel):
    question: str
    type: str  # 原因/应对/影响等

class AgentState(TypedDict):
    question: str
    sub_questions: List[SubQuestion]
    answer_1: str
    answer_2: str
    final_answer: str

# 提示 1：识别子问题
def decompose_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是问题分析专家。请分析以下问题，识别需要回答的核心子问题：

问题：{question}

输出格式：JSON 格式，包含 sub_questions 数组，每个子问题包含 question 和 type（原因/应对/影响等）。""")
    parser = JsonOutputParser(pydantic_object=List[SubQuestion])
    chain = prompt | llm | parser
    result = chain.invoke({"question": state["question"]})
    return {"sub_questions": result}

# 提示 2：检索第一个子问题
def retrieve_1_node(state: AgentState):
    sub_questions = state.get("sub_questions", [])
    if not sub_questions:
        return {"answer_1": ""}
    sub_q = sub_questions[0]
    prompt = ChatPromptTemplate.from_template("""你是信息检索专家。请检索关于以下问题的相关信息：

子问题：{sub_question}

输出格式：详细的信息摘要，包含关键事实和数据。""")
    chain = prompt | llm | StrOutputParser()
    return {"answer_1": chain.invoke({"sub_question": sub_q.question})}

# 提示 3：检索第二个子问题
def retrieve_2_node(state: AgentState):
    sub_questions = state.get("sub_questions", [])
    if len(sub_questions) < 2:
        return {"answer_2": ""}
    sub_q = sub_questions[1]
    prompt = ChatPromptTemplate.from_template("""你是信息检索专家。请检索关于以下问题的相关信息：

子问题：{sub_question}

输出格式：详细的信息摘要，包含关键事实和数据。""")
    chain = prompt | llm | StrOutputParser()
    return {"answer_2": chain.invoke({"sub_question": sub_q.question})}

# 提示 4：综合答案
def synthesize_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是答案综合专家。请基于以下信息，形成完整的答案：

原问题：{question}
子问题1的答案：{answer_1}
子问题2的答案：{answer_2}

输出格式：结构化的答案，包含：
1. 问题概述
2. 各子问题的详细回答
3. 综合结论""")
    chain = prompt | llm | StrOutputParser()
    return {"final_answer": chain.invoke({
        "question": state["question"],
        "answer_1": state.get("answer_1", ""),
        "answer_2": state.get("answer_2", "")
    })}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("decompose", decompose_node)
workflow.add_node("retrieve_1", retrieve_1_node)
workflow.add_node("retrieve_2", retrieve_2_node)
workflow.add_node("synthesize", synthesize_node)

# 定义连线
workflow.set_entry_point("decompose")
workflow.add_edge("decompose", "retrieve_1")
workflow.add_edge("retrieve_1", "retrieve_2")
workflow.add_edge("retrieve_2", "synthesize")
workflow.add_edge("synthesize", END)

# 编译
app = workflow.compile()

# 运行
inputs = {"question": "1929 年股市崩盘的主要原因及政府政策应对？"}
result = app.invoke(inputs)
print(result["final_answer"])

3. 数据提取与转换 🔄

场景：从非结构化文本（发票、表单、邮件）中提取结构化数据

提示链：

  1. 尝试从文档中提取指定字段（姓名、地址、金额、日期）
  2. 检查字段是否齐全且格式正确
  3. 若字段缺失或格式错误，重新提示模型查找缺失/错误信息
  4. 再次验证结果，必要时重复步骤3
  5. 格式化输出为标准格式

流程构建：

from typing import TypedDict, Optional
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
import json

llm = ChatOpenAI(temperature=0)

class ExtractedData(BaseModel):
    name: Optional[str] = None
    address: Optional[str] = None
    amount: Optional[str] = None
    date: Optional[str] = None

class AgentState(TypedDict):
    document: str
    extracted_data: str
    validation_count: int
    validated_data: str
    formatted_data: str

# 提示 1：初步提取
def extract_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是数据提取专家。请从以下文档中提取指定字段：

文档内容：{document}

需要提取的字段：
- 姓名
- 地址
- 金额
- 日期

输出格式：JSON 格式，包含所有字段，如果某个字段无法提取，使用 null。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({"document": state["document"]})
    return {"extracted_data": json.dumps(result, ensure_ascii=False), "validation_count": 0}

# 提示 2：验证和补全
def validate_node(state: AgentState):
    extracted_data = state.get("extracted_data", "{}")
    validation_count = state.get("validation_count", 0)
    
    prompt = ChatPromptTemplate.from_template("""你是数据验证专家。请检查以下提取结果，查找缺失或错误的字段：

提取结果：{extracted_data}
原始文档：{document}

如果字段缺失或格式错误，请重新查找并修正。

输出格式：JSON 格式，包含所有字段的完整和正确值。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({
        "extracted_data": extracted_data,
        "document": state["document"]
    })
    
    # 检查是否所有字段都有值
    all_fields_present = all([
        result.get("name"),
        result.get("address"),
        result.get("amount"),
        result.get("date")
    ])
    
    return {
        "validated_data": json.dumps(result, ensure_ascii=False),
        "validation_count": validation_count + 1
    }

# 条件判断：是否需要重新验证
def should_revalidate(state: AgentState) -> str:
    validated_data = json.loads(state.get("validated_data", "{}"))
    validation_count = state.get("validation_count", 0)
    
    # 检查字段完整性
    all_fields_present = all([
        validated_data.get("name"),
        validated_data.get("address"),
        validated_data.get("amount"),
        validated_data.get("date")
    ])
    
    # 如果字段不完整且未超过最大重试次数，继续验证
    if not all_fields_present and validation_count < 3:
        return "validate"
    else:
        return "format"

# 提示 3：格式化输出
def format_node(state: AgentState):
    validated_data = json.loads(state.get("validated_data", "{}"))
    prompt = ChatPromptTemplate.from_template("""你是数据格式化专家。请将以下数据转换为标准格式：

数据：{validated_data}

要求：
- 金额统一为数字格式（如：1050.00）
- 日期统一为 YYYY-MM-DD 格式
- 地址规范化

输出格式：JSON 格式，包含格式化后的所有字段。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({"validated_data": json.dumps(validated_data, ensure_ascii=False)})
    return {"formatted_data": json.dumps(result, ensure_ascii=False)}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("extract", extract_node)
workflow.add_node("validate", validate_node)
workflow.add_node("format", format_node)

# 定义连线
workflow.set_entry_point("extract")
workflow.add_edge("extract", "validate")
workflow.add_conditional_edges(
    "validate",
    should_revalidate,
    {
        "validate": "validate",  # 继续验证
        "format": "format"       # 格式化
    }
)
workflow.add_edge("format", END)

# 编译
app = workflow.compile()

# 运行
inputs = {"document": "发票内容：张三，北京市朝阳区，壹仟零五十元，2024年1月15日"}
result = app.invoke(inputs)
print(result["formatted_data"])

4. 内容生成流程 📝

场景：自动化创意写作、技术文档生成

提示链：

  1. 根据用户兴趣生成 5 个主题创意
  2. 用户选择或自动选定一个主题
  3. 基于选定主题生成详细大纲
  4. 根据大纲第一点撰写草稿
  5. 根据第二点撰写草稿，并提供前一段上下文，依次完成所有大纲点
  6. 整体审阅并优化草稿的连贯性、语气和语法

流程构建：

from typing import TypedDict, List, Dict
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
import json

llm = ChatOpenAI(temperature=0)

class Topic(BaseModel):
    title: str
    brief_description: str

class OutlineItem(BaseModel):
    section_title: str
    key_points: List[str]

class AgentState(TypedDict):
    user_interests: str
    content_type: str
    topics: List[Dict]
    selected_topic: str
    outline: List[Dict]
    sections: List[str]
    current_section_index: int
    full_draft: str
    optimized_draft: str

# 提示 1：生成主题创意
def generate_topics_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是创意策划专家。根据用户的兴趣和需求，生成 5 个主题创意：

用户兴趣：{user_interests}
内容类型：{content_type}

输出格式：JSON 格式，包含 topics 数组，每个主题包含 title 和 brief_description。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({
        "user_interests": state["user_interests"],
        "content_type": state["content_type"]
    })
    return {"topics": result.get("topics", [])}

# 提示 2：选择主题（这里简化为自动选择第一个）
def select_topic_node(state: AgentState):
    topics = state.get("topics", [])
    if topics:
        selected = topics[0]
        return {"selected_topic": json.dumps(selected, ensure_ascii=False)}
    return {"selected_topic": ""}

# 提示 3：生成大纲
def generate_outline_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是内容结构专家。基于选定的主题，生成详细的内容大纲：

主题：{selected_topic}

输出格式：JSON 格式，包含 outline 数组，每个大纲点包含 section_title 和 key_points。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({"selected_topic": state["selected_topic"]})
    return {
        "outline": result.get("outline", []),
        "sections": [],
        "current_section_index": 0
    }

# 提示 4：分段撰写（循环执行）
def write_section_node(state: AgentState):
    outline = state.get("outline", [])
    current_index = state.get("current_section_index", 0)
    previous_sections = state.get("sections", [])
    
    if current_index >= len(outline):
        # 所有段落已完成，合并为完整草稿
        full_draft = "\n\n".join(previous_sections)
        return {"full_draft": full_draft}
    
    current_section = outline[current_index]
    previous_text = "\n\n".join(previous_sections) if previous_sections else "无"
    
    prompt = ChatPromptTemplate.from_template("""你是内容撰写专家。请根据以下大纲点撰写内容：

大纲点：{section}
前文上下文：{previous_sections}

要求：
- 与前文保持连贯
- 风格一致
- 内容丰富

输出格式：Markdown 格式的完整段落。""")
    chain = prompt | llm | StrOutputParser()
    section_content = chain.invoke({
        "section": json.dumps(current_section, ensure_ascii=False),
        "previous_sections": previous_text
    })
    
    new_sections = previous_sections + [section_content]
    return {
        "sections": new_sections,
        "current_section_index": current_index + 1
    }

# 判断是否继续撰写
def should_continue_writing(state: AgentState) -> str:
    outline = state.get("outline", [])
    current_index = state.get("current_section_index", 0)
    if current_index < len(outline):
        return "write_section"
    else:
        return "optimize"

# 提示 5：整体优化
def optimize_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是内容优化专家。请审阅并优化以下草稿：

完整草稿：{full_draft}

优化方向：
- 连贯性和流畅度
- 语气和风格统一
- 语法和拼写
- 逻辑结构

输出格式：优化后的完整 Markdown 文档。""")
    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({"full_draft": state["full_draft"]})
    return {"optimized_draft": result}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("generate_topics", generate_topics_node)
workflow.add_node("select_topic", select_topic_node)
workflow.add_node("generate_outline", generate_outline_node)
workflow.add_node("write_section", write_section_node)
workflow.add_node("optimize", optimize_node)

# 定义连线
workflow.set_entry_point("generate_topics")
workflow.add_edge("generate_topics", "select_topic")
workflow.add_edge("select_topic", "generate_outline")
workflow.add_edge("generate_outline", "write_section")
workflow.add_conditional_edges(
    "write_section",
    should_continue_writing,
    {
        "write_section": "write_section",
        "optimize": "optimize"
    }
)
workflow.add_edge("optimize", END)

# 编译
app = workflow.compile()

# 运行
inputs = {
    "user_interests": "人工智能和机器学习",
    "content_type": "技术博客"
}
result = app.invoke(inputs)
print(result["optimized_draft"])

5. 有状态对话智能体 💭

场景：多轮对话，需要维护上下文和状态

提示链：

  1. 处理用户发言，识别意图和实体
  2. 更新对话状态（保存关键信息）
  3. 基于当前状态生成回复或识别下一步所需信息
  4. 后续轮次重复，每次新发言启动链式流程，利用累积的对话历史

流程构建：

from typing import TypedDict, List, Dict
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
import json

llm = ChatOpenAI(temperature=0)

class IntentAnalysis(BaseModel):
    intent: str
    entities: Dict[str, str]
    needs_clarification: bool

class AgentState(TypedDict):
    user_message: str
    conversation_history: List[str]
    intent_analysis: Dict
    current_state: Dict
    reply: str

# 提示 1：理解用户意图
def intent_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是对话理解专家。请分析用户的发言，识别意图和实体：

用户发言：{user_message}
对话历史：{conversation_history}

需要识别：
- 用户意图（询问/请求/确认等）
- 关键实体（人名/地点/时间等）
- 需要澄清的信息

输出格式：JSON 格式，包含 intent、entities 和 needs_clarification。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    history_text = "\n".join(state.get("conversation_history", []))
    result = chain.invoke({
        "user_message": state["user_message"],
        "conversation_history": history_text
    })
    return {"intent_analysis": result}

# 提示 2：更新对话状态
def update_state_node(state: AgentState):
    intent_analysis = state.get("intent_analysis", {})
    current_state = state.get("current_state", {})
    
    # 更新状态：保存关键实体和意图
    updated_state = {
        "last_intent": intent_analysis.get("intent", ""),
        "entities": intent_analysis.get("entities", {}),
        "needs_clarification": intent_analysis.get("needs_clarification", False),
        **current_state  # 保留之前的状态
    }
    
    return {"current_state": updated_state}

# 提示 3：生成回复
def reply_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是对话助手。基于以下信息生成回复：

用户意图：{intent}
识别实体：{entities}
对话历史：{conversation_history}
当前状态：{current_state}

要求：
- 保持对话连贯性
- 利用历史上下文
- 如果需要更多信息，礼貌地询问

输出格式：自然流畅的回复文本。""")
    chain = prompt | llm | StrOutputParser()
    history_text = "\n".join(state.get("conversation_history", []))
    result = chain.invoke({
        "intent": state["intent_analysis"].get("intent", ""),
        "entities": json.dumps(state["intent_analysis"].get("entities", {}), ensure_ascii=False),
        "conversation_history": history_text,
        "current_state": json.dumps(state["current_state"], ensure_ascii=False)
    })
    
    # 更新对话历史
    new_history = state.get("conversation_history", []) + [
        f"用户：{state['user_message']}",
        f"助手：{result}"
    ]
    
    return {"reply": result, "conversation_history": new_history}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("intent", intent_node)
workflow.add_node("update_state", update_state_node)
workflow.add_node("reply", reply_node)

# 定义连线
workflow.set_entry_point("intent")
workflow.add_edge("intent", "update_state")
workflow.add_edge("update_state", "reply")
workflow.add_edge("reply", END)

# 编译
app = workflow.compile()

# 运行（第一轮）
inputs = {
    "user_message": "我想了解人工智能",
    "conversation_history": [],
    "current_state": {}
}
result = app.invoke(inputs)
print(result["reply"])

# 运行（第二轮，使用第一轮的历史）
inputs2 = {
    "user_message": "它有哪些应用？",
    "conversation_history": result["conversation_history"],
    "current_state": result["current_state"]
}
result2 = app.invoke(inputs2)
print(result2["reply"])

重点：每轮对话都启动链式流程，利用累积的对话历史（状态），使系统能跨多轮对话保持上下文和连贯性。

6. 代码生成与优化 💻

场景：AI 辅助开发，生成和优化代码

提示链：

  1. 理解用户代码需求，生成伪代码或大纲
  2. 根据大纲撰写初稿代码
  3. 识别代码潜在错误或改进点（可用静态分析工具或再次调用 LLM）
  4. 根据问题重写或优化代码
  5. 补充文档或测试用例

流程构建：

from typing import TypedDict, List, Dict
from pydantic import BaseModel
from langgraph.graph import StateGraph, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
import json

llm = ChatOpenAI(temperature=0)

class RequirementAnalysis(BaseModel):
    description: str
    tech_stack: List[str]
    pseudocode: str

class CodeIssue(BaseModel):
    type: str
    severity: str
    suggestion: str

class AgentState(TypedDict):
    user_requirement: str
    language: str
    requirement_analysis: Dict
    generated_code: str
    review_issues: List[Dict]
    optimized_code: str
    documentation: str

# 提示 1：理解需求
def understand_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是需求分析专家。请分析用户的代码需求：

用户需求：{user_requirement}

输出：
- 功能描述
- 技术栈建议
- 伪代码或算法思路

输出格式：JSON 格式，包含 description、tech_stack 和 pseudocode。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({"user_requirement": state["user_requirement"]})
    return {"requirement_analysis": result}

# 提示 2：生成代码
def generate_code_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是代码生成专家。根据以下需求生成代码：

需求分析：{requirement_analysis}
编程语言：{language}

要求：
- 代码规范
- 包含注释
- 处理边界情况

输出格式：完整的代码文件，包含必要的导入和函数。""")
    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({
        "requirement_analysis": json.dumps(state["requirement_analysis"], ensure_ascii=False),
        "language": state["language"]
    })
    return {"generated_code": result}

# 提示 3：代码审查
def review_code_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是代码审查专家。请审查以下代码，识别潜在问题：

代码：{generated_code}
原始需求：{user_requirement}

检查项：
- 逻辑错误
- 性能问题
- 代码风格
- 安全性

输出格式：JSON 格式，包含 issues 数组，每个问题包含 type、severity 和 suggestion。""")
    parser = JsonOutputParser()
    chain = prompt | llm | parser
    result = chain.invoke({
        "generated_code": state["generated_code"],
        "user_requirement": state["user_requirement"]
    })
    return {"review_issues": result.get("issues", [])}

# 提示 4：优化代码
def optimize_code_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是代码优化专家。请根据审查结果优化代码：

原代码：{generated_code}
审查问题：{review_issues}

输出格式：优化后的完整代码。""")
    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({
        "generated_code": state["generated_code"],
        "review_issues": json.dumps(state["review_issues"], ensure_ascii=False)
    })
    return {"optimized_code": result}

# 提示 5：生成文档和测试
def document_node(state: AgentState):
    prompt = ChatPromptTemplate.from_template("""你是文档专家。请为以下代码生成文档和测试用例：

代码：{optimized_code}

要求：
- API 文档
- 使用示例
- 单元测试用例

输出格式：Markdown 格式的文档和测试代码。""")
    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({"optimized_code": state["optimized_code"]})
    return {"documentation": result}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("understand", understand_node)
workflow.add_node("generate_code", generate_code_node)
workflow.add_node("review_code", review_code_node)
workflow.add_node("optimize_code", optimize_code_node)
workflow.add_node("document", document_node)

# 定义连线
workflow.set_entry_point("understand")
workflow.add_edge("understand", "generate_code")
workflow.add_edge("generate_code", "review_code")
workflow.add_edge("review_code", "optimize_code")
workflow.add_edge("optimize_code", "document")
workflow.add_edge("document", END)

# 编译
app = workflow.compile()

# 运行
inputs = {
    "user_requirement": "实现一个快速排序算法",
    "language": "Python"
}
result = app.invoke(inputs)
print("优化后的代码：")
print(result["optimized_code"])
print("\n文档和测试：")
print(result["documentation"])

注意：以上所有示例都使用 LangGraph 的 StateGraph 来构建提示链，这是更推荐的实践方式，因为它提供了：

  • 状态管理：统一管理整个流程的状态
  • 灵活控制流：支持条件分支和循环
  • 可扩展性：易于添加新节点和修改流程
  • 可调试性：每个节点的输入输出都可以清晰追踪

总结

提示链通过将复杂问题拆解为一系列更简单、易管理的子任务，为引导大语言模型提供了稳健框架。分而治之策略让模型每次专注于单一操作，显著提升输出的可靠性和可控性。

关键要点：

  • 明确步骤目标：每个提示都有清晰、单一的目标
  • 结构化输出：使用 JSON/XML 格式确保数据传递的准确性
  • 角色定位：为每个步骤分配明确的专业角色
  • 合理设计流程：根据任务特点设计合适的链式流程

经验法则：当任务过于复杂、包含多阶段处理、需在步骤间调用外部工具，或需构建多步推理、状态管理的智能体时，建议采用提示链模式。

最近更新:
Contributors: kingmusi