从LangChain到LangGraph：AI智能体提示词工程的系统化学习

AI 的世界正在飞速演变，从简单的问答系统升级成了复杂、多步骤推理的智能代理。不管你是想打造客服机器人、数据分析工具，还是复杂的自动化工作流程，掌握 LangChain 和 LangGraph 的提示工程（Prompt Engineering）是你成功的关键！

🎯 为什么提示工程比以往任何时候都重要
传统的提示方式对简单任务还行，但现代 AI 应用需要：

• • 多步骤推理 🧠
• • 动态决策 ⚡
• • 记忆与上下文感知 🧐
• • 工具集成 🔧
• • 错误处理与自我纠正 🔄

下面我们来聊聊 LangChain 和 LangGraph 怎么让这一切变成现实！

🏗️ 基础：LangChain 提示模板

1. 字符串模板 —— 基础入门 📝

适合简单、单一输入的场景：

from langchain_core.prompts import PromptTemplate

# 客服邮件生成器
email_template = PromptTemplate.from_template(
    """
    为以下情况撰写一封专业的客服邮件：
    
    客户问题：{issue}
    客户姓名：{customer_name}
    紧急程度：{urgency}
    
    语气：共情且以解决方案为导向
    包含：问题确认、解决方案步骤、后续跟进提议
    """
)
# 生成个性化回复
prompt = email_template.invoke({
    "issue": "支付后无法访问高级功能",
    "customer_name": "Sarah",
    "urgency": "High"
})
print(prompt.to_string())

小贴士 💡：在模板中使用描述性的变量名和清晰的指令！

2. 聊天模板 —— 对话的超能力 💬

适合打造复杂的聊天体验：

from langchain_core.prompts import ChatPromptTemplate

# AI 编程导师设置
coding_mentor_template = ChatPromptTemplate.from_messages([
    ("system", """你是 Python 编程专家，你的教学风格是：
    • 耐心且鼓励
    • 提供清晰的解释和示例
    • 提出引导性问题帮助学生思考
    • 庆祝学生的进步和学习时刻
    """),
    ("user", "我在 {topic} 上有困难，能帮我理解 {specific_question} 吗？")
])
# 创建学习会话
chat_prompt = coding_mentor_template.invoke({
    "topic": "列表推导式",
    "specific_question": "什么时候用它，什么时候用普通循环"
})

3. 使用 MessagesPlaceholder 实现动态消息历史 📚

对保持对话上下文至关重要：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

# 带记忆的项目管理助手
project_template = ChatPromptTemplate.from_messages([
    ("system", "你是项目管理助手，跟踪任务、截止日期和团队进展。"),
    MessagesPlaceholder(variable_name="conversation_history"),
    ("user", "{current_request}")
])
# 模拟对话历史
conversation_history = [
    HumanMessage(content="我们有个新项目：移动应用开发"),
    AIMessage(content="好的！已记录移动应用开发项目。时间线是怎样的？"),
    HumanMessage(content="我们需要在3个月内上线"),
    AIMessage(content="已记录3个月时间线。关键里程碑有哪些？")
]
# 继续对话
current_prompt = project_template.invoke({
    "conversation_history": conversation_history,
    "current_request": "添加任务：设计用户界面 mockups，下周五截止"
})

⛓️ 使用 LCEL 构建智能链

LangChain Expression Language (LCEL) 让你能创建强大的处理流水线：

简单链示例 🔗

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 内容创作流水线
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
output_parser = StrOutputParser()
# 博客文章生成链
blog_chain = (
    {"topic": lambda x: x, "audience": lambda x: "developers"}
    | PromptTemplate.from_template(
        "为 {audience} 撰写一篇关于 {topic} 的吸引人博客文章，"
        "包含实用示例和可操作的建议。"
    )
    | llm
    | output_parser
)
result = blog_chain.invoke("API Rate Limiting Best Practices")
print(result)

复杂多步骤分析 🔍

from langchain_core.runnables import RunnablePassthrough

# 数据分析工作流
defextract_metrics(data):
    returnf"从 {data[:100]} 中提取的关键指标..."
defgenerate_insights(context):
    returnf"基于 {context['metrics']} 的洞察：[分析内容]"
# 多步骤分析链
analysis_chain = (
    RunnablePassthrough.assign(metrics=extract_metrics)
    | RunnablePassthrough.assign(insights=generate_insights)
    | PromptTemplate.from_template(
        "创建执行摘要报告：\n"
        "数据：{input}\n"
        "指标：{metrics}\n"
        "洞察：{insights}\n"
        "格式为专业商务报告。"
    )
    | llm
    | output_parser
)

🕸️ 使用 LangGraph 实现高级编排

LangGraph 超越线性链，带来智能、有状态的工作流：

状态管理 📊

from typing import TypedDict, List
from langgraph.graph import StateGraph, END

classCustomerSupportState(TypedDict):
    customer_message: str
    sentiment: str
    category: str
    priority: str
    resolution_steps: List[str]
    escalated: bool

