LangGraph--聊天机器人构建（3）-育师

在人工智能快速发展的今天，智能聊天机器人已经不仅仅是问答工具，它们正在向多轮对话、知识库检索和工具调用的方向升级。本篇文章将系统讲解如何搭建一个完整的智能聊天机器人，涵盖多轮上下文记忆、RAG检索、以及计算器工具调用，并提供可直接运行的示例代码。

1. 核心架构设计理念

一个完整的智能聊天机器人需要解决三个核心问题：

1）多轮对话上下文记忆

聊天机器人最基本的能力是理解上下文。我们通过State.history来记录对话历史，每轮生成回答前，将历史内容拼接进 prompt，让 LLM 知道前文发生了什么。

2）RAG 文档检索机制

智能机器人必须具备知识库能力，以便回答专业问题或动态信息查询。我们引入RAG（Retrieval-Augmented Generation）流程：

嵌入模型：将文本转成向量
向量数据库：使用 FAISS / Chroma 存储向量
检索节点：根据用户问题匹配相关文档
LLM增强：将检索结果传入 LLM 生成最终回答

3）工具调用：计算器

智能机器人不仅回答问题，还能进行计算。我们设计了一个自动识别数学表达式并调用计算器的流程：

检测用户输入是否为数学表达式
调用安全计算器获取结果
将计算结果传给 LLM 生成最终回答

2. 定义对话状态（State）

我们用TypedDict定义状态对象，记录必要信息：

fromtypingimportTypedDict,ListclassState(TypedDict):question:str# 当前用户问题answer:str# 最终回答session_id:str# 会话 IDhistory:List[str]# 对话历史retrieved_docs:str# RAG 检索结果文本tool_result:str# 工具计算结果

通过State，每轮对话的信息都能被记录并传递给下一个节点。

3. 嵌入与向量检索

3.1 文本向量化（Embedding）

这里我们使用硅基流动（SiliconFlow）的中文大模型bge-large-zh-v1.5：

importrequests EMBED_URL="https://api.siliconflow.cn/v1/embeddings"EMBED_MODEL="BAAI/bge-large-zh-v1.5"defembed_text(text:str):payload={"input":text,"model":EMBED_MODEL}headers={"Authorization":f"Bearer{API_KEY}"}res=requests.post(EMBED_URL,json=payload,headers=headers)returnres.json()["data"][0]["embedding"]

3.2 构建向量库

使用 FAISS 存储和检索向量：

importfaissimportnumpyasnpclassVectorStore:def__init__(self):self.index=faiss.IndexFlatL2(1024)# bge-zh 大模型维度1024self.texts=[]defadd(self,text:str):vec=np.array([embed_text(text)],dtype="float32")self.index.add(vec)self.texts.append(text)defsearch(self,query:str,topk=3):qvec=np.array([embed_text(query)],dtype="float32")D,I=self.index.search(qvec,topk)results=[self.texts[i]foriinI[0]]return"\n".join(results)

初始化向量库：

vs=VectorStore()vs.add("太阳能发电的基本原理是…")vs.add("风电机组的主要部件包括…")vs.add("深度学习模型训练需要…")

4. 安全计算器工具

defcalculator(expression:str)->str:importreifnotre.match(r"^[0-9+\-*/(). ]+$",expression):return"表达式不合法"try:result=eval(expression)returnstr(result)except:return"计算失败"

注意：这里只允许基本运算符，防止任意代码执行风险。

5. LangGraph 工作流节点设计

我们把对话流程拆成三个节点：

retrieve：知识库检索
tool：计算器工具
answer：整合历史、RAG 和工具结果生成最终回答

5.1 检索节点：retrieve

defretrieve(state:State):query=state["question"]docs=vs.search(query)state["retrieved_docs"]=docsreturn{"retrieved_docs":docs}

5.2 工具节点：tool

importredeftool(state:State):q=state["question"]math_pattern=r"^[0-9+\-*/(). ]+$"ifre.match(math_pattern,q):result=calculator(q)state["tool_result"]=resultreturn{"tool_result":result}return{"tool_result":""}

5.3 最终回答节点：answer

结合历史对话、RAG 文档和工具结果：

MAX_TURNS=10deftrim_history(history):returnhistory[-MAX_TURNS*2:]iflen(history)>MAX_TURNS*2elsehistorydefanswer(state:State):history=trim_history(state.get("history",[]))rag=state.get("retrieved_docs","")tool_result=state.get("tool_result","")prompt="你是一个智能助手，请基于对话历史、知识库检索结果和工具结果回答。\n\n"ifhistory:prompt+="【对话历史】\n"+"\n".join(history)+"\n\n"ifrag:prompt+="【检索到的相关文档】\n"+rag+"\n\n"iftool_result:prompt+="【计算器结果】\n"+tool_result+"\n\n"prompt+=f"用户问题：{state['question']}\n\n请给出最终回答："ans=call_siliconflow_llm(prompt)history.append(f"用户:{state['question']}")history.append(f"助手:{ans}")return{"answer":ans,"history":history}

6. LangGraph 构建

fromlanggraph.graphimportStateGraph graph=StateGraph(State)graph.add_node("retrieve",retrieve)graph.add_node("tool",tool)graph.add_node("answer",answer)graph.set_entry_point("retrieve")graph.add_edge("retrieve","tool")graph.add_edge("tool","answer")app=graph.compile()

这样就形成了RAG + 工具调用 + 多轮对话的完整工作流。

7. 多轮对话示例

state={"question":"太阳能发电的原理是什么？","answer":"","session_id":"user001","history":[],"retrieved_docs":"","tool_result":""}res=app.invoke(state)print("机器人：",res["answer"])state["question"]="那它的效率一般是多少？"res=app.invoke(state)print("机器人：",res["answer"])state["question"]="200*(1+0.05)^3"res=app.invoke(state)print("机器人：",res["answer"])