AI 代理的世界正在迅速扩展，掌握 LangChain 和 LangGraph 的提示工程让你站在这场革命的前沿。无论你是打造客服机器人、内容创建系统还是复杂分析工具，这些框架为你创建真正智能的应用提供了基础。

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(cnotallow="我们有个新项目：移动应用开发"),
    AIMessage(cnotallow="好的！已记录移动应用开发项目。时间线是怎样的？"),
    HumanMessage(cnotallow="我们需要在3个月内上线"),
    AIMessage(cnotallow="已记录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(
        respnotallow=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. 客服自动化 

• 基于内容分析的自动工单路由

• 根据客户情绪调整的情感感知回复

• 知识库集成提供即时答案

• 复杂问题的升级工作流

2. 内容生成流水线 

• 网络爬取与事实核查的自动化研究

• 多格式内容（博客、社交媒体、邮件）

• 通过自定义提示保持品牌声音一致性

• 自动审查流程确保质量

3. 数据分析助手 

• 自然语言到 SQL 的转换

• 从原始数据自动生成报告

• 洞察提取与趋势识别

• 基于数据类型的可视化推荐

未来趋势与创新

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

• 处理文本、图像和音频的多模态代理

• 带持久学习能力的长期记忆系统

• 解决复杂问题的协作代理网络

• 针对医疗、金融、法律等行业的领域特定优化

关键收获

✅ 从简单的提示模板开始，逐步增加复杂性

✅ 使用 LangChain 构建线性工作流和简单链

✅ 利用 LangGraph 实现复杂、有状态和条件的工作流

✅ 实现适当的错误处理和回退机制

✅ 考虑内存和上下文以提升用户体验

✅ 针对边缘情况和错误场景进行广泛测试

✅ 监控性能并针对你的特定用例优化

准备好打造你的 AI 代理了吗？

AI 代理的世界正在迅速扩展，掌握 LangChain 和 LangGraph 的提示工程让你站在这场革命的前沿。无论你是打造客服机器人、内容创建系统还是复杂分析工具，这些框架为你创建真正智能的应用提供了基础。

今天就开始实验吧！从简单链开始，逐步增加复杂性，别害怕挑战可能的边界。AI 的未来是代理化的，你现在已经装备好成为其中的一部分！

AI大模型学习福利

作为一名热心肠的互联网老兵，我决定把宝贵的AI知识分享给大家。 至于能学习到多少就看你的学习毅力和能力了 。我已将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

一、全套AGI大模型学习路线

AI大模型时代的学习之旅：从基础到前沿，掌握人工智能的核心技能！

因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获取

二、640套AI大模型报告合集

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。

因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获

三、AI大模型经典PDF籍

随着人工智能技术的飞速发展，AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型，如GPT-3、BERT、XLNet等，以其强大的语言理解和生成能力，正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。

因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获

四、AI大模型商业化落地方案

因篇幅有限，仅展示部分资料，需要点击文章最下方名片即可前往获

作为普通人，入局大模型时代需要持续学习和实践，不断提高自己的技能和认知水平，同时也需要有责任感和伦理意识，为人工智能的健康发展贡献力量