本文详细介绍了使用LangChain和LangGraph进行提示工程，构建AI智能代理的核心技术。涵盖基础模板、智能链构建、高级编排、多代理协作及生产实践，通过代码示例展示从简单客服到复杂分析工具的开发方法，帮助开发者掌握现代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}

条件逻辑与路由

defshould_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("回复已生成！")

多代理协作

classMultiAgentSystem:
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. 自我纠正循环

defself_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

# 持久化对话内存
defcreate_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 代理的世界正在迅速扩展，掌握 LangChain 和 LangGraph 的提示工程让你站在这场革命的前沿。无论你是打造客服机器人、内容创建系统还是复杂分析工具，这些框架为你创建真正智能的应用提供了基础。

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

---

如何系统学习掌握AI大模型？

AI大模型作为人工智能领域的重要技术突破，正成为推动各行各业创新和转型的关键力量。抓住AI大模型的风口，掌握AI大模型的知识和技能将变得越来越重要。

学习AI大模型是一个系统的过程，需要从基础开始，逐步深入到更高级的技术。

这里给大家精心整理了一份全面的AI大模型学习资源，包括：AI大模型全套学习路线图（从入门到实战）、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等，资料免费分享！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

5. 大模型项目实战

学以致用 ，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

全套的AI大模型学习资源已经整理打包，有需要的小伙伴可以微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】