LLMOps记忆模块解析与Runnable深入

本章介绍

拆解 LLMOps 项目记忆模块所实现的功能以及对应的需求，了解记忆模块的运行流程。

编写 LLMOps 记忆模块 API 文档并约定 API 响应数据格式。

整理并设计 LLMOps 记忆模块所需的应用配置 / 会话 / 消息数据表，编写 ORM 模型代码。

学习 Runnable 动态绑定运行时参数使用技巧与解析，掌握构建动态 LLM 应用的技巧。

掌握 Runnable 在运行时配置链内部的两种操作技巧，涵盖修改运行状态与替换组件。

掌握 Runnable 生命周期监听器、重试机制、回退机制，确保 LLM 应用健康运行。

开源智能体记忆模块解析

拆解智能体框架 MetaGPT 记忆模块运行流程与源码速读。

优秀智能体项目学习推荐，例如：AutoGPT、BabyAGI、AutoGen 等。

项目记忆功能需求拆解分析

运行流程拆解与基础数据表设计

01. 模块运行流程拆解

1.1 应用配置运行流程

[流程图]

1.2 应用调试运行流程

[流程图]

1.3 其他运行流程

更新长记忆运行流程

传递
前端发起请求
更新长记忆
逻辑判断
后端接收参数
查询应用信息
已开启
未开启
获取应用配置
是否开启长记忆
逻辑判断
获取应用的会话记录
返回更新结果
抛出错误信息
不存在
存在
检测会话记录
是否存在
更新会话记录信息
摘要汇总
创建会话记录
并更新应用关联会话

清空应用调试历史信息运行流程

前端发起
清空历史信息请求
后端接收参数
查询应用信息
存在
不存在
检测应用是否
存在调试会话id
更新应用信息
调试会话
设置为空
id
摘要总结设置为空
返回清空结果

删除指定消息运行流程

前端发起
删除指定消息请求
后端接收参数
查询应用信息
存在
不存在
检测应用是否
存在调试会话id
更新消息状态
执行软删除
抛出错误信息
存在
检测应用是否
存在该消息id
返回删除结果

02. 基础数据表设计

2.1 默认值与表迁移

在前面的课时中，我们在 ORM 模型中，使用 default 来为表设置默认值，例如 App 模型。这里的 default 其实是通过代码层面上去设置默认值，也就是在使用 App 实例时，如果没有设置默认值，代码层面会填充对应的默认值，但是在数据库层面时没有设置默认值的。

例如从 postgres 中导出对应的 app 表结构 SQL，如下：

python

class App(db.Model):
    """AI应用基础模型类"""
    __tablename__ = "app"
    __table_args__ = (
        PrimaryKeyConstraint("id", name="pk_app_id"),
        Index("idx_app_account_id", "account_id"),
    )
    id = Column(UUID, default=uuid.uuid4, nullable=False)
    account_id = Column(UUID, nullable=False)
    name = Column(String(255), default="", nullable=False)
    icon = Column(String(255), default="", nullable=False)
    description = Column(Text, default="", nullable=False)
    status = Column(String(255), default="", nullable=False)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False)
    created_at = Column(DateTime, default=datetime.now, nullable=False)

sql

-- ----------------------------
-- Table structure for app
-- ----------------------------

Page 10 大量数字（可能为图片或乱码，忽略）

Page 11

python

