langchain 创建智能体，并使用saver保存会话消息-程序员充电站

简单创建智能体并调用

""" 可参考官方文档地址：https://docs.langchain.com/oss/python/langchain/agents 智能体会遵循 ReAct（“推理+行动”）模式，交替进行简短的推理步骤和针对性工具调用，并将所得观察反馈到后续决策中，直到能够给出最终答案。 第一步：初始化测试用的工具 """fromlangchain_core.toolsimporttoolfrompydanticimportBaseModel,FieldclassAddInputArgs(BaseModel):a:float=Field(description="第一个数字",example=1)b:float=Field(description="第二个数字",example=2)@tool(description="对输入的两个数字执行加法运算",args_schema=AddInputArgs,return_direct=False)defadd(a,b)->float:returna+b""" 第二步：初始化大模型 这里直接初始化一个千问大模型 """importosfromlangchain_openaiimportChatOpenAIfromdotenvimportload_dotenv load_dotenv()llm=ChatOpenAI(model="qwen-max-latest",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",api_key=os.getenv("DASHSCOPE_API_KEY"),streaming=True,)""" 第三步：创建智能体 通过 langchian.agent的create_agent方法 """fromlangchain.agentsimportcreate_agent agent=create_agent(model=llm,tools=[add],system_prompt="你是一个擅长计算的专家，你会使用数学工具完成计算")""" 第四步：调用智能体 """result=agent.invoke(input={"messages":[{"role":"user","content":"请计算1+2的值"}]})"""美化 打印result的消息列表"""formsginresult['messages']:t=type(msg).__name__ift=='HumanMessage':print(f"👤{msg.content}")elift=='AIMessage':ifhasattr(msg,'tool_calls')andmsg.tool_calls:fortoolinmsg.tool_calls:args=', '.join(f'{k}={v}'fork,vintool['args'].items())print(f"🤖 调用{tool['name']}({args})")elifmsg.content:print(f"🤖{msg.content}")elift=='ToolMessage':print(f"🔧{msg.name}→{msg.content}")

控制台输出内容： 👤 请计算1+2的值 🤖 调用 add(a=1,b=2)🔧add→3.0🤖 计算结果为\(1+2=3\)。

使用Saver保存会话消息

langchian的对话消息，有多种保存方式，可以使用redis、MongoDB、或者写一个自定义saver保存到本地都是可以的

在早期可以使用社区提供的ChatMessageHistory 来实现简单的会话消息保存，保存和提取的动作都需要自己维护
后期官方提供了MemorySaver 这是完整的会话管理系统，langchian会自动处理一起

案例代码

""" agent会话保存 可以使用临时内存保存、也可以使用redis、MongoDB甚至是自己实现本地文件保存"""importasyncioimportjsonimportos from dotenvimportload_dotenv from langchain_core.messagesimportSystemMessage, HumanMessage, AIMessage from langchain_core.runnablesimportRunnableConfig from langchain.agentsimportcreate_agent load_dotenv()""" 第一步：定义Saver""" from langgraph.checkpoint.memoryimportMemorySaver memory_saver=MemorySaver()""" 第二步：初始化大模型""" from langchain_openaiimportChatOpenAI llm=ChatOpenAI(model="qwen-max-latest",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",api_key=os.getenv("DASHSCOPE_API_KEY"),streaming=True,)""" 第三步：创建react Agent 并传入checkpointer""" agent=create_agent(model=llm,checkpointer=memory_saver,debug=True,system_prompt=SystemMessage(content="你是3DM的游戏助手，专注游戏问题"),)""" 第四步 创建RunnableConfig,用来传入ThreadId 也就是会话标识""" config=RunnableConfig(configurable={"thread_id":1},recursion_limit=100)""" 第五步 流式调用agent连续对话""" async def run_conversation(): print("="*50)print("游戏助手已启动！输入 'exit' 退出，输入 'history' 查看历史记录")print("="*50)conversation_history=[]# 本地存储对话历史whileTrue: user_input=input("\n用户：")# 退出命令ifuser_input.lower()=="exit":print("\n对话结束！")breakprint("\n助手：",end="",flush=True)try:# 流式异步调用大模型asyncforchunkinagent.astream(input={"messages":user_input},config=config):# 根据你提供的格式解析chunkawait process_chunk(chunk)except Exception as e: print(f"\n[错误] 调用大模型时出错: {e}")importtraceback traceback.print_exc()print()# 换行async def process_chunk(chunk):"""处理流式返回的chunk"""ifisinstance(chunk, dict):# 处理 [values] 类型的chunkif'messages'inchunk: print(f"\n[DEBUG] 收到消息chunk:")formsginchunk['messages']:ifhasattr(msg,'content'): print(f"[DEBUG] 消息类型: {type(msg).__name__}")print(f"[DEBUG] 消息内容: {msg.content[:100]}...")# 只显示前100字符# 处理 [updates] 类型的chunkelif'model'inchunk: model_updates=chunk.get('model',{})if'messages'inmodel_updates:formsginmodel_updates['messages']:ifisinstance(msg, AIMessage)and msg.content:# 这是AI的回复内容，打印出来print(msg.content,end="",flush=True)elifisinstance(msg, HumanMessage): print(f"\n[DEBUG] 用户消息: {msg.content}")else: print(f"\n[DEBUG] 其他消息类型: {type(msg).__name__}")# 打印完整的chunk结构用于调试else: print(f"\n[DEBUG] 完整chunk结构:")print(json.dumps(chunk,indent=2,default=str,ensure_ascii=False)[:500]+"...")elifhasattr(chunk,'__dict__'):# 处理对象类型的chunkprint(f"\n[DEBUG] 对象chunk: {type(chunk).__name__}")ifhasattr(chunk,'content'): print(f"[DEBUG] 内容: {chunk.content}")ifisinstance(chunk, AIMessage): print(chunk.content,end="",flush=True)elifisinstance(chunk, str):# 直接字符串内容print(chunk,end="",flush=True)else:# 其他类型print(f"\n[DEBUG] 未知chunk类型: {type(chunk)}")print(f"[DEBUG] 值: {chunk}")if__name__=='__main__':# 运行对话asyncio.run(run_conversation())