Runtime:agent在运行中能得到的对象，可以获取agent在运行中的数据（context，state，store）

一、context

1、概念

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( model="deepseek-flash", model_provider="openai", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com" ) #定义context Schema（传递数据：从数据库拿，然后工具再从context拿） #True 不能修改 @dataclass(frozen=True) 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="请你帮我查查我的课程信息")]}, context=UserContext( user_id="u001", tenant_id = "schools" ) ) for message in response['messages']: message.pretty_print()

二、State

AgentState中只有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( model="deepseek-flash", model_provider="openai", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://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":count+1, "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_schema=CustomState, checkpointer=InMemorySaver(), 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))

三、Store

store保存跨会话需要复用的长期数据

案例：用Store实现用户偏好的长期记忆，通过Store保存用户学习偏好，实现跨会话后Agent依然记得用户学习偏好和习惯

· 结构：Store是多级目录结构，用namespace（目录）+key对应一个唯一的JSON文档；

· 格式：namespace：元组，key：字符串

· 操作：读取：get（namespace，key），存储：put（namespace，key，value）

namespace和key还有value的关系大概就是：

用户需要根据namespace这个元组确认，然后再根据namespace的key去获取value

总结：value需要namespace和key才能获取

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( model="deepseek-flash", model_provider="openai", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://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_schema=UserContext, checkpointer=InMemorySaver(), store=InMemoryStore(), system_prompt=("你是课程平台助手，" "涉及用户资料时调用资料工具，" "设计长期学习偏好或给用户讲解代码时调用偏好工具" ) ) #调用Agent,调用时通过Context传递用户信息 response = agent.invoke( {"messages":[HumanMessage(content="我喜欢先看代码再看注释")]}, thread_config, context = UserContext(user_id='u001',tenant_id='school_a') ) for message in response['messages']: message.pretty_print() response = agent.invoke( {"messages":[HumanMessage(content="你还记得我的学习偏好吗？")]}, thread_config_01, context=UserContext(user_id='u001',tenant_id='school_a') ) for message in response['messages']: message.pretty_print()