尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

119.Agent-LangChain核心组件-Runtime运行时

119.Agent-LangChain核心组件-Runtime运行时 摘要本文介绍 LangChain 1.0 中 Runtime 运行时的核心概念与实战用法。Runtime 是由 LangGraph 提供的依赖注入容器和执行环境负责统一调度临时信息、外部资源、会话记忆、流式输出、权限身份等让 Agent 在执行任务时通过 Runtime 获取 Context 上下文、Store 存储对象和 State 状态。文章通过源码截图和完整代码示例详细演示了如何创建自定义上下文、通过 invoke 的 context 参数传递数据、在工具中使用 ToolRuntime 注入并访问上下文以及通过 Command 更新状态和 Store 保存长期记忆帮助读者理解 Runtime 如何简化工具开发、减少重复传参。内容参考于图灵AI大模型全栈Runtime运行时在LangChain1.0版本后出的它是由LangGraph提供的一个依赖注入的容器和执行环境可以把Runtime看着是Agent使用的一个微型的工作台操作系统所有临时的信息、外部资源、会话的记忆、流式是醋壶的内容、权限身份都由Runtime统一调度就是给Agent的大脑和手脚提供一个完整的有序的工作环境简单说Agent在执行一次任务的时候可以通过Runtime获取Context上下文、Store存储对象可以用来获取长期记忆、State状态Agent任务执行完Runtime也就没有了Agent再次执行新任务时又会创建一个新Runtime在现在的LangChain中Runtime指的是ToolRuntimeRuntime如上图如果不存在Runtime我们在写工具时如果用到了配置信息比如数据库连接、用户id、apikey这种每个工具都要通过参数进行传递你写10个工具就要写十份配置信息的入参这就导致代码重复传参测试也很难全局变量还容易冲突而如果有了Runtime我们只需要传递一个Runtime类型的参数就能都解决了LangChain把配置信息、全局数据全都放到了一个类中这些信息只需要初始化一次后面直接在工具中使用即可这样工具从多参数就变成了单个Runtime类型的参数Runtime里封装的内容说明Context上下文它里面存放用户信息、配置、Apikey等它的规定是静态数据不能随意修改想修改数据可以修改state里的数据Store存储它是数据库可以读写长期记忆可以通过工具存储用户的偏好和手动保存的内容StreamWriter流式输出自定义流式输出工具中的日志、工具执行状态可以通过它反馈出来让我查看工具的执行状态ExecutionInfo执行信息元数据这次会话的信息、线程id、运行id、重试次数用来做日志追踪ServerInfo服务器信息它需要LangSmith部署才会用到有了Runtime我们在开发工具时就可以只关注业务代码、业务参数不需要关心配置信息了它会创建多个 Runtime 实例来保存不同用户的信息也就从环境配置中脱离出来只关注业务如下图红框源码中写的 Runtime 可以获取的所有内容在执行不同的工具中 Runtime 它的内容在当前Agent的任务中都是固定的除非在工具里手动修改了看代码比较好理解如下图首先创建一个类然后再调用大模型时创建Context的实例并给到invoke方法中的context参数然后如下图红框LangChain会传进来一个Runtime类型的数据就是通过 invoke 调用大模型大模型思考我们的问题然后生成调用工具的入参然后LangChain会把大模型的入参拼接成Runtime类型传递到工具里注入的过程它 ToolRuntime[Context] 是固定的参数但是被ToolRuntime[Context]修饰的runtime变量可以动态的添加数据然后就可以通过LangChain提供的变量来访问对应的数据了这样工具的参数只需传递一个Runtime其它的我们都不需要管它的整体逻辑就是首先创建一个自定义的上下文然后在工具如入参里使用ToolRuntime类型做修饰把上下文传递给ToolRuntime进行注入注入的意思是把我们自定义的上下文里的内容放到runtime变量里然后通过invoke方法的context参数传递自定义的上下文LangChain会把大模型回复的内容自动转换成ToolRuntime类型然后就可以通过ToolRuntime类型提供的变量取出对应的数据上下文可以传递的类型按着CTRL鼠标左键单击下图红框然后再按着CTRL鼠标左键单击下图红框然后再按着CTRL鼠标左键单击下图红框它支持下图红框的四种类型TypedDictLikeV1 / TypedDictLikeV2的代码实例class State(TypedDict): messages: list count: int state: State {messages: [], count: 0} 正确 state[messages] state[count] state.get(count, 0) count in state 错误dict 不支持属性访问 state.messages # AttributeErrorDataclassLike的代码实例本章的代码实例就用的DataclassLike方式class State(TypedDict): messages: list count: int state: State {messages: [], count: 0} # 正确 state[messages] state[count] state.get(count, 0) count in state # 错误dict 不支持属性访问 state.messages # AttributeErrorBaseModel代码实例它是pydanticfrom pydantic import BaseModel, Field class State(BaseModel): messages: list Field(default_factorylist) count: int 0 state State(messages[], count0) # 正确 state.messages state.count getattr(state, count) # 通常不支持字典下标 state[messages] # TypeError: State object is not subscriptable效果图它的流程这个流程先看完下方的代码再看这里的流程否则看不懂1.创建自定义上下文创建方式有TypedDictLikeV1、TypedDictLikeV2、DataclassLike、BaseModel2.然后通过agent.invoke函数中的 context 参数传递自定义上下文LangChain会通过大模型的回复给我们自动组装成Runtime类型3.在工具中通过ToolRuntime获取上下文 ToolRuntime[Context].state[messages]可以得到短期记忆ToolRuntime[Context].context可以得到我们通过invoke函数中的 context 参数传递的内容4.修改state中的内容在工具中通过return Command函数来实现上面看不懂直接看代码from dataclasses import dataclass from langchain.tools import tool, ToolRuntime from langchain_core.messages import ToolMessage from langgraph.types import Command # 可以在工具中修改状态 from langgraph.store.memory import InMemoryStore from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent, AgentState from langchain_qwq import ChatQwen from dotenv import load_dotenv import os # 加载环境变量 load_dotenv() # 初始化模型 llm ChatQwen( modelqwen3.7-flash, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlos.getenv(DASHSCOPE_BASE_URL) ) # 自定义状态 class CustomAgent(AgentState): # 用户爱好 user_hobby: list[str] # 自定义上下文的属性 dataclass # class Context: # 名字随便取 UserContext user_id: str # 用户id user_name: str # 用户名称 # 创建工具 tool def get_user_name(runtime: ToolRuntime[Context]): # 在工具中使用runtimeruntime只存在一次任务中 获取用户姓名 print(runtime.state[messages]) # 获取状态短期记忆 return runtime.context.user_name # 工具返回值会自动处理转成ToolMessage tool def set_user_hobby(hobby: list[str], runtime: ToolRuntime[Context]): # state是存在一次会话中的 设置用户爱好 print(f提取的用户爱好{hobby}) # state是短期记忆要想长期保存需要使用长期记忆 长期记忆会在langgraph详细介绍 print(f{runtime.config.get(configurable).get(thread_id)}, runtime.context.user_name) # 构造的 Namespace 和存放路径命名空间 # runtime.context.user_id是用户id # memories意思是记忆 # 如果想要存储别的东西可以在这里修改 # 注意 命名空间 的值是要有意义的就是通过看命名空间就可以知道它是做什么的 # 比如下方 runtime.context.user_id是用户id memories是记忆一眼可以看出这是用户的记忆信息 namespace (runtime.context.user_id, memories) # 对应(用户ID, 记忆类别) key user_profile # 记忆条目的 Key # 新增长期记忆 store.put( # 存储当前用户的记忆相关的内容比如爱好 namespacenamespace, keykey, value{ user_name: runtime.context.user_name, hobby: hobby } ) # 可以使用command在工具中更新状态它是专门用来改状态信息的 # runtime.state[messages] 可以获取状态 return Command( update{ # 更新Agent的state状态 user_hobby: hobby, messages: [ ToolMessage( contentf已更新用户爱好{hobby}, tool_call_idruntime.tool_call_id ) ] } ) # 创建一个内存的长期记忆 store InMemoryStore() # 存储容器-用户的长期记忆 # 创建一个内存的短期记忆state checkpointer InMemorySaver() # 创建代理 agent create_agent( modelllm, # 模型 # 设置工具 tools[get_user_name, set_user_hobby], # 设置自定义的state状态 state_schemaCustomAgent, # 设置长期记忆存储方式 storestore, # 设置短期记忆存储方式 checkpointercheckpointer, # 设置上下文类型 context_schemaContext, ) config {configurable: {thread_id: user_1}} # runtime的生命周期是Agent执行一次任务 response agent.invoke( {messages: [{role: user, content: 我的名字叫什么}]}, configconfig, # 传递当前任务的上下文它可以传递对象类型 contextContext(user_iduser1, user_name计算机王) ) print(response[messages][-1].content) print(------更新用户爱好------) response agent.invoke( {messages: [{role: user, content: 我目前的爱好喜欢编程、看电视、看球赛等}]}, configconfig, contextContext(user_iduser1, user_name计算机王) ) print(长期记忆中存储的内容, store.get((user1, memories), user_profile))
返回列表