别让大模型把你拖死:Java 客户端熔断降级实战细节
2026/6/5 20:12:55
在人工智能快速发展的今天,智能聊天机器人已经不仅仅是问答工具,它们正在向多轮对话、知识库检索和工具调用的方向升级。本篇文章将系统讲解如何搭建一个完整的智能聊天机器人,涵盖多轮上下文记忆、RAG检索、以及计算器工具调用,并提供可直接运行的示例代码。
一个完整的智能聊天机器人需要解决三个核心问题:
聊天机器人最基本的能力是理解上下文。我们通过State.history来记录对话历史,每轮生成回答前,将历史内容拼接进 prompt,让 LLM 知道前文发生了什么。
智能机器人必须具备知识库能力,以便回答专业问题或动态信息查询。我们引入RAG(Retrieval-Augmented Generation)流程:
智能机器人不仅回答问题,还能进行计算。我们设计了一个自动识别数学表达式并调用计算器的流程:
我们用TypedDict定义状态对象,记录必要信息:
fromtypingimportTypedDict,ListclassState(TypedDict):question:str# 当前用户问题answer:str# 最终回答session_id:str# 会话 IDhistory:List[str]# 对话历史retrieved_docs:str# RAG 检索结果文本tool_result:str# 工具计算结果通过
State,每轮对话的信息都能被记录并传递给下一个节点。
这里我们使用硅基流动(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"]使用 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("深度学习模型训练需要…")defcalculator(expression:str)->str:importreifnotre.match(r"^[0-9+\-*/(). ]+$",expression):return"表达式不合法"try:result=eval(expression)returnstr(result)except:return"计算失败"注意:这里只允许基本运算符,防止任意代码执行风险。
我们把对话流程拆成三个节点:
defretrieve(state:State):query=state["question"]docs=vs.search(query)state["retrieved_docs"]=docsreturn{"retrieved_docs":docs}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":""}结合历史对话、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}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 + 工具调用 + 多轮对话的完整工作流。
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"])效果解析:
通过以上设计,我们实现了一个功能丰富的智能聊天机器人:
未来可以进一步优化:
通过这样的架构,聊天机器人从单纯的问答系统进化为可检索、可计算、可持续学习的智能助手。