# 定义工作流节点
defanalyze_sentiment(state: CustomerSupportState):
    # 情感分析逻辑
    message = state["customer_message"]
    # 简化的情感检测
    if"angry"in message.lower() or"frustrated"in message.lower():
        sentiment = "negative"
    elif"happy"in message.lower() or"great"in message.lower():
        sentiment = "positive"
    else:
        sentiment = "neutral"
    
    return {"sentiment": sentiment}

defcategorize_issue(state: CustomerSupportState):
    message = state["customer_message"].lower()
    if"billing"in message or"payment"in message:
        category = "billing"
    elif"technical"in message or"bug"in message:
        category = "technical"
    else:
        category = "general"
    
    return {"category": category}

defdetermine_priority(state: CustomerSupportState):
    if state["sentiment"] == "negative"and state["category"] == "billing":
        priority = "high"
    elif state["category"] == "technical":
        priority = "medium"
    else:
        priority = "low"
    
    return {"priority": priority}

条件逻辑与路由 🎯

def should_escalate(state: CustomerSupportState):
    """决定是否需要升级工单"""
    if state["priority"] == "high"or state["sentiment"] == "negative":
        return"escalate"
    else:
        return"resolve"

defescalate_ticket(state: CustomerSupportState):
    return {
        "escalated": True,
        "resolution_steps": ["已升级至高级支持团队", "经理将在1小时内联系"]
    }

defauto_resolve(state: CustomerSupportState):
    steps = {
        "billing": ["检查支付状态", "发送账单说明", "提供支付计划"],
        "technical": ["收集系统信息", "应用常见修复", "安排后续跟进"],
        "general": ["提供相关文档", "提供额外协助"]
    }
    
    return {
        "escalated": False,
        "resolution_steps": steps.get(state["category"], ["已提供一般协助"])
    }

# 构建工作流图
workflow = StateGraph(CustomerSupportState)
# 添加节点
workflow.add_node("analyze_sentiment", analyze_sentiment)
workflow.add_node("categorize_issue", categorize_issue)
workflow.add_node("determine_priority", determine_priority)
workflow.add_node("escalate_ticket", escalate_ticket)
workflow.add_node("auto_resolve", auto_resolve)
# 定义流程
workflow.set_entry_point("analyze_sentiment")
workflow.add_edge("analyze_sentiment", "categorize_issue")
workflow.add_edge("categorize_issue", "determine_priority")
# 条件路由
workflow.add_conditional_edges(
    "determine_priority",
    should_escalate,
    {
        "escalate": "escalate_ticket",
        "resolve": "auto_resolve"
    }
)
workflow.add_edge("escalate_ticket", END)
workflow.add_edge("auto_resolve", END)
# 编译并运行
app = workflow.compile()

🤖 打造智能 AI 代理

ReAct 框架实战 🎭

from langchain.tools import Tool
from langchain.Agents import create_react_agent

# 定义自定义工具
@tool
defsearch_knowledge_base(query: str) -> str:
    """搜索内部知识库获取信息"""
    knowledge_db = {
        "api limits": "标准计划：每小时1000次请求，高级计划：每小时10000次",
        "password reset": "使用忘记密码链接，检查邮件，遵循说明",
        "billing cycle": "按注册日期月度计费，提供年度折扣"
    }
    
    for key, value in knowledge_db.items():
        if key in query.lower():
            return value
    return"知识库中未找到信息"

@tool
defcreate_support_ticket(issue: str, priority: str) -> str:
    """为复杂问题创建支持工单"""
    ticket_id = f"TICKET-{hash(issue) % 10000}"
    returnf"支持工单 {ticket_id} 已创建，优先级：{priority}"

@tool
defsend_email_notification(recipient: str, subject: str, message: str) -> str:
    """向客户发送邮件通知"""
    returnf"邮件已发送至 {recipient}，主题：{subject}"

# 创建带工具的代理
tools = [search_knowledge_base, create_support_ticket, send_email_notification]
agent_prompt = """
你是一个乐于助人的客服代理，你可以：
1. 搜索知识库获取答案
2. 为复杂问题创建支持工单
3. 向客户发送邮件通知
始终保持帮助性和专业性，必要时使用工具。
客户问题：{input}
{agent_scratchpad}
"""

# 代理执行示例
defrun_support_agent(customer_query: str):
    # 这里将与实际代理执行整合
    print(f"正在处理：{customer_query}")
    print("代理思考中...")
    print("工具使用：search_knowledge_base")
    print("回复已生成！")

多代理协作 🤝

class MultiAgentSystem:
    def__init__(self):
        self.agents = {
            "researcher": self.create_researcher_agent(),
            "writer": self.create_writer_agent(),
            "reviewer": self.create_reviewer_agent()
        }
    
    defcreate_researcher_agent(self):
        return {
            "role": "研究专家",
            "tools": ["web_search", "data_analysis", "fact_checking"],
            "prompt": "收集给定主题的全面信息"
        }
    
    defcreate_writer_agent(self):
        return {
            "role": "内容创作者",
            "tools": ["content_generation", "style_formatting"],
            "prompt": "基于研究结果创建吸引人的内容"
        }
    
    defcreate_reviewer_agent(self):
        return {
            "role": "质量保证",
            "tools": ["grammar_check", "fact_verification", "style_review"],
            "prompt": "审查并提升内容质量"
        }
    
    defcoordinate_workflow(self, task: str):
        """协调多代理协作"""
        # 研究阶段
        research_data = self.agents["researcher"]["process"](task)
        
        # 写作阶段
        draft_content = self.agents["writer"]["process"](research_data)
        
        # 审查阶段
        final_content = self.agents["reviewer"]["process"](draft_content)
        
        return final_content

