这是一个综合使用LangChain、LangGraph和LangSmith的完整代码示例。这个示例将创建一个智能客服聊天机器人，展示如何将这三个工具结合使用。这个综合示例展示了如何将LangChain、LangGraph和LangSmith三个工具有机结合使用。
## 🔗 LangChain 的使用
**核心组件：**
- `ChatOpenAI`: 提供LLM能力，处理自然语言理解和生成
- `ChatPromptTemplate`: 创建结构化的提示模板，确保一致的输入格式
- `DuckDuckGoSearchRun`: 集成外部搜索能力，获取实时信息
**主要作用：** 提供基础的AI能力和工具链，是整个系统的"大脑"
## 🕸️ LangGraph 的使用
**核心概念：**
- `StateGraph`: 定义有状态的工作流程图
- **节点系统**: 将复杂逻辑分解为独立的处理单元
- **条件边**: 实现智能的流程分支控制
**工作流程：**
1. **意图识别** → 分析用户需求
2. **条件判断** → 决定是否需要搜索
3. **信息搜索** → 获取外部信息（如需要）
4. **响应生成** → 创建个性化回答
5. **质量检查** → 确保回答质量
**主要优势：** 管理复杂的多步骤处理流程，支持动态决策
## 📊 LangSmith 的使用
**监控功能：**
- **自动追踪**: 监控所有LLM调用和工具使用
- **性能指标**: 记录响应时间、token使用量等
- **错误日志**: 捕获和分析系统异常
**调试功能：**
- **完整链路追踪**: 查看每个步骤的详细信息
- **输入输出记录**: 分析数据流转过程
- **元数据标记**: 便于后续分析和优化
## 🚀 实际使用建议
**环境配置：**
```bash
# 安装必要的包
pip install langchain langgraph langsmith openai duckduckgo-search
# 设置环境变量
export OPENAI_API_KEY="your-key"
export LANGCHAIN_API_KEY="your-langsmith-key"
```
**扩展方向：**
1. **添加更多工具**: 集成数据库查询、API调用等
2. **优化工作流**: 根据实际业务需求调整节点逻辑
3. **增强监控**: 设置告警和自动化分析
4. **A/B测试**: 使用LangSmith比较不同版本的性能
这个示例展示了现代AI应用开发的最佳实践：使用LangChain构建AI能力，用LangGraph管理复杂流程，通过LangSmith确保系统的可观测性和持续优化。
``` python
"""
智能客服聊天机器人 - 综合使用LangChain、LangGraph和LangSmith
这个示例展示了如何将三个工具结合使用：
- LangChain: 提供基础的LLM交互和工具链
- LangGraph: 实现复杂的工作流程和状态管理
- LangSmith: 提供监控、调试和优化功能
"""
import os
from typing import Dict, List, Any, Annotated
from datetime import datetime
# LangChain 核心组件
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool
# LangGraph 组件 - 用于构建复杂的工作流
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict
# LangSmith 组件 - 用于监控和调试
from langsmith import Client
from langchain_core.tracers import LangChainTracer
# 配置环境变量（实际使用时需要设置真实的API密钥）
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "customer-service-bot"
class CustomerServiceBot:
"""
智能客服机器人类
整合了LangChain的LLM能力、LangGraph的工作流管理和LangSmith的监控功能
"""
def __init__(self):
# 初始化LangChain组件
self.llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.7,
# LangSmith会自动追踪这个LLM的调用
)
# 初始化搜索工具（LangChain工具）
self.search_tool = DuckDuckGoSearchRun()
# 初始化LangSmith客户端用于监控
self.langsmith_client = Client()
# 创建LangGraph工作流
self.workflow = self._create_workflow()
# 编译工作流为可执行的图
self.app = self.workflow.compile()
def _create_workflow(self):
"""
使用LangGraph创建客服机器人的工作流程
工作流程包括：
1. 意图识别
2. 信息搜索（如需要）
3. 响应生成
4. 质量检查
"""
# 定义状态结构 - LangGraph的核心概念
class State(TypedDict):
messages: Annotated[List, add_messages] # 对话历史
intent: str # 用户意图
search_results: str # 搜索结果
confidence: float # 响应置信度
needs_search: bool # 是否需要搜索
# 创建状态图
workflow = StateGraph(State)
# 添加节点（每个节点是一个处理步骤）
workflow.add_node("classify_intent", self._classify_intent)
workflow.add_node("search_information", self._search_information)
workflow.add_node("generate_response", self._generate_response)
workflow.add_node("quality_check", self._quality_check)
# 设置入口点
workflow.set_entry_point("classify_intent")
# 添加边（定义节点之间的流转逻辑）
workflow.add_conditional_edges(
"classify_intent",
self._should_search, # 条件函数
{
"search": "search_information", # 需要搜索
"direct": "generate_response" # 直接回答
}
)
workflow.add_edge("search_information", "generate_response")
workflow.add_edge("generate_response", "quality_check")
# 添加条件结束边
workflow.add_conditional_edges(
"quality_check",
self._should_regenerate,
{
"regenerate": "generate_response", # 重新生成
"end": END # 结束
}
)
return workflow
def _classify_intent(self, state: Dict) -> Dict:
"""
意图识别节点 - 使用LangChain的提示模板和LLM
"""
# 创建意图分类提示模板（LangChain组件）
intent_prompt = ChatPromptTemplate.from_messages([
("system", """你是一个客服意图分类器。请分析用户消息并分类为以下之一：
- product_inquiry: 产品咨询
- technical_support: 技术支持
- complaint: 投诉
- general_question: 一般问题
- order_status: 订单状态
只返回分类结果，不要其他内容。"""),
("human", "{message}")
])
# 获取最新的用户消息
last_message = state["messages"][-1].content
# 使用LangChain的链式调用
chain = intent_prompt | self.llm
# LangSmith会自动追踪这个调用
intent_result = chain.invoke({"message": last_message})
intent = intent_result.content.strip()
# 判断是否需要搜索外部信息
needs_search = intent in ["product_inquiry", "technical_support", "general_question"]
return {
"intent": intent,
"needs_search": needs_search
}
def _should_search(self, state: Dict) -> str:
"""
LangGraph条件函数 - 决定是否需要搜索
"""
return "search" if state["needs_search"] else "direct"
def _search_information(self, state: Dict) -> Dict:
"""
信息搜索节点 - 使用LangChain工具
"""
last_message = state["messages"][-1].content
# 使用LangChain的搜索工具
try:
# 构造搜索查询
search_query = f"客服 {state['intent']} {last_message}"
# LangSmith会追踪工具的使用
search_results = self.search_tool.run(search_query)
return {"search_results": search_results}
except Exception as e:
print(f"搜索出错: {e}")
return {"search_results": "搜索暂时不可用"}
def _generate_response(self, state: Dict) -> Dict:
"""
响应生成节点 - 核心的LangChain LLM调用
"""
# 构建响应生成提示
system_message = """你是一个专业的客服代表。请根据以下信息回答用户问题：
用户意图: {intent}
搜索结果: {search_results}
要求：
5. 回答要专业、友好、有帮助
6. 如果有搜索结果，要结合使用
7. 如果无法解决问题，要诚实说明并提供替代方案
"""
# 创建消息列表
messages = [
SystemMessage(content=system_message.format(
intent=state.get("intent", "未知"),
search_results=state.get("search_results", "无")
))
]
# 添加对话历史
messages.extend(state["messages"])
# 生成响应（LangSmith自动追踪）
response = self.llm.invoke(messages)
# 简单的置信度计算
confidence = 0.8 if state.get("search_results") else 0.6
return {
"messages": [response],
"confidence": confidence
}
def _quality_check(self, state: Dict) -> Dict:
"""
质量检查节点 - 确保响应质量
"""
# 获取最新的AI响应
last_ai_message = state["messages"][-1].content
# 简单的质量检查逻辑
quality_indicators = [
len(last_ai_message) > 20, # 响应长度合理
"抱歉" not in last_ai_message or "帮助" in last_ai_message, # 积极响应
state["confidence"] > 0.5 # 置信度足够
]
quality_score = sum(quality_indicators) / len(quality_indicators)
return {"confidence": quality_score}
def _should_regenerate(self, state: Dict) -> str:
"""
LangGraph条件函数 - 决定是否需要重新生成响应
"""
return "end" if state["confidence"] > 0.6 else "regenerate"
def chat(self, user_message: str) -> str:
"""
主要的聊天接口
Args:
user_message: 用户输入的消息
Returns:
AI生成的响应
"""
# 使用LangSmith创建一个追踪会话
with self.langsmith_client.trace(
name="customer_service_chat",
inputs={"user_message": user_message},
tags=["customer_service", "production"]
) as trace:
try:
# 初始化状态
initial_state = {
"messages": [HumanMessage(content=user_message)],
"intent": "",
"search_results": "",
"confidence": 0.0,
"needs_search": False
}
# 运行LangGraph工作流
final_state = self.app.invoke(initial_state)
# 获取最终响应
ai_response = final_state["messages"][-1].content
# 记录成功的追踪信息
trace.update(
outputs={"ai_response": ai_response},
metadata={
"intent": final_state["intent"],
"confidence": final_state["confidence"],
"timestamp": datetime.now().isoformat()
}
)
return ai_response
except Exception as e:
# 记录错误信息到LangSmith
trace.update(
error=str(e),
outputs={"error": "处理请求时发生错误"}
)
return "抱歉，我现在无法处理您的请求，请稍后再试。"
def main():
"""
主函数 - 演示如何使用这个综合的客服机器人
"""
print("=== LangChain + LangGraph + LangSmith 综合示例 ===")
print("智能客服机器人启动中...")
# 创建客服机器人实例
bot = CustomerServiceBot()
# 测试对话
test_messages = [
"你好，我想了解一下你们的产品",
"我的订单什么时候能到？",
"产品出现了故障，怎么办？",
"你们的退换货政策是什么？"
]
print("\n开始测试对话:")
print("-" * 50)
for i, message in enumerate(test_messages, 1):
print(f"\n[测试 {i}]")
print(f"用户: {message}")
# 调用聊天机器人
response = bot.chat(message)
print(f"客服: {response}")
print("-" * 30)
print(f"\n✅ 测试完成！")
print(f"📊 您可以在LangSmith仪表板中查看详细的追踪信息:")
print(f" - 每次对话的完整流程")
print(f" - LLM调用的token使用情况")
print(f" - 工具调用的性能指标")
print(f" - 错误日志和调试信息")
if __name__ == "__main__":
main()
"""
代码结构说明：
1. LangChain 使用:
- ChatOpenAI: LLM核心引擎
- ChatPromptTemplate: 结构化提示模板
- DuckDuckGoSearchRun: 外部信息搜索工具
2. LangGraph 使用:
- StateGraph: 定义复杂的工作流程
- 节点系统: 将处理逻辑分解为独立节点
- 条件边: 实现动态的流程控制
3. LangSmith 使用:
- 自动追踪: 监控所有LLM和工具调用
- 手动追踪: 创建自定义的追踪会话
- 元数据记录: 保存重要的上下文信息
主要优势：
- 模块化设计，易于维护和扩展
- 完整的监控和调试能力
- 复杂的工作流程管理
- 自动的性能优化建议
实际部署时，记得：
1. 设置正确的API密钥
2. 配置LangSmith项目
3. 根据需要调整工作流程
4. 监控生产环境的性能指标
"""
```