【无标题】
1、面试官细问机器学习方面的场景方案设计2、最好自己写一下各个模型的一个提示词。3、熟悉一下face API接口最好是自己手敲一下代码。4、需要把模型换成32B的。5、RAG优化手段和NLP机器学习6、在数据及构建这块是人工构建的吗?7、chatBI模型的跨表多吗?8、SQL agent的取数结果,比如用户需要查询八月份的数据,但是最后返回出来,七月份的数据用户怎么确定他是不是八月份的数据。
1、面试官细问机器学习方面的场景方案设计
2、最好自己写一下各个模型的一个提示词。
3、熟悉一下face API接口最好是自己手敲一下代码。
4、需要把模型换成32B的。
5、RAG优化手段和NLP机器学习
6、在数据及构建这块是人工构建的吗?
7、chatBI模型的跨表多吗?
8、SQL agent的取数结果,比如用户需要查询八月份的数据,但是最后返回出来,七月份的数据用户怎么确定他是不是八月份的数据
9、机器学习相关的知识什么是XGBoost,k-means,KNN,有监督和无监督,区别,
10、agent监控
11、RagFlow、coze、dify的区别
12、Python Gil详细介绍一下
13、模型的选择为什么要这样选?选择一个大小的?
14、lora的原理是什么,怎么对两个低秩矩阵做初始化
我们的理解,它是通过矩阵分解,去构造两个低秩矩阵,然后通过对低秩矩阵的一个梯度,去做一个梯度的更新,然后做训练完后再对它们两个去做一个合并,他们两个还有一个秩r。
15、bert的位置编码是怎么做的,旋转位置编码的原理是什么,具体是怎么做的?
16、normalization了解吗batch normalization和layer normalization吗?
17、你知道残差网络吗,为什么要加入残差网络,像bert这种小模型他是把normalization放在残差网络里面的,像大模型是放在残差网络外面的,你觉得他这样做的好处是什么?
18、做多重标签的模型的时候,怎么去设计它的网络类别?bert的分类头在哪里,是哪一个token作为他的分类头输出?因为我们知道bert它本身除了用户的query这些本身的token,还有一些特殊的token,cls token和nsp token ,他是用哪一个token去做这个分类的输出的?
19、比如说
20、你用到了langgraph中的什么功能?
21、通过什么去分库的,分库的策略是什么?
22、召回的时候是一次性把所有的都召回回来后,再用matedata去过滤的吗?为什么不在检索的时候就过滤掉这些?
23、一般召回多少个chunk块?
24、微调数据
instruction agent这块的微调数据要素提取是总共7000条的数据。
instruction中的表名的获取用bert模型做多分类,微调数据总共2500条。
SQL agent中微调模型用4000条数据量
25、你的每个模型选择的时候怎么选择的,有做对比吗?
26.你的项目中有用到mcp吗,具体是怎么用的
27、什么时候用工作流,什么时候用agent?
当工作环境很复杂,不能预先枚举出所有可能的路径。或者要求实时性的任务,比如智能驾驶,这种就需要采用自主agent的形式,
fastAPI
Flask基于WSGI协议,但是只能同步,fastapi基于ASGI异步协议。
fastapi内置pydantic支持自动校验,数据类型,必填项校验,取值范围等。
fastapi是单线程多协程的方式。1、协程的切换开销近乎为0,线程的切换需要保存线程上下文,每切换一次就需要一些时间。2、单个协程占用仅需几KB内存,线程需要几MB内存。3、协程是主动释放的,是退让与恢复,线程是抢占式的,是需要抢占GIL,由操作系统内核调度,而非线程控制。
项目负责
chatbi主要有3个人负责,我主要负责sql这块,一个人负责supervision agent,另一个负责echarts agent和历史数据的召回。
rag主要有2个人负责,我主要负责处理pdf数据和召回,另外一个人负责处理其他的数据和模型的微调。
项目痛点
rag项目主要有3点,一是文档的查找和定位如同大海捞针,二是即使找到了相关文档,也需要花费大量时间去阅读大量的文档去找到具体条款。三是可能找不到相关文档的人就会去问, 比如说问hr关于合同审批流程或者可能还得找其他人咨询,可能当时这个人去开会了或者其他事情不在工位,一个简单的问题,可能因为等待而拖慢手头的工作进度。所以我们开发了这样的一套系统,当用户输入他的需求以后可以快速返回结果,并且24小时在线。
rag相关知识
🗂️ 向量数据库:为何选择Milvus
在向量数据库的选型中,Milvus相较于ChromaDB和Elasticsearch(ES),在以下几个方面展现出明显优势,更贴合企业级应用的需求:
| 评估维度 | Milvus优势 |
|---|---|
| 功能完备性 | 原生支持混合检索(如密集向量+稀疏向量),能更好地配合BGE-M3这类多功能模型。支持多种索引类型(如HNSW、IVF-PQ)和相似度度量,适配不同的精度与速度需求。 |
| 性能与扩展性 | 专为处理大规模向量数据设计,采用云原生架构,计算与存储分离,支持弹性扩展,能从容应对企业知识库未来可能的数据增长。 |
| 生态与企业级特性 | 社区活跃度高(在GitHub上有大量Stars),文档和工具完善。提供高可用性、灾难恢复及多租户支持等企业级功能,保障系统稳定运行。 |
| 与其他方案对比 | - ChromaDB:轻量级,适合本地开发和测试,但在处理大规模数据、企业级特性及混合检索支持上不如Milvus。 - Elasticsearch (ES):作为传统搜索引擎,其强项在于全文检索和BM25算法,虽然在后续版本中加入了对向量检索的支持,但在专门的大规模、高并发向量相似性搜索场景下,性能通常不及Milvus这类专用向量数据库。 |
rag是什么,什么时候用rag,什么时候用模型微调?
1、数据检索
import uuid
import os
import json
from typing import Optional
import redis
from redis import ConnectionPool
# Deep Agent 核心依赖
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
# LangGraph 检查点依赖
from langgraph.checkpoint.memory import MemorySaver
# -------------------------- Redis全局配置(核心改造)--------------------------
# Redis连接配置(可抽离到.env,生产/开发环境通用)
REDIS_CONFIG = {
"host": os.getenv("REDIS_HOST", "127.0.0.1"),
"port": int(os.getenv("REDIS_PORT", 6379)),
"db": int(os.getenv("REDIS_DB", 0)),
"password": os.getenv("REDIS_PASSWORD", None),
"decode_responses": True,
"socket_timeout": 5,
"max_connections": 100,
}
# 初始化Redis连接池+客户端(单例,避免重复连接)
_redis_pool = ConnectionPool(**REDIS_CONFIG)
redis_client = redis.Redis(connection_pool=_redis_pool)
# -------------------------- LangGraph Redis存储适配(核心改造)--------------------------
class RedisLangGraphStore:
"""
自定义Redis存储适配LangGraph/Deep Agent的Store接口
模拟文件系统结构,Redis Key格式:memories:{path},值为文件内容/目录结构
"""
def __init__(self, client: redis.Redis):
self.client = client
self.prefix = "memories:" # Redis Key前缀,区分其他业务数据
def _get_redis_key(self, path: str) -> str:
"""拼接Redis完整Key:memories:/path/to/file"""
return f"{self.prefix}{path.lstrip('/')}"
# 实现Store核心接口:ls/read_file/write_file/edit_file(适配Deep Agent自带工具)
def ls(self, path: str = "/") -> list:
"""列出指定路径下的文件/目录"""
base_key = self._get_redis_key(path)
if base_key.endswith(":"):
pattern = f"{base_key}*"
else:
pattern = f"{base_key}:*" if path != "/" else f"{self.prefix}*"
keys = self.client.keys(pattern)
# 提取路径后缀,返回文件/目录名列表
return [k.replace(self.prefix, "", 1) for k in keys]
def read_file(self, path: str) -> str:
"""读取文件内容"""
redis_key = self._get_redis_key(path)
content = self.client.get(redis_key)
return content if content is not None else ""
def write_file(self, path: str, content: str) -> None:
"""写入文件(覆盖式)"""
redis_key = self._get_redis_key(path)
self.client.set(redis_key, content)
def edit_file(self, path: str, content: str, append: bool = True) -> None:
"""编辑文件(默认追加,append=False则覆盖)"""
if append:
old_content = self.read_file(path)
new_content = old_content + "\n" + content
else:
new_content = content
self.write_file(path, new_content)
# 实现LangGraph Store必要的通用接口
def get(self, key: str, default: Optional[any] = None) -> any:
val = self.client.get(self._get_redis_key(key))
return json.loads(val) if val else default
def set(self, key: str, value: any) -> None:
self.client.set(self._get_redis_key(key), json.dumps(value))
def delete(self, key: str) -> None:
self.client.delete(self._get_redis_key(key))
# 初始化Redis存储实例(替代原InMemoryStore/PostgresStore)
redis_store = RedisLangGraphStore(redis_client)
# -------------------------- 全局配置:检查点/运行模式 --------------------------
# 检查点:用于代理状态持久化(保留原逻辑)
checkpointer = MemorySaver()
# 仅用于标识运行模式,Redis环境下无差异,可根据需要控制Redis DB
DEV_MODE = os.getenv("DEV_MODE", "True").lower() == "true"
# -------------------------- 初始化复合后端:保留原逻辑 --------------------------
def make_composite_backend(runtime):
"""创建CompositeBackend,实现/memories/持久化(Redis),其余瞬态"""
return CompositeBackend(
default=StateBackend(runtime), # 默认:短期瞬态存储
routes={"/memories/": StoreBackend(runtime)} # 路由:/memories/前缀=Redis长期存储
)
# -------------------------- 初始化深度代理:整合所有核心功能 --------------------------
def create_long_memory_agent():
"""
初始化基于Redis的带长期记忆的深度代理,保留原所有核心功能:
1. 用户偏好自动留存 2. 自更新系统指令 3. 知识库跨会话积累 4. 研究进度持久化
"""
# 系统提示词:完全保留原逻辑,无需修改
system_prompt = """
你是一个拥有长期记忆的深度代理,你的持久化内存结构如下:
- /memories/user_preferences.txt: 存储用户的所有偏好、设置,永久留存
- /memories/instructions.txt: 你的自定义指令,可根据用户反馈编辑更新
- /memories/knowledge_base/: 跨会话积累的通用知识库,按主题分类
- /memories/research/: 研究项目进度,包含sources.txt/notes.txt/report.md
- /memories/project/: 项目相关笔记、需求文档
【核心规则】
1. 当用户告知偏好(如"我喜欢XX"、"请始终XX"),立即写入/memories/user_preferences.txt
2. 每次会话开始时,先读取/memories/instructions.txt,遵守自定义指令
3. 当用户提供反馈(如"请修改XX规则"),立即编辑/memories/instructions.txt更新
4. 学习到的通用知识/项目信息,写入/memories/knowledge_base/或/memories/project/对应文件
5. 处理研究任务时,按规范将资源/笔记/报告分别写入/memories/research/下的对应文件
6. 所有文件操作使用ls/read_file/write_file/edit_file工具,路径严格遵循上述结构
"""
# 创建深度代理(替换store为Redis存储)
agent = create_deep_agent(
store=redis_store, # 核心替换:Redis存储(替代原内存/Postgres)
backend=make_composite_backend, # 保留原混合存储后端
checkpointer=checkpointer, # 保留原检查点
system_prompt=system_prompt, # 保留原长期记忆规则
tools=None # 无需自定义工具,代理自带文件系统工具适配Redis Store
)
return agent
# -------------------------- 测试用例:保留原跨线程/功能验证逻辑 --------------------------
def test_long_memory_agent():
"""测试基于Redis的长期记忆代理:跨线程读写、偏好留存、指令更新、持久化验证"""
# 1. 初始化代理
agent = create_long_memory_agent()
print("✅ 基于Redis的长期记忆深度代理初始化完成")
print(f"🔧 Redis连接状态:{'正常' if redis_client.ping() else '异常'}")
print(f"🔧 运行环境:{'开发环境' if DEV_MODE else '生产环境'}(Redis通用)")
# 2. 线程1:写入用户偏好+自定义指令+知识库(Redis持久化)
print("\n===== 线程1:写入Redis长期记忆 =====")
config1 = {"configurable": {"thread_id": str(uuid.uuid4())}}
agent.invoke(
input={"messages": [
{"role": "user", "content": "我的偏好:喜欢极简风格,所有回答控制在3句话内,写入偏好文件"},
{"role": "user", "content": "给你添加自定义指令:回答时必须分点,写入instructions.txt"},
{"role": "user", "content": "记录知识库:Python是解释型语言,核心特性是动态类型/缩进语法,写入/memories/knowledge_base/python.txt"}
]},
config=config1
)
# 3. 线程2:读取线程1的Redis长期记忆(跨线程,无上下文传递)
print("\n===== 线程2:读取跨线程Redis长期记忆 =====")
config2 = {"configurable": {"thread_id": str(uuid.uuid4())}}
agent.invoke(
input={"messages": [
{"role": "user", "content": "读取并告诉我我的偏好是什么?"},
{"role": "user", "content": "读取你的自定义指令是什么?"},
{"role": "user", "content": "Python的核心特性有哪些?"}
]},
config=config2
)
# 4. 线程3:更新Redis长期记忆+验证研究进度存储
print("\n===== 线程3:更新Redis长期记忆+研究进度 =====")
config3 = {"configurable": {"thread_id": str(uuid.uuid4())}}
agent.invoke(
input={"messages": [
{"role": "user", "content": "修改偏好:回答除了极简,还要加emoji,更新user_preferences.txt"},
{"role": "user", "content": "我现在在做LangGraph研究,记录资源:官方文档https://langgraph.ai,写入/memories/research/sources.txt;记录笔记:LangGraph核心是状态机/多代理,写入/memories/research/notes.txt"}
]},
config=config3
)
# 5. 线程4:验证更新后的Redis记忆+研究进度(最终验证)
print("\n===== 线程4:验证更新后的Redis长期记忆 =====")
config4 = {"configurable": {"thread_id": str(uuid.uuid4())}}
agent.invoke(
input={"messages": [
{"role": "user", "content": "我的最新偏好是什么?"},
{"role": "user", "content": "LangGraph的研究资源和核心笔记是什么?"},
{"role": "user", "content": "总结一下你现在的所有长期记忆内容"}
]},
config=config4
)
# 可选:手动验证Redis中的数据(可视化)
print("\n===== 手动验证Redis存储数据 =====")
all_memory_keys = redis_client.keys(f"{redis_store.prefix}*")
for key in all_memory_keys:
content = redis_client.get(key)
print(f"📄 {key.replace(redis_store.prefix, '/')}: {content[:50]}..." if len(content) > 50 else f"📄 {key.replace(redis_store.prefix, '/')}: {content}")
# -------------------------- 主函数:运行入口 --------------------------
if __name__ == "__main__":
# 安装依赖提示(运行前执行)
install_cmd = "pip install deepagents langgraph redis>=5.0.0"
print(f"📦 运行前请安装依赖:{install_cmd}")
# 前置检查:Redis连接
if not redis_client.ping():
raise ConnectionError("❌ Redis连接失败,请检查REDIS_CONFIG配置")
# 运行测试用例
test_long_memory_agent()
print("\n✅ 所有测试完成,Redis长期记忆持久化生效(重启程序后数据仍存在)")
更多推荐



所有评论(0)