Runtime:agent在运行中能得到的对象可以获取agent在运行中的数据contextstatestore概念作用读写典型生命周期Context提供本次运行所需要的身份、配置、依赖只读单次调用State保存Agent执行中产生的动态数据可读写同一会话thread_idStore保存跨会话需要复用的长期数据可读写跨会话一、context1、概念Context是一种依赖注入机制调用者在运行开始时把用户身份、租户信息、权限等信息传给Agent节点、中间件和工具在执行时再从Runtime读取。·调用时注入·工具通过runtime.context按需读取·生命周期仅限本次运行·作用把隐私数据与Agent解耦2、案例需求多租户的在线教育Agent具备一个查询用户资料的工具可以查询当前登陆用户的资料from dataclasses import dataclass from langchain.tools import tool,ToolRuntime from langchain.agents import create_agent from langchain.messages import HumanMessage from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os user_database { schools:{ u001:{name:虎哥,course:LangChain开发指南}, u002:{name:张飞,course:数据结构} }, schools2:{ u001:{name:小王,course:Python入门}, u002:{name:小小,course:人工智能} } } load_dotenv(Api.env) llm init_chat_model( modeldeepseek-flash, model_provideropenai, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) #定义context Schema传递数据从数据库拿然后工具再从context拿 #True 不能修改 dataclass(frozenTrue) class UserContext: user_id:str tenant_id:str #定义tool拿信息从context tool def data(runtime:ToolRuntime[UserContext]): description:查询当前数据 runtime:泛型参数tool注入runtime参数由LangChain自动注入 #获取context ctx runtime.context #获取用户信息 profile user_database.get(ctx.tenant_id,{}).get(ctx.user_id) return ( f姓名{profile[name]}, f课程{profile[course]} ) #定义agent指定contect类型 agent create_agent( model llm, tools [data], #指定context类型 context_schema UserContext, system_prompt你是课程平台助手需要用户资料时调用工具 ) #调用Agent,调用时通过Context传递用户信息 response agent.invoke( {messages:[HumanMessage(content请你帮我查查我的课程信息)]}, contextUserContext( user_idu001, tenant_id schools ) ) for message in response[messages]: message.pretty_print()二、StateAgentState中只有messages信息保存历史会话。但是我们可以继承AgentState拓展自定义字段保存其他信息。· State结构本质是一个字典但是结构不能改变TypeDict· 操作方式必须以字典方式操作State案例自定义State记录调用模型的次数。定义工具每次调用模型后次数就累加from dataclasses import dataclass from langchain.tools import tool,ToolRuntime from langchain.agents import create_agent,AgentState from langchain_core.messages import HumanMessage,ToolMessage from langchain.chat_models import init_chat_model from dotenv import load_dotenv from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command import os load_dotenv(Api.env) llm init_chat_model( modeldeepseek-flash, model_provideropenai, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) #自定义state dataclass class CustomState(AgentState): count:int #通过runtime访问state tool def update_count(runtime:ToolRuntime[CustomState]): 获取CustomState的count值 #获取state的计数器 count runtime.state.get(count,0) update { count:count1, messages:[ToolMessage( content 成功更新state, tool_call_id runtime.tool_call_id )] } #这个就不返回给模型了这个返回值是要去更新state的值 return Command(update update) thread_config {configurable: {thread_id: 1}} #定义agent指定contect类型 agent create_agent( model llm, tools [update_count], #指定context类型 state_schemaCustomState, checkpointerInMemorySaver(), system_prompt你是AI助手你需要在每次回答用户前调用工具记录调用次数 ) #调用Agent,调用时通过Context传递用户信息 response agent.invoke( {messages:[HumanMessage(content你是谁)]}, thread_config ) response agent.invoke( {messages:[HumanMessage(content你还记得我是谁吗)]}, thread_config ) for message in response[messages]: message.pretty_print() print(agent.get_state(thread_config))三、Storestore保存跨会话需要复用的长期数据案例用Store实现用户偏好的长期记忆通过Store保存用户学习偏好实现跨会话后Agent依然记得用户学习偏好和习惯· 结构Store是多级目录结构用namespace目录key对应一个唯一的JSON文档· 格式namespace元组key字符串· 操作读取getnamespacekey存储putnamespacekeyvaluenamespace和key还有value的关系大概就是用户需要根据namespace这个元组确认然后再根据namespace的key去获取value总结value需要namespace和key才能获取namespacekeyvalue(a_platform,school_a,u001)learning_preferences{“学习偏好”“喜欢中午时间段学习”}code_preferences{“编码偏好”“简介但要有注释”}from dataclasses import dataclass from langchain.tools import tool,ToolRuntime from langchain.agents import create_agent from langchain_core.messages import HumanMessage from langchain.chat_models import init_chat_model from dotenv import load_dotenv from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore from pydantic import BaseModel import os load_dotenv(Api.env) llm init_chat_model( modeldeepseek-flash, model_provideropenai, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) class UserContext(BaseModel): user_id:str tenant_id:str key 学习代码顺序偏好 #定义namespace和key def preference_namespace(ctx:UserContext)-tuple[str,...]: return(a_platform,ctx.tenant_id,ctx.user_id) #通过runtime访问state tool def save_learning_preferences(preference_name:str,preference_value:str,runtime:ToolRuntime[UserContext])-str: 保存当前用户的长期学习偏好 # runtime.context通过上下文的tenant_id和user_id来构建namespace namespace preference_namespace(runtime.context) #通过namespace和key来读取用户偏好返回结果item.value是dict格式store是双层寻址 item runtime.store.get(namespace,key) prefences dict(item.value) if item else {} #记录本次要保存的用户偏好 prefences[preference_name] preference_value #保存到store中 runtime.store.put(namespace,key,prefences) return f已保存偏好{preference_name} {preference_value} tool def get_learning_preference(runtime:ToolRuntime[UserContext])-str: 读取长期学习偏好 namespace preference_namespace(runtime.context) item runtime.store.get(namespace,key) if item is None: return 当前用户还没有保存学习偏好 return f当前学习偏好{item.value} thread_config {configurable: {thread_id: 1}} thread_config_01 {configurable: {thread_id: 2}} #定义agent agent create_agent( model llm, tools [save_learning_preferences,get_learning_preference], context_schemaUserContext, checkpointerInMemorySaver(), storeInMemoryStore(), system_prompt(你是课程平台助手 涉及用户资料时调用资料工具 设计长期学习偏好或给用户讲解代码时调用偏好工具 ) ) #调用Agent,调用时通过Context传递用户信息 response agent.invoke( {messages:[HumanMessage(content我喜欢先看代码再看注释)]}, thread_config, context UserContext(user_idu001,tenant_idschool_a) ) for message in response[messages]: message.pretty_print() response agent.invoke( {messages:[HumanMessage(content你还记得我的学习偏好吗)]}, thread_config_01, contextUserContext(user_idu001,tenant_idschool_a) ) for message in response[messages]: message.pretty_print()