updated_at = Column(DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP(0)'), server_onupdate=text('CURRENT_TIMESTAMP(0)'))
created_at = Column(DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP(0)'))

在上述的代码中，使用 uuid_generate_v4() 这个 postgres 函数，但是 postgres 默认安装是不带 uuid 函数的，可以通过 create 命令安装对应的扩展：

sql

create extension "uuid-ossp"

安装扩展后，即可在 postgres 命令行执行如下语句，检查是否开启成功：

sql

SELECT uuid_generate_v4()

2.2 ORM 生成提示词修正

修改了开发过程中表设计规范，对应的生成 ORM 模型的 Prompt 也需要进行调整，修改后如下：

角色

你是一个拥有10年经验的资深Python工程师，精通Flask、Flask-SQLAlchemy、Postgres，以及其他Python开发工具，能够为用户提出的需求或者提供的代码段生成指定的完整代码。

技能说明

如果需要实现SQLAlchemy的ORM类，集成db.Model时，从 from internal.extension.database_extension import db 这里导入db；
创建ORM模型时，表名 __tablename__ 及类名全部都是单数；
所有的字段都要添加 nullable=False 代表字段不允许为空，除非特定说明，或者没有设置默认值的情况；
UUID类型的主键字段添加默认值 server_default = text('uuid_generate_v4()')，String类型的字段长度均设置为 String(255)，如果没有指定默认值则设置为 server_default = text("character varying")；
String类型的默认值请写 server_default = text("character varying") 而不是 server_default = text("")，这点非常重要；
Text类型的默认值请写 server_default = text("text") 而不是 server_default = text("") 的格式；
所有模型都有 updated_at 和 created_at 字段，类型均是 DateTime，其中 updated_at 包含 server_default 和 server_onupdate，而 created_at 仅包含 server_default，值全部都是 text('CURRENT_TIMESTAMP(0)')；
请给ORM模型添加上 __table_args__ 属性，涵盖 PrimaryKeyConstraint 为主键，所有模型都以 id 为主键，主键的类型为 UUID，如果用户声明其他约束，例如 UniqueConstraint，Index 等时，请按照需求进行添加；
属性的类型全部从 sqlalchemy 包中导入，例如：from sqlalchemy import (Column, UUID, String, DateTime, PrimaryKeyConstraint, UniqueConstraint)；
对于 description 等字段，通过字面意思，可以看出是描述，一般内容比较长，可以使用 Text 类型；
用户如果表名了某个字段类型为 json，则统一设置成 JSON，并从 from sqlalchemy.dialects.postgresql import JSONB 导入，这是 Postgres 特有的；
其他的规范请根据你的知识库进行操作，项目使用的数据库是 Postgres；

操作示例

python

import uuid
from datetime import datetime

Page 12

2.3 Draw.io ER 图与数据库

在某个功能未确定下来之前，如果需要设计数据表，一般情况下会先考虑绘制数据表的 ER 图，用于清晰展示不同数据表实体之间的关系，这样可以避免了后期频繁修改数据库。

python

from sqlalchemy import (
    Column,
    UUID,
    String,
    DateTime,
    PrimaryKeyConstraint,
    UniqueConstraint,
    Index,
    text,
)
from internal.extension.database_extension import db

class AccountOAuth(db.Model):
    """第三方授权认证账号模型"""
    __tablename__ = "account_oauth"
    __table_args__ = (
        PrimaryKeyConstraint("id", name="pk_account_oauth_id"),
        UniqueConstraint("account_id", "provider", name="uk_account_oauth_account_id_provider"),
        UniqueConstraint("provider", "openid", name="uk_account_oauth_provider_openid"),
        Index("idx_account_oauth_account_id", "account_id")
    )
    id = Column(UUID, nullable=False, server_default=text('uuid_generate_v4()'))
    account_id = Column(UUID)
    provider = Column(String(255), nullable=False, server_default=text("''::character varying"))
    openid = Column(String(255), nullable=False, server_default=text("''::character varying"))
    encrypted_token = Column(String(255), nullable=False, server_default=text("''::character varying"))
    updated_at = Column(DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP(0)'), server_onupdate=text('CURRENT_TIMESTAMP(0)'))
    created_at = Column(DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP(0)'))

Python 只处理与生成测试用例相关的提问，对于其他非相关行业问题，请婉拒回答。
只使用用户使用的语言进行回答，不使用其他语言。
确保回答的针对性和专业性。

用户的需求是：

Page 13

绘制数据库ER图的软件有很多，例如：MySQL Workbench、Navicat、PyCharm插件、Draw.io等。

对比其他软件，Draw.io只需打开网页即可快速绘制ER图，链接：https://draw.io。

记忆功能相关 API 文档编写整理

应用API接口统一以JSON格式返回，并且包含3个字段：code、data和message，分别代表业务状态码、业务数据和接口附加信息。

业务状态码共有6种，其中只有success（成功）代表业务操作成功，其他5种状态均代表失败，并且失败时会附加相关的信息：fail（通用失败）、not_found（未找到）、unauthorized（未授权）、forbidden（无权限）和validate_error（数据验证失败）。

接口示例：

json

{
  "code": "success",
  "data": {
    "direct_url": "https://github.com/login/oauth/authorize?client_id=f69102c6b97d90d69768&redirect_uri=http%3A%2F%2Flocalhost%3A5001%2Foauth%2Fauthorize%2Fgithub&scope=user%3Aemail"
  },
  "message": ""
}

带有分页数据的接口会在data内固定传递list和paginator字段，其中list代表分页后的列表数据，paginator代表分页的数据。

paginator内存在4个字段：current_page（当前页数）、page_size（每页数据条数）、total_page（总页数）、total_record（总记录条数），示例数据如下：

json

{
  "code": "success",
  "data": {
    "list": [
      {
        "app_count": 0,
        "description": "这是专门用来存储慕课LLMOps课程信息的知识库",
        "document_count": 13,
        "icon": "https://imooc-llmops-1257184990.cos.ap-guangzhou.myqcloud.com/2024/04/07/96b5e270-c54a-4424-acee-ff8a2b7e4331.png",
        "id": "c0759ca8-2d35-4480-83a8-141f29d1401",
        "name": "慕课LLMOps课程知识库",
        "updated_at": 1713106758,
        "word_count": 8850
      }
    ],
    "paginator": {
      "current_page": 1,
      "page_size": 20,
      "total_page": 1,
      "total_record": 2
    }
  },
  "message": ""
}

如果接口需要授权，需要在 headers 中添加 Authorization，并附加 access_token 即可完成授权登录，示例：

Authorization: Bearer eyJhbGciOiJIUz1NilslnR5cCl6lkpxVCJ9.eyJleHAiOjE3MTYONTY3OTgsimlzcyl6lmxsbW9wcyl slnN1Yi6lmM5MDljMWRilWlyMmUtNGZINi04OGlyLWlyZTkxZWFiMWE3Yi9JDAtWDBBGixA_XFihf opRe4Cz- RQ9_TAcno9w81tNbE

01. 应用模块

1.1 [todo]获取应用基础信息

接口说明：传递对应的应用 id，获取当前应用的基础信息+配置信息等内容。
接口信息：授权 + GET:/apps/:app_id
接口参数：
- 请求参数：
  - app_id -> uuid：路由参数，必填，需要获取的应用 id。
- 响应参数：
  - id -> uuid：应用 id，类型为 uuid。
  - name -> string：应用名称。
  - icon -> string：应用图标。
  - description -> string：应用描述。
  - published_app_config_id -> uuid：已发布应用配置 id，如果不存在则为 null。
  - drafted_app_config_id -> uuid：草稿应用配置 id，如果不存在则为 null。
  - debug_conversation_id -> uuid：调试会话记录 id，如果不存在则为 null。
  - published_app_config / drafted_app_config -> json：应用配置信息，涵盖草稿配置、已发布配置，如果没有则为 null，两个配置的变量信息一致。
    - id -> uuid：应用配置 id。
    - model_config -> json：模型配置，类型为 json。
    - dialog_round -> int：携带上下文轮数，类型为非负整型。
    - memory_mode -> string：记忆类型，涵盖长记忆 long_term_memory 和 none 代表无。
    - status -> string：应用配置的状态，drafted 代表草稿、published 代表已发布配置。
    - updated_at -> int：应用配置的更新时间。
    - created_at -> int：应用配置的创建时间。
  - updated_at -> int：应用的更新时间。
  - created_at -> int：应用的创建时间。
响应示例：

json

{
  "code": "success",
  "data": {
    "id": "5e7834dc-bbc4-4ee5-9591-8f297f5acded",
    "name": "慕课LLMOps聊天机器人",
    "icon": "https://imooc-llmops-1257184990.cos.ap-guangzhou.myqcloud.com/2024/04/23/e4422149-4cf7-41b3-ad55-ca8d2caa8f13.png",
    "description": "这是一个慕课LLMOps的Agent应用",
    "published_app_config_id": null,
    "drafted_app_config_id": null,
    "debug_conversation_id": "1550b71a-1444-47ed-a59d-c2f080fbae94",
    "published_app_config": null,
    "drafted_app_config": {
      "id": "755dc464-67cd-42ef-9c56-b7528b44e7c8",
      "model_config": {},
      "dialog_round": 3,
      "memory_mode": "long_term_memory",
      "status": "draft",
      "updated_at": 1714053834,
      "created_at": 1714053834
    },
    "updated_at": 1714053834,
    "created_at": 1714053834
  },
  "message": ""
}

1.2 [todo]更新应用草稿配置信息

接口说明：更新应用的草稿配置信息，涵盖：模型配置、长记忆模式等，该接口会查找该应用原始的草稿配置并进行更新，如果没有原始草稿配置，则创建一个新配置作为草稿配置。
接口信息：授权 + POST:/apps/:app_id/config
接口参数：
- 请求参数：
  - app_id -> str：需要修改配置的应用 id。
  - model_config -> json：模型配置信息。
  - dialog_round -> int：携带上下文轮数，类型为非负整型。
  - memory_mode -> string：记忆类型，涵盖长记忆 long_term_memory 和 none 代表无。
请求示例：

json

{
  "model_config": {},
  "dialog_round": 10,
  "memory_mode": "long_term_memory"
}

响应示例：（未提供，可留空或补充默认成功响应）

1.3 [todo]获取应用调试长记忆

接口说明：用于获取指定应用的长记忆内容，如果该应用并没有开启长记忆，则会抛出错误信息。
接口信息：授权 + GET:/apps/:app_id/long-term-memory
接口参数：
- 请求参数：
  - app_id -> str：需要获取长记忆的应用 id。
- 响应参数：
  - summary -> str：该应用最新调试会话的长记忆内容。
响应示例：

json

{
  "code": "success",
  "data": {
    "summary": "人类自我介绍为暮小课，并要求人工智能解释LLM（大型语言模型）的概念。人工智能将LLM描述为一种基于深度学习的模型，通常建立在Transformer架构上，用于自然语言处理任务。LLM经历了一个预训练阶段，在那里他们从大量的文本数据中学习语言结构，比如维基百科的文章和书籍。它们利用自我注意机制来有效地处理长程依赖关系。经过预训练后，LLM可以针对特定的应用程序进行微调，使其功能适应文本生成、理解和分类等任务。LLM由于其多功能性和强大的语言理解和生成能力，被广泛应用于虚拟助理、翻译、情绪分析、医疗保健、金融等领域，代表了自然语言处理的前沿技术。"
  },
  "message": ""
}

1.4 [todo]更新应用调试长记忆

接口说明：用于更新对应应用的调试长记忆内容，如果应用没有开启长记忆功能，则调用接口会发生报错。
接口信息：授权 + POST:/apps/:app_id/long-term-memory
接口参数：
- 请求参数：
  - app_id -> str：路由参数，需要更新长记忆的应用 id。
  - summary -> str：需要更新的长记忆内容。
请求示例：

json

{
  "summary": "人类介绍自己叫暮小课，喜欢打篮球。"
}

响应示例：（未提供）

1.5 [todo]应用调试对话

接口说明：用于在编排AI应用时进行debug调试，如果当前应用没有草稿配置，则使用发布配置进行调试，如果有草稿配置则以草稿配置信息进行调试。
接口信息：授权 + POST:/apps/:app_id/debug
接口参数：
- 请求参数：
  - app_id -> str：路由参数，需要调试的AI应用 id，格式为 uuid。
  - query -> str：用户发起的提问信息。
- 响应参数：
  - id -> uuid：响应消息的 id，类型为 uuid。
  - conversation_id -> uuid：消息关联会话的 id，类型为 uuid。
  - query -> str：人类的输入字符串。
  - answer -> str：AI的生成内容。
  - answer_tokens -> int：生成内容消耗的 Token 数。
  - response_latency -> float：响应消耗的时间，单位为毫秒。
  - updated_at -> int：消息的更新时间。
  - created_at -> int：消息的创建时间。
请求示例：

json

{
  "query": "能详细讲解下LLM是什么吗？"
}

响应示例：

json

{
  "code": "success",
  "data": {
    "id": "1550b71a-1444-47ed-a59d-c2f080fbae94",
    "conversation_id": "2d7d3ef-95c9-4d9d-ba9c-9daa09cc8a8",
    "query": "能详细讲解下LLM是什么吗？",
    "answer": "LLM即Large Language Model，大语言模型，是一种基于深度学习的自然语言处理模型，具有很高的语言理解和生成能力，能够处理各式各样的自然语言任务，例如文本生成、问答、翻译、摘要等。它通过在大量的文本数据上进行训练，学习到语言的模式、结构和语义知识。",
    "answer_tokens": 1454,
    "response_latency": 8541,
    "updated_at": 1714053834,
    "created_at": 1714053834
  },
  "message": ""
}

1.6 [todo]获取应用调试历史对话列表

接口说明：用于获取应用调试历史对话列表信息，该接口支持分页，单次最多返回20组对话消息，并且分页以时间字段进行降序，接口不会返回软删除对应的数据。
接口信息：授权 + GET:/apps/:app_id/messages
接口参数：
- 请求参数：
  - app_id -> str：路由参数，需要调试的 AI 应用 id，格式为 uuid。
- 响应参数：
  - id -> uuid：响应消息的 id，类型为 uuid。
  - conversation_id -> uuid：消息关联会话的 id，类型为 uuid。
  - query -> str：人类的输入字符串。
  - answer -> str：AI的生成内容。
  - answer_tokens -> int：生成内容消耗的 Token 数。
  - response_latency -> float：响应消耗的时间，单位为毫秒。
  - updated_at -> int：消息的更新时间。
  - created_at -> int：消息的创建时间。
响应示例：

json

{
  "code": "success",
  "data": {
    "list": [
      {
        "id": "1550b71a-1444-47ed-a59d-c2f080fbae94",
        "conversation_id": "2d7d3e3f-95c9-4d9d-ba9c-9daaf09cc8a8",
        "query": "能详细讲解下LLM是什么吗？",
        "answer": "LLM即Large Language Model，大语言模型，是一种基于深度学习的自然语言处理模型，具有很高的语言理解和生成能力，能够处理各式各样的自然语言任务，例如文本生成、问答、翻译、摘要等。它通过在大量的文本数据上进行训练，学习到语言的模式、结构和语义知识。",
        "answer_tokens": 1454,
        "response_latency": 8541,
        "updated_at": 1714053834,
        "created_at": 1714053834
      }
    ],
    "paginator": {
      "current_page": 1,
      "page_size": 20,
      "total_page": 1,
      "total_record": 2
    }
  },
  "message": ""
}

1.7 [todo]删除特定的调试消息

接口说明：用于删除 AI 应用调试对话过程中指定的消息，该删除会在后端执行软删除操作，并且只有当会话 id 和消息 id 都匹配上时，才会删除对应的调试消息。
接口信息：授权 + POST:/apps/:app_id/messages/:message_id/delete
接口参数：
- 请求参数：
  - app_id -> uuid：路由参数，需要删除消息归属的应用 id，格式为 uuid。
  - message_id -> uuid：路由参数，需要删除的消息 id，格式为 uuid。
请求示例：

json

{
  "app_id": "1550b71a-1444-47ed-a59d-c2f080fbae94",
  "message_id": "2d7d3e3f-95c9-4d9d-ba9c-9daaf09cc8a8"
}

响应示例：

json

{
  "code": "success",
  "data": {},
  "message": "删除调试信息成功"
}

1. bind 函数用途与使用技巧

在使用LangChain开发的时候，某些场合我们希望在一个Runnable可运行队列中调用另一个Runnable，并传递一些常量参数，但是这些参数不是前一个Runnable的输出的一部分，也不是用户输入的一部分，而是某个Runnable组件的一部分参数。

我们就可以考虑使用Runnable.bind()来传递这些默认参数。

例如以下场景：

创建了一个ChatOpenAI的LLM大语言模型，利用这个LLM来构建两条链；
第1条链的temperature为0.7，即生成的内容确定性更强；第2条链的temperature为1.2，生成的内容会更随机，更有创意；
在构建时，即可通过LLM.bind(temperature=0.7)和LLM.bind(temperature=1.2)来为LLM设置不同的默认调用参数；

bind()函数用于修改Runnable底层的默认调用参数，并在调用时会自动传递该参数，无需手动传递，像原始链一样正常调用即可。所以如果在构建Runnable链应用时就知道对应的参数，可以使用bind函数来绑定参数（事先指定）。

1.1 动态添加默认调用参数

例如在构建链应用时，初始化一个通用的LLM大语言模型，在构建链应用时才绑定上对应的停止词，可以让LLM在更多链上被使用，灵活性更强，而无需实例化多个LLM，实例如下：

python

import dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

prompt = ChatPromptTemplate.from_messages([
    ("system", "你正在执行一项测试，请重复用户传递的内容，除了重复其他均不要操作"),
    ("human", "{query}")
])
llm = ChatOpenAI(model="gpt-4o")
chain = prompt | llm.bind(stop=["world"]) | StrOutputParser()
content = chain.invoke({"query": "Hello world"})
print(content)

输出：

Hello

1.2 解决多参 RunnableLambda 函数传参

在 LangChain 中，如果要将一个函数变成 Runnable 组件，可以通过 RunnableLambda 函数进行包装。但是封装后，所有的 Runnable 组件的 invoke 函数，调用时，只能传递一个参数（类型不限制），如果原本的函数支持多个参数，并且是必填参数，就会出现报错。

例如：

python

import random
from langchain_core.runnables import RunnableLambda

def get_weather(location: str, unit: str) -> str:
    """根据传入的位置+温度单位获取对应的天气信息"""
    print("location:", location)
    print("unit:", unit)
    return f"{location}天气为{random.randint(24, 40)}{unit}"

get_weather_runnable = RunnableLambda(get_weather)
resp = get_weather_runnable.invoke({"location": "广州", "unit": "摄氏度"})
print(resp)

上述代码在执行 invoke 时虽然传递了字典，并且包含了 location 和 unit 两个参数，但是这个参数只会作为唯一的一个值，传递给 get_weather 函数的 location 参数，所以实际上 get_weather 函数接收的参数如下：

location: {"location": "广州", "unit": "摄氏度"}
unit: None

而使用 bind() 函数绑定添加其他默认调用参数，从而巧妙实现 RunnableLambda 组件实现接收多个参数，修改示例：

python

import random
from langchain_core.runnables import RunnableLambda

def get_weather(location: str, unit: str) -> str:
    """根据传入的位置+温度单位获取对应的天气信息"""
    print("location:", location)
    print("unit:", unit)
    return f"{location}天气为{random.randint(24, 40)}{unit}"

get_weather_runnable = RunnableLambda(get_weather).bind(unit="摄氏度")
resp = get_weather_runnable.invoke("广州")
print(resp)

输出：

location: 广州
unit: 摄氏度
广州天气为31摄氏度

02. bind 函数运行流程解析

通过上述的案例，可以知道 bind() 函数是在构建应用的时候添加上对应的默认调用参数，而在 Runnable.bind() 函数的底层，本质上是往 Runnable 的 kwargs 属性添加对应的字段，并生成一个新的 Runnable，当 Runnable 组件执行调用时（invoke、ainvoke、stream、abatch、batch等），会自动将 kwargs 字段里的所有参数合并并覆盖默认调用参数。

从而完成动态添加默认调用参数的效果，Runnable.bind() 的运行流程如下：

image[[116, 271, 725, 433]]

虽然 bind() 是所有Runnable共有的方法，但是并不是所有的Runnable组件都支持绑定默认调用参数，部分组件底层并没有默认调用参数的概念，例如PromptTemplate底层的invoke方法，并没有使用到bind()的逻辑。

python

# langchain_core/prompts/base.py -> BasePromptTemplate
def invoke(self, input: Dict, config: Optional[RunnableConfig] = None) -> PromptValue:
    config = ensure_config(config)
    if self.metadata:
        config["metadata"] = {**config["metadata"], **self.metadata}
    if self.tags:
        config["tags"] = config["tags"] + self.tags
    return self._call_with_config(
        self._format_prompt_with_error_handling,
        input,
        config,
        run_type="prompt",
    )

组件动态添加默认调用参数

组件配置运行时链内部

01. configurable_fields 方法使用技巧

在某些场合下除了在构建链的时候配置对应的调用参数，也可能让链在执行调用的时候才去配置对应的运行时链内部（运行时修改链相应参数），包括运行时动态调整温度、停止词、传递自定义参数、甚至是运行时动态替换模型为另外一个。

针对这类需求，在LangChain也提供了相应的解决方案：

configurable_fields()：和bind()方法接近，但是并不是在构建时传递对应的参数，而是在链运行时为链中的给定步骤指定参数，比bind()更灵活。
configurable_alternatives()：使用这个方法可以在链运行时，将链中的某一个部分替换成其他替换方案，例如：运行中更换提示模板、更换大语言模型等。

configurable_fields()方法使用分成两个流程：

为Runnable定义哪些字段可以在链运行时动态配置；
在调用invoke()函数时，传递对应的配置信息configurable完成动态配置；

例如，在链的调用过程中，将temperature温度设置为0，使用代码如下：

python

import dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

prompt = ChatPromptTemplate.from_template("请生成一个小于{x}的随机整数")

# 2.创建LLM大语言模型，并配置temperature参数为可在运行时配置，配置键为llm_temperature
llm = ChatOpenAI(model="gpt-3.5-turbo-16k").configurable_fields(
    temperature=ConfigurableField(
        id="llm_temperature",
        name="大语言模型温度",
        description="用于调整大语言模型生成内容的随机性"
    )
)

# 3.构建链应用
chain = prompt | llm | StrOutputParser()

# 4.正常调用内容
content = chain.invoke({"x": 1000})
print(content)

# 5.将temperature修改为0调用内容
content = chain.invoke(
    {"x": 1000},
    config={"configurable": {"llm_temperature": 0}}
)
print(content)

02. configurable_fields 运行流程与解析

configurable_fields() 和 bind() 非常接近，但是可配置范围更广，只要 Runnable 组件下有的所有属性，都可以通过 configurable_fields() 进行配置，而 bind() 只能配置调用参数（一般调用参数都和组件参数有关系）。

可以通过以下函数来查看 configurable_fields() 支持配置哪些字段（父类属性也可以配置，但是在这里不显示）：

Runnable_fields_keys()

例如 PromptTemplate 组件下存在 template 字段，这个字段是提示模板的模板本体，同样也可以快速实现动态配置。

代码如下：

python

from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import ConfigurableField

# 1.创建提示模板并配置支持动态配置的字段
prompt = PromptTemplate.from_template("请写一篇关于{subject}主题的冷笑话").configurable_fields(
    template=ConfigurableField(
        id="prompt_template",
        name="提示模板",
        description="提示模板字符串本身"
    )
)

# 2.传递配置更改prompt_template并调用生成内容
content = prompt.invoke(
    {"subject": "程序员"},
    config={"configurable": {"prompt_template": "请写一篇关于{subject}主题的藏头诗"}}
).to_string()
print(content)

输出结果：

请写一篇关于程序员主题的藏头诗

configurable_fields() 的运行原理及流程其实也非常简单，在调用 invoke() 函数调用的时候，会调用 _prepare() 预处理函数，在预处理函数这里，会重新依据原有参数+配置的参数创建对应的组件进行覆盖，从而实现配置运行时链内部。

运行流程：

image[[核心代码图]]

核心代码：

python

# 示例代码（省略）

Page 30 仅有数字 0（可能为图片，忽略）

组件动态替换运行时组件

01. configurable_alternatives 方法与使用技巧

在LLMOps项目中，应用编排页面可以在调试的过程中替换大语言模型继续之前的对话进行调试，这就是需要运行时组件替换功能，例如在构建的链应用中，动态替换掉特定的模型、提示词等整个组件本身，而不是替换组件里的参数信息。

在LangChain中，提供了一个叫 configurable_alternatives() 方法来实现这个功能，所有的Runnable组件均支持这个函数。

使用动态替换的流程也存在几个步骤：

为有可能替换的组件定义一个键，这样在链应用中就可以区分出来是哪一个组件；
接下来为当前的组件选项设置一个默认值，当没有传递任何配置信息时，使用的就是默认值，组件不会发生替换；
接下来创建所有替换组件的实例，并为备选方案添加上对应的key值，以便配置信息知道值和对应的组件的关系；
调用链，并传递配置信息，执行对应的动态替换运行组件。

例如，构建一条链，可以同时选择gpt-4o、gpt-3.5-turbo-16k、文心一言等模型，代码如下：

python

import dotenv
from langchain_community.chat_models.baidu_qianfan_endpoint import QianfanChatEndpoint
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1.创建提示模板&定义默认大语言模型
prompt = ChatPromptTemplate.from_template("{query}")
llm = ChatOpenAI(model="gpt-3.5-turbo-16k").configurable_alternatives(
    ConfigurableField(id="llm"),
    gpt4=ChatOpenAI(model="gpt-4o"),
    wenxin=QianfanChatEndpoint(),
)

# 2.构建链应用
chain = prompt | llm | StrOutputParser()

# 3.调用链并传递配置信息
content = chain.invoke(
    {"query": "你好，你是什么模型呢？"},
    # config={"configurable": {"llm": "gpt4"}},
    config={"configurable": {"llm": "wenxin"}}
)
print(content)

输出如下：

您好，我是百度研发的知识增强大语言模型，中文名是文心一言，英文名是ERNIE Bot。我能够与人对话互动，回答问题，协助创作，高效便捷地帮助人们获取信息、知识和灵感。如果您有任何问题，请随时告诉我。

02. configurable_alternatives 运行流程与解析

configurable_alternatives() 的运行流程非常简单，在底层会通过一个字典 alternatives 存储所有替换组件，并且从传递的 configurable 字典中获取当前需要选择的组件 key，根据对应的 key 在 alternatives 中找到对应的组件返回并执行后续的操作。

运行流程图如下：

核心代码：

python

# langchain_core/runnables/configurable.py => RunnableConfigurableAlternatives
def _prepare(self, config: Optional[RunnableConfig] = None) -> Tuple[Runnable[Input, Output], RunnableConfig]:
    config = ensure_config(config)
    which = config.get("configurable", {}).get(self.which.id, self.default_key)
    # remap configurable keys for the chosen alternative
    if self.prefix_keys:
        config = cast(
            RunnableConfig,
            {
                **config,
                "configurable": {
                    _strremoveprefix(k, f"{self.which.id}=={which}/"): v
                    for k, v in config.get("configurable", {}).items()
                },
            },
        )
    # return the chosen alternative
    if which == self.default_key:
        return (self.default, config)
    elif which in self.alternatives:
        alt = self.alternatives[which]
        if isinstance(alt, Runnable):
            return (alt, config)
        else:
            return (alt(), config)
    else:
        raise ValueError(f"unknown alternative: {which}")

组件重试 / 回退机制降低程序错误率

1. Runnable 重试机制

在 LangChain 中，针对 Runnable 抛出的异常提供了重试机制——with_retry()，当 Runnable 组件出现异常时，支持针对特定的异常或所有异常，重试特定的次数，并且配置每次重试时间的时间进行指数增加。

with_retry 函数的参数如下：

retry_if_exception_type：需要重试的异常，默认为所有异常，类型为元组。
wait_exponential_jitter：是否在重试之间添加抖动，默认为True，即每次重试时间指数增加（并随机再增加1秒内的时间）。
stop_after_attempt：重试的次数，默认为3，即3次重试后没有正常结果就暂停。

例如，想要让一个Runnable组件最多重试2次，只需在with_retry()函数中传递stop_after_attempt参数即可，代码：

python

from langchain_core.runnables import RunnableLambda

counter = -1

def func(x):
    global counter
    counter += 1
    print(f"当前值为{counter}")
    return x / counter

chain = RunnableLambda(func).with_retry(stop_after_attempt=2)
resp = chain.invoke(2)
print(resp)

with_retry() 函数的运行原理非常简单，通过构建一个新的 Runnable，在执行调用类的函数时，循环特定次数，直到组件能正常执行结束即暂停，并且在每次循环的过程中，休眠特定的时间，运行流程图如下：

image[[流程图]]

02. Runnable 回退机制

在某些场合中，对于Runnable组件的出错，并不想执行重试方案，而是执行特定的备份/回退方案，例如OpenAI的LLM大模型出现异常时，自动切换到文心一言的模型上，在LangChain中也提供了对应的回退机制——with_fallback，

with_fallback 函数的参数：

fallbacks：原始组件运行失败，进行回退/替换的Runnable组件列表，必填参数。
exceptions_to_handle：需要回退的异常，默认为所有异常，类型为元组。
exception_key：错误异常键，当指定错误信息后，Runnable组件产生的错误异常作为输入的一部分传递给回退组件，且以指定的键名存储，默认为None，表示异常不会传递给回退处理程序。

例如，为LLM添加回退备选方案，当OpenAI模型调用失败时，自动切换到文心一言执行任务：

python

import dotenv
from langchain_community.chat_models import QianfanChatEndpoint
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1.构建prompt与LLM，并将model切换为gpt-3.5-turbo-18k引发出错
prompt = ChatPromptTemplate.from_template("{query}")
llm = ChatOpenAI(model="gpt-3.5-turbo-18k").with_fallbacks([QianfanChatEndpoint()])

# 2.构建链应用
chain = prompt | llm | StrOutputParser()

# 3.调用链并输出结果
content = chain.invoke({"query": "你好，你是？"})
print(content)

运行流程如下：

image[[流程图]]

组件生命周期监听器与使用场景

01. Runnable 生命周期监听器

在LangChain中，除了在链执行时传递config=callbacks来对链进行监听，Runnable还提供了一个简约的方法with_listeners来监听开始、结束、出错这3个常见的生命周期，并且with_listeners提供的方法比callbackHandler更简洁，更统一。

with_listeners配置的on_start、on_end、on_error函数格式参数一模一样，如下：

run_obj：Run：运行时对象，内部涵盖了运行id、运行名称、开始时间、结束时间、运行类型、额外数据、错误信息、输入字典、输出字典、标签等内容。
config：RunnableConfig：执行时传递的config配置信息，类型为字典。

例如，为RunnableLambda添加对应的生命周期监听器：

python

import time
from langchain_core.runnables import RunnableLambda, RunnableConfig
from langchain_core.tracers.schemas import Run

def on_start(run_obj: Run, config: RunnableConfig) -> None:
    print("on_start")
    print("run_obj:", run_obj.inputs)
    print("config:", config)
    print("---")

def on_end(run_obj: Run, config: RunnableConfig) -> None:
    print("on_end")
    print("run_obj:", run_obj)
    print("config:", config)
    print("---")

def on_error(run_obj: Run, config: RunnableConfig) -> None:
    print("on_error")
    print("run_obj:", run_obj)
    print("config:", config)
    print("---")

runnable = RunnableLambda(lambda x: time.sleep(x))
chain = runnable.with_listeners(on_start=on_start, on_end=on_end, on_error=on_error)

chain.invoke(2)

输出：

on_start
run_obj: {'input': 2}
config: {'configurable': {'name': '慕小谋'}}
---
on_end
run_obj: id=UUID('3479d7ab-3bc7-4c4c-9db9-aaaff0c1fd04') name='RunnableLambda' start_time=datetime.datetime(2024, 7, 6, 9, 24, 52, 820817, tzinfo=datetime.timezone.utc) run_type='chain' end_time=datetime.datetime(2024, 7, 6, 9, 24, 54, 821575, tzinfo=datetime.timezone.utc) extra={'metadata': {'name': '慕小谋'}} error=None serialized={'lc': 1, 'type': 'not_implemented', 'id': ['langchain_core', 'runnables', 'base', 'RunnableLambda'], 'repr': 'RunnableLambda(lambda x: time.sleep(x))'} events=[{'name': 'start', 'time': datetime.datetime(2024, 7, 6, 9, 24, 52, 820817, tzinfo=datetime.timezone.utc)}, {'name': 'end', 'time': datetime.datetime(2024, 7, 6, 9, 24, 54, 821575, tzinfo=datetime.timezone.utc)}] inputs={'input': 2} outputs={'output': None} reference_example_id=None parent_run_id=None tags=[] child_runs=[] trace_id=UUID('3479d7ab-3bc7-4c4c-9db9-aaaff0c1fd04') dotted_order='20240706T09245282081723479d7ab-3bc7-4c4c-9db9-aaaff0c1fd04'
config: {'configurable': {'name': '慕小谋'}}
---

02. with_listeners 运行流程与解析

with_listeners() 在底层会将传递的 on_start、on_end、on_error 合并到 config 配置选项中的 callback 中，本质上就是使用 callbackHandler 的逻辑来实现对应的监听。

运行流程：

image[[101, 415, 727, 583]]

核心代码：

python

# langchain_core/runnables/base.py -> Runnable
def with_listeners(
    self,
    on_start: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None,
    on_end: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None,
    on_error: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None,
) -> Runnable[Input, Output]:
    from langchain_core.tracers.root_listeners import RootListenersTracer
    return RunnableBinding(
        bound=self,
        config_factories=[
            lambda config: {
                "callbacks": [
                    RootListenersTracer(
                        config=config,
                        on_start=on_start,
                        on_end=on_end,
                        on_error=on_error,
                    )
                ],
            }
        ],
    )

# langchain_core/runnables/base.py -> RunnableBindingBase
def _merge_configs(self, *configs: Optional[RunnableConfig]) -> RunnableConfig:
    config = merge_configs(self.config, *configs)
    return merge_configs(config, *(f(config) for f in self.config_factories))

基于 Runnable 封装记忆链实现记忆自动管理

01. Runnable 封装记忆组件思路

在Runnable链应用中，可以考虑将memory通过config-configurable的形式传递给链，在链的执行函数（invoke、stream等）中可以通过第2个参数获取到对应的memory实例，从而获取到记忆历史，并且为链添加on_end函数，即可获取到整个链的输入与输出，在on_end生命周期中将对话信息存储到记忆系统中。

运行流程如下：

image[[99, 197, 728, 346]]

后端优化后的核心代码：

python

@classmethod
def _load_memory_variables(cls, input: Dict[str, Any], config: RunnableConfig) -> Dict[str, Any]:
    """加载记忆变量"""
    configurable = config.get("configurable", {})
    configurable_memory = configurable.get("memory", None)
    if configurable_memory is not None and isinstance(configurable_memory, BaseMemory):
        return configurable_memory.load_memory_variables(input)
    return {"history": []}

@classmethod
def _save_context(cls, run: Run, config: RunnableConfig) -> None:
    configurable = config.get("configurable", {})
    configurable_memory = configurable.get("memory", None)
    if configurable_memory is not None and isinstance(configurable_memory, BaseMemory):
        configurable_memory.save_context(run.inputs, run.outputs)

def debug(self, app_id: UUID):
    """聊天接口"""
    chain = (
        (RunnablePassthrough.assign(
            history=RunnableLambda(self._load_memory_variables) | itemgetter("history")
        ) | prompt | llm | StrOutputParser())
        .with_listeners(on_end=self._save_context)
    )
    # 5.调用链生成内容
    chain_input = {"query": req.query.data}
    content = chain.invoke(chain_input, config={"configurable": {"memory": memory}})

开源智能体 MetaGPT 记忆模块解读

01. MetaGPT 智能体框架

MetaGPT是一个多智能体框架，使GPT以软件公司的形式工作，协作处理更复杂的任务，例如，使用MetaGPT输入一句话需求，输出用户故事/竞品分析/需求/数据结构/APIs/文件等。

MetaGPT内部包括产品经理/架构师/项目经理/工程师，它提供了一个软件公司的全过程与精心调配的SOP。

Code = SOP（Team）是核心哲学。MetaGPT将SOP具象化，并且用于LLM构成的团队。

image[[103, 206, 725, 360]]

MetaGPT官网：https://docs.deepwisdom.ai/main/zh/
Github仓库：https://github.com/geekan/MetaGPT

MetaGPT底层是通过多个Agent之间相互协作，相互共享信息来完成复杂项目的编写，由于LLM上下文长度的限制，MetaGPT必须通过外部的记忆机制来实现对Agent交流对话之间以及历史对话信息/知识的管理。

MetaGPT记忆模块相关的功能，全部存储到memory文件夹下，涵盖了记忆基础类、长期记忆、记忆存储库、记忆大脑等，类图与对应功能如下：

2.1 AutoGPT

一个自主运行的GPT，其运行过程无需或少量人工干预，能够根据GPT自主决策结果并结合外部资源执行相应的操作，通过循环评估策略实时评估目标达成程序，来决定程序是否完成。

Github仓库：https://github.com/Significant-Gravitas/AutoGPT

2.2 BabyAGI

BabyAGI是一种能够根据任务结果自动创建、排序和执行新任务的AI智能体，和AutoGPT、MetaGPT非常接近，不过会更小巧一些。

使用教程：https://github.com/yoheinakajima/babyagi/blob/main/docs/README-cn.md
GitHub仓库：https://github.com/yoheinakajima/babyagi

2.3 AutoGen

AutoGen是一个多智能体协同的大语言模型应用开发框架，支持开发者自定义多个智能体，这些智能体可以相互交流来解决复杂的问题，相比LangChain，AutoGen天生支持多智能体交流对话，无需额外的程序，LangChain更像是构建智能应用的一块一块的积木。

Github仓库：https://github.com/microsoft/autogen

总结

LLMOps项目记忆功能拆解
- 记忆模块需求拆解，涵盖了调试对话、应用配置、清空历史信息、长记忆功能等。
- 记忆模块各个功能运行流程以及数据库表ER图设计。
- 基于运行流程+ER图实现对API文档的编写。
学习掌握Runnable相关技巧
- Runnable.bind函数动态绑定组件默认参数。
- Runnable.configurable_fields动态绑定运行时参数。
- Runnable.configurable_alternatives动态替换运行时组件。
- Runnable.with_retry和with_fallback实现组件重试与回退机制。
- Runnable.with_listeners实现对组件生命周期的检测。
开源智能体学习
- MetaGPT记忆模块各功能拆解，涵盖：基础记忆类、长期记忆、记忆存储等功能。
- 其他优质Agent项目介绍，涵盖AutoGPT、BabyAGI、AutoGen等。

LLMOps记忆模块解析与Runnable深入 ​

本章介绍 ​

项目记忆功能需求拆解分析 ​

运行流程拆解与基础数据表设计 ​

01. 模块运行流程拆解 ​

1.1 应用配置运行流程 ​

1.2 应用调试运行流程 ​

1.3 其他运行流程 ​

02. 基础数据表设计 ​

2.1 默认值与表迁移 ​

2.2 ORM 生成提示词修正 ​

角色 ​

技能说明 ​

操作示例 ​

2.3 Draw.io ER 图与数据库 ​

记忆功能相关 API 文档编写整理 ​

01. 应用模块 ​

1.1 [todo]获取应用基础信息 ​

1.2 [todo]更新应用草稿配置信息 ​

1.3 [todo]获取应用调试长记忆 ​

1.4 [todo]更新应用调试长记忆 ​

1.5 [todo]应用调试对话 ​

1.6 [todo]获取应用调试历史对话列表 ​

1.7 [todo]删除特定的调试消息 ​

1. bind 函数用途与使用技巧 ​

1.1 动态添加默认调用参数 ​

1.2 解决多参 RunnableLambda 函数传参 ​

02. bind 函数运行流程解析 ​

组件动态添加默认调用参数 ​

组件配置运行时链内部 ​

01. configurable_fields 方法使用技巧 ​

02. configurable_fields 运行流程与解析 ​

组件动态替换运行时组件 ​

01. configurable_alternatives 方法与使用技巧 ​

02. configurable_alternatives 运行流程与解析 ​

组件重试 / 回退机制降低程序错误率 ​

1. Runnable 重试机制 ​

02. Runnable 回退机制 ​

组件生命周期监听器与使用场景 ​

01. Runnable 生命周期监听器 ​

02. with_listeners 运行流程与解析 ​

基于 Runnable 封装记忆链实现记忆自动管理 ​

01. Runnable 封装记忆组件思路 ​

开源智能体 MetaGPT 记忆模块解读 ​

01. MetaGPT 智能体框架 ​

2.1 AutoGPT ​

2.2 BabyAGI ​

2.3 AutoGen ​

总结 ​

LLMOps记忆模块解析与Runnable深入

本章介绍

项目记忆功能需求拆解分析

运行流程拆解与基础数据表设计

01. 模块运行流程拆解

1.1 应用配置运行流程

1.2 应用调试运行流程

1.3 其他运行流程

02. 基础数据表设计

2.1 默认值与表迁移

2.2 ORM 生成提示词修正

角色

技能说明

操作示例

2.3 Draw.io ER 图与数据库

记忆功能相关 API 文档编写整理

01. 应用模块

1.1 [todo]获取应用基础信息

1.2 [todo]更新应用草稿配置信息

1.3 [todo]获取应用调试长记忆

1.4 [todo]更新应用调试长记忆

1.5 [todo]应用调试对话

1.6 [todo]获取应用调试历史对话列表

1.7 [todo]删除特定的调试消息

1. bind 函数用途与使用技巧

1.1 动态添加默认调用参数

1.2 解决多参 RunnableLambda 函数传参

02. bind 函数运行流程解析

组件动态添加默认调用参数

组件配置运行时链内部

01. configurable_fields 方法使用技巧

02. configurable_fields 运行流程与解析

组件动态替换运行时组件

01. configurable_alternatives 方法与使用技巧

02. configurable_alternatives 运行流程与解析

组件重试 / 回退机制降低程序错误率

1. Runnable 重试机制

02. Runnable 回退机制

组件生命周期监听器与使用场景

01. Runnable 生命周期监听器

02. with_listeners 运行流程与解析

基于 Runnable 封装记忆链实现记忆自动管理

01. Runnable 封装记忆组件思路

开源智能体 MetaGPT 记忆模块解读

01. MetaGPT 智能体框架

2.1 AutoGPT

2.2 BabyAGI

2.3 AutoGen

总结