实战项目
更新: 4/9/2026 字数: 0 字 时长: 0 分钟
本章通过完整的实战项目，展示 LangChain Agent 的综合应用能力。
项目一：智能客服
项目概述
智能客服系统
────────────────────────────────────────────────────────
功能：
1. 意图识别
2. 问题分类
3. 自动回复
4. 工单创建
5. 人工转接
架构：LangGraph StateGraph完整代码
python
"""
智能客服系统
"""
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, MessagesState, END, START
from langchain_core.messages import HumanMessage, AIMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
# FAQ 数据库
FAQ_DB = {
"退货": "您好，关于退货政策：\n1. 7天内可申请退货\n2. 商品需保持原包装",
"退款": "退款说明：\n1. 退款将在1-3个工作日处理",
"运费": "运费说明：\n1. 订单满99元免运费",
}
def search_faq(query: str) -> str:
"""搜索 FAQ"""
for keyword, answer in FAQ_DB.items():
if keyword in query:
return answer
return None
def create_ticket(title: str, description: str, priority: str) -> str:
"""创建工单"""
import random
ticket_id = f"TK{random.randint(10000, 99999)}"
return f"✅ 工单已创建\n\n工单编号：{ticket_id}\n标题：{title}"
class CustomerServiceState(TypedDict):
messages: list
intent: str | None
faq_answer: str | None
ticket_id: str | None
def classify_intent(state: CustomerServiceState) -> CustomerServiceState:
"""意图识别"""
user_msg = state["messages"][-1]["content"]
if any(kw in user_msg for kw in ["退货", "退款", "运费"]):
intent = "faq"
elif any(kw in user_msg for kw in ["投诉", "问题"]):
intent = "ticket"
else:
intent = "general"
return {"intent": intent}
def handle_faq(state: CustomerServiceState) -> CustomerServiceState:
answer = search_faq(state["messages"][-1]["content"])
return {
"faq_answer": answer,
"messages": state["messages"] + [AIMessage(content=answer or "未找到相关答案")]
}
def route_after_classify(state: CustomerServiceState) -> str:
intent = state.get("intent", "general")
return f"{intent}_handler"
graph = StateGraph(CustomerServiceState)
graph.add_node("classify", classify_intent)
graph.add_node("faq_handler", handle_faq)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
"classify",
route_after_classify,
{"faq_handler": "faq_handler", "ticket_handler": END, "general_handler": END}
)
graph.add_edge("faq_handler", END)
customer_service = graph.compile(checkpointer=InMemorySaver())
if __name__ == "__main__":
config = {"configurable": {"thread_id": "cs_session_001"}}
result = customer_service.invoke(
{"messages": [{"role": "user", "content": "我想退货"}]},
config=config
)
print(result["messages"][-1]["content"])项目二：个人助手
项目概述
个人 AI 助手
────────────────────────────────────────────────────────
功能：
1. 日程管理
2. 邮件处理
3. 任务管理
4. 信息查询
5. 对话记忆完整代码
python
"""
个人 AI 助手
"""
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
CALENDAR_DB = []
TASK_DB = []
def get_calendar(date: str = "today") -> str:
"""查看日历"""
if not CALENDAR_DB:
return "今天的日程是空的"
events = [e for e in CALENDAR_DB if e["date"] == date]
return "\n".join([f"• {e['time']} - {e['title']}" for e in events])
def add_calendar_event(title: str, time: str, duration: int = 60) -> str:
"""添加日历事件"""
CALENDAR_DB.append({"title": title, "date": time.split()[0], "time": time})
return f"✅ 已添加日程：{title}\n📅 {time}"
def get_tasks(status: str = "all") -> str:
"""查看任务"""
tasks = TASK_DB if status == "all" else [t for t in TASK_DB if t["status"] == status]
if not tasks:
return "没有任务"
return "\n".join([f"• {t['title']} ({t['status']})" for t in tasks])
personal_assistant = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_calendar, add_calendar_event, get_tasks],
system_prompt="""你是一个专业的个人 AI 助手。
你的能力：
1. 📅 日程管理 - 查看和添加日历事件
2. 📋 任务管理 - 查看待办任务
3. 🔍 信息查询 - 查询天气、时间等
沟通风格：
- 简洁专业
- 使用表情符号增加可读性""",
checkpointer=InMemorySaver()
)
if __name__ == "__main__":
config = {"configurable": {"thread_id": "assistant_001"}}
result = personal_assistant.invoke(
{"messages": [{"role": "user", "content": "今天有什么日程？"}]},
config=config
)
print(result["messages"][-1]["content"])项目三：研究助手
项目概述
AI 研究助手
────────────────────────────────────────────────────────
功能：
1. 主题研究
2. 论文搜索
3. 内容总结
4. 生成报告完整代码
python
"""
AI 研究助手
"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
def search_arxiv(query: str) -> str:
"""搜索学术论文"""
return f"""学术搜索结果：
1. **Attention Is All You Need** (Vaswani et al., 2017)
- Transformer 架构的开创性论文
2. **BERT: Pre-training of Deep Bidirectional Transformers**
- Google, 2018"""
def summarize(text: str) -> str:
"""总结文本"""
return f"[总结]\n这是对原文的总结..."
class ResearchState(TypedDict):
topic: str
search_results: dict
summaries: dict
draft: str
current_step: str
def search_node(state: ResearchState) -> ResearchState:
topic = state["topic"]
arxiv_results = search_arxiv(topic)
return {
"search_results": {"arxiv": arxiv_results},
"current_step": "search_complete"
}
def summarize_node(state: ResearchState) -> ResearchState:
results = state.get("search_results", {})
summary = summarize(results.get("arxiv", ""))
return {
"summaries": {"arxiv": summary},
"current_step": "summarize_complete"
}
def draft_node(state: ResearchState) -> ResearchState:
summaries = state.get("summaries", {})
draft = f"""# {state['topic']} 研究报告
## 摘要
{summaries.get('arxiv', 'N/A')}
[完整报告内容...]
"""
return {"draft": draft, "current_step": "draft_complete"}
def route_research(state: ResearchState) -> str:
step = state.get("current_step", "")
routes = {
"search_complete": "summarize",
"summarize_complete": "draft",
"draft_complete": END
}
return routes.get(step, END)
graph = StateGraph(ResearchState)
graph.add_node("search", search_node)
graph.add_node("summarize", summarize_node)
graph.add_node("draft", draft_node)
graph.add_edge(START, "search")
graph.add_conditional_edges("search", route_research, {"summarize": "summarize", END: END})
graph.add_conditional_edges("summarize", route_research, {"draft": "draft", END: END})
graph.add_conditional_edges("draft", route_research, {END: END})
research_app = graph.compile(checkpointer=InMemorySaver())
if __name__ == "__main__":
result = research_app.invoke({
"topic": "Transformer 架构",
"search_results": {},
"summaries": {},
"draft": "",
"current_step": ""
})
print(result.get("draft", ""))本章小结
实战项目要点
────────────────────────────────────────────────────────
✅ 智能客服 = 意图识别 + FAQ + 工单
✅ 个人助手 = 日程 + 任务 + 记忆
✅ 研究助手 = 搜索 + 总结 + 报告
项目构建模式
────────────────────────────────────────────────────────
1. 定义状态 (State)
2. 实现工具 (Tools)
3. 创建节点 (Nodes)
4. 配置边 (Edges)
5. 编译执行 (Compile)