# 使用示例
multi_agent = MultiAgentSystem()
result = multi_agent.coordinate_workflow("撰写关于 AI 伦理的综合指南")

🧠 高级提示技巧

1. 链式思维提示（Chain-of-Thought Prompting） 🔗

cot_template = PromptTemplate.from_template("""
一步步解决这个问题：
问题：{problem}
让我仔细思考：
步骤1：理解问题要求
步骤2：识别关键信息
步骤3：应用相关原则/公式
步骤4：计算/推理解决方案
步骤5：验证答案是否合理
解决方案：
""")

2. 少样本学习示例（Few-Shot Learning Examples） 📚

few_shot_template = PromptTemplate.from_template("""
将以下客户反馈分类为 Positive、Negative 或 Neutral：
示例：
输入："产品很好用，运输也很快！"
输出：Positive
输入："客服很差，非常失望"
输出：Negative
输入："产品按时到达，符合预期"
输出：Neutral
现在分类这个：
输入：{feedback}
输出：
""")

3. 自我纠正循环 🔄

def self_correcting_agent(initial_response: str, validation_criteria: str):
    """实现自我纠正机制"""
    
    correction_template = PromptTemplate.from_template("""
    原始回复：{response}
    
    验证标准：{criteria}
    
    根据标准审查你的回复。如果需要改进：
    1. 识别具体问题
    2. 提供修正版本
    3. 解释改进的内容
    
    最终回复：
    """)
    
    return correction_template.format(
        response=initial_response,
        criteria=validation_criteria
    )

🚀 生产环境最佳实践

错误处理与韧性 🛡️

from langchain_core.runnables import RunnableLambda
import logging

defsafe_llm_call(prompt: str, fallback_response: str = "抱歉，我在处理你的请求时遇到问题。"):
    """安全的 LLM 调用包装器，带错误处理"""
    try:
        # 这里是你的 LLM 调用
        response = llm.invoke(prompt)
        return response
    except Exception as e:
        logging.error(f"LLM 调用失败：{e}")
        return fallback_response

# 带回退机制的韧性链
resilient_chain = (
    prompt_template
    | RunnableLambda(safe_llm_call)
    | output_parser
)

内存管理 💾

from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory

# 持久化对话内存
def create_persistent_memory(session_id: str):
    return ConversationBufferWindowMemory(
        chat_memory=RedisChatMessageHistory(
            session_id=session_id,
            url="redis://localhost:6379"
        ),
        memory_key="chat_history",
        k=10  # 保留最近10次交流
    )

性能优化 ⚡

import asyncio
from langchain_core.runnables import RunnableParallel

asyncdefparallel_processing_example():
    """并发处理多个任务"""
    
    parallel_chain = RunnableParallel(
        summary=summarization_chain,
        sentiment=sentiment_chain,
        keywords=keyword_extraction_chain
    )
    
    results = await parallel_chain.ainvoke({"text": document})
    return results

# 缓存重复查询
from functools import lru_cache
@lru_cache(maxsize=1000)
defcached_llm_call(prompt_hash: str):
    """缓存相同提示的 LLM 回复"""
    return llm.invoke(prompt_hash)

🎯 现实世界的应用

1. 1. 客服自动化 📞

• • 基于内容分析的自动工单路由
• • 根据客户情绪调整的情感感知回复
• • 知识库集成提供即时答案
• • 复杂问题的升级工作流

1. 2. 内容生成流水线 ✍️

• • 网络爬取与事实核查的自动化研究
• • 多格式内容（博客、社交媒体、邮件）
• • 通过自定义提示保持品牌声音一致性
• • 自动审查流程确保质量

1. 3. 数据分析助手 📊

• • 自然语言到 SQL 的转换
• • 从原始数据自动生成报告
• • 洞察提取与趋势识别
• • 基于数据类型的可视化推荐

🔮 未来趋势与创新

AI 代理的未来超级激动人心：

• • 处理文本、图像和音频的多模态代理
• • 带持久学习能力的长期记忆系统
• • 解决复杂问题的协作代理网络
• • 针对医疗、金融、法律等行业的领域特定优化

🎉 关键收获

✅ 从简单的提示模板开始，逐步增加复杂性
✅ 使用 LangChain 构建线性工作流和简单链
✅ 利用 LangGraph 实现复杂、有状态和条件的工作流
✅ 实现适当的错误处理和回退机制
✅ 考虑内存和上下文以提升用户体验
✅ 针对边缘情况和错误场景进行广泛测试
✅ 监控性能并针对你的特定用例优化