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)
🎯 现实世界的应用
🔮 未来趋势与创新
AI 代理的未来超级激动人心：
🎉 关键收获
✅ 从简单的提示模板开始，逐步增加复杂性
✅ 使用 LangChain 构建线性工作流和简单链
✅ 利用 LangGraph 实现复杂、有状态和条件的工作流
✅ 实现适当的错误处理和回退机制
✅ 考虑内存和上下文以提升用户体验
✅ 针对边缘情况和错误场景进行广泛测试
✅ 监控性能并针对你的特定用例优化
粤ICP备14082021号
免费POC，
零成本试错
