一代明君的小屋 - python程序练习
type: channel
link: https://yidaimingjvn.xyz/index.php/category/程序/
description: python程序练习
items: 使用 pipx 专门管理 Python 应用
items: celery与多ocr任务调度的一些思路梳理和情况
items: 5090关于使用paddleocr和gpu进行ocr的demo
items: 基于mac的设备客户端任务队列调度服务 ——key设计
items: 基于mac的设备客户端任务队列调度服务
items: 分布式任务调度与资源管理系统设计
items: celery基于对象的逻辑栈分配设计
items: 关于鲁班猫rk3568适配的linux系统使用新版本python的情况
items: agent开发的memory导致上下文污染问题
items: 自定协议，串口数据处理粘包方法

使用 pipx 专门管理 Python 应用
type: item
date: 2026-07-13T23:10:44+08:00
description: 使用 pipx 专门管理 Python 应用pipx 是专门用来安装 Python 命令行工具（如 virtualenv、black、httpie）的隔离工具，它自动为每个工具创建独立的虚拟环境，完全符合您的“不用 -m venv”的要求，而且也不会触发 PEP 668。安装 pipx（推荐通过 apt 安装，避开权限问题）：sudo apt update sudo apt install pipx pipx ensurepath # 自动将 pipx 的 bin 目录加入 PATH（如果不想用 apt，也可以用 python3.12 -m pip install --user pipx）用 pipx 安装 virtualenv：pipx install virtualenv现在您可以全局使用 virtualenv 命令了：virtualenv .venv source .venv/bin/activate

celery与多ocr任务调度的一些思路梳理和情况
type: item
date: 2026-07-13T11:16:38+08:00
description: Celery 与 OCR 调度逻辑梳理1. 总体架构系统不是把接收到的 OCR 请求直接投递到 Celery OCR 队列，而是分成两级队列：Redis 逻辑队列：按设备 MAC 隔离，用于削峰、公平调度、任务 TTL 和过载淘汰。(保证对每个设备的处理更公平，避免设备任务过多堵塞)Celery 执行队列：分为调度队列、OCR 队列和文本检测队列，用于真正执行任务。核心链路：业务请求 -> 图片上传 FS -> Redis 按 MAC 逻辑队列 -> 独立调度循环挑选 MAC -> task.dispatch -> process_mac_tasks 批量弹出任务 -> sensitive.ocr -> execute_ocr_job -> FS 读取图片 -> OCR 推理 -> 敏感词检测 -> 命中后写入业务库和日志 -> PostgreSQL/DAG 写状态 -> 清理 FS 临时图片Redis 同时承担 Celery broker/backend、逻辑任务队列、调度索引、MAC 锁、Worker 心跳与运行槽位记录。2. Celery 初始化与队列Celery 使用 REDIS_URL 作为 broker 和 backend，定义三类队列：队列默认名称职责MAC 调度队列task.dispatch执行 MAC 领取和逻辑队列拆分OCR 队列task.ocr执行图片读取、OCR 和后续敏感词检测文本队列task.text执行已有文本的敏感词检测任务路由如下：dispatch_next_mac_tasks、process_mac_tasks -> 调度队列。execute_ocr_job -> OCR 队列。execute_text_job -> 文本队列。可靠性参数：task_acks_late=True：任务完成后确认；Worker 异常退出时可重新投递。task_reject_on_worker_lost=True：Worker 丢失时拒绝并回队列。worker_prefetch_multiplier=1：避免单 Worker 预取大量重任务。ignore_result=True：业务状态写数据库/DAG，不依赖 Celery result backend，减少 Redis 写入。注意：晚确认意味着任务是 at-least-once，可能重复执行，并不等同于 exactly-once。3. 启动方式与实际并发可以设计支持 worker、beat来发发挥celery的任务执行和beat机制：GPU Worker 默认 concurrency=2、pool=solo。CPU Worker 默认 concurrency=4、pool=prefork。Worker 默认同时消费三个队列。Beat 只执行调度循环、维护服务的健康检查。可以通过自己写一个脚本把声明的并发、pool、Worker 类型和队列写入环境变量，Worker ready 后注册到 Redis。关键点：solo 的有效并发固定为 1即使 GPU Worker 传入 -c 2，solo pool 仍只有一个执行槽。注册逻辑专门将其 effective_concurrency 记为 1，以免调度器按声明值 2 过量投递。因此：单进程 GPU 推理优先使用 solo + concurrency=1，表达更准确。若确实需要 GPU 并行，通常应启动多个独立 Worker 进程，并谨慎评估每个进程各自加载一份模型造成的显存占用。（worker采用prefork，开启多个运行槽位每次都会加载模型，有显存风险，建议使用solo模式来手动开启worker，便于管理）4. OCR 请求入队业务入口先检查逻辑队列是否过载：若过载，优先丢弃当前逻辑队列中最旧任务，为新任务腾出空间。没有可淘汰任务时，返回繁忙和 retry_after=30。OCR 入队过程：标准化 MAC 为 12 位小写十六进制字符串。将图片 bytes 上传至 FS，Celery/Redis payload 只保存远程 key，避免大二进制进入 Redis broker。生成 task_id、记录 detection_id、敏感词和创建时间。RPUSH 到该 MAC 的 Redis list。更新单 MAC pending、全局 pending 和调度 ZSET。初始化 DAG 的 enqueued 节点。按 MAC 分队列避免单个设备持续上传时垄断整个执行队列。5. 独立调度循环实际主调度器是 utils/ocr_task_dispatch.py(独立于celery开发的进程，不采用celery地beat来调度，避免抢占woker的任务机会)。独立进程默认每 0.5 秒执行一轮：恢复 defer 退避到期…

5090关于使用paddleocr和gpu进行ocr的demo
type: item
date: 2026-07-10T16:13:17+08:00
description: 配置gpu: 5090cuda: 12.9 -- 务必使用12.9，此前测试使用12.6 显卡根本无法正常返回数据，5090和126应该有适配问题。demo代码#!/usr/bin/env python3 import argparse import json import sys from pathlib import Path from typing import Any import paddle from paddleocr import PaddleOCR def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="使用 PaddleOCR 和 NVIDIA GPU 识别图片或 PDF" ) parser.add_argument("input", type=Path, help="待识别的图片或 PDF 路径") parser.add_argument("--gpu", type=int, default=0, help="GPU 编号，默认 0") parser.add_argument("--output", type=Path, help="JSON 输出文件，默认输出到终端") return parser.parse_args() def check_gpu(gpu_id: int) -> str: if not paddle.is_compiled_with_cuda(): raise RuntimeError( "当前 PaddlePaddle 不支持 CUDA，请安装 cu129 版 paddlepaddle-gpu。" ) gpu_count = paddle.device.cuda.device_count() if gpu_count == 0: raise RuntimeError("PaddlePaddle 未检测到 NVIDIA GPU，请检查驱动和容器 GPU 映射。") if gpu_id < 0 or gpu_id >= gpu_count: raise ValueError(f"GPU 编号 {gpu_id} 无效，当前检测到 {gpu_count} 张 GPU。") device = f"gpu:{gpu_id}" paddle.device.set_device(device) capability = paddle.device.cuda.get_device_capability(gpu_id) name = paddle.device.cuda.get_device_name(gpu_id) print( f"使用设备: {name}, compute capability {capability[0]}.{capability[1]}", file=sys.stderr, ) return device def to_json_data(result: Any) -> dict[str, Any]: data = result.json data = data() if callable(data) else data return json.loads(data) if isinstance(data, str) else data import time def main() -> int: args = parse_args() if not args.input.is_file(): raise FileNotFoundError(f"输入文件不存在: {args.input}") device = check_gpu(args.gpu) ocr = PaddleOCR( device=device, text_detection_model_name="PP-OCRv6_medium_det", text_recognition_model_name="PP-OCRv6_medium_rec", use_doc_orientation_classify=False, use_doc_unwarping=False, use_textline_orientation=False, ) n = 0 # while n < 20: if True: n+=1 start = time.time() pages = [to_json_data(result) for result in ocr.predict(str(args.input))] payload = { "input": str(args.input.resolve()), "device"…

基于mac的设备客户端任务队列调度服务 ——key设计
type: item
date: 2026-06-22T14:54:00+08:00
description: 调度服务前文基于mac的设备客户端任务队列调度服务一、MAC 任务调度相关 Keysensitive:mac:queue:<mac> — List（每个 MAC 一个）每个设备的任务队列。入队时 RPUSH，消费时 RPOP（LIFO 栈行为）。过期检查、死信重投也在此操作。sensitive:mac:active:<mac> — String（SET NX EX）运行锁，防止同一 MAC 被多个 Worker 同时调度。NX 保证原子性获得锁，EX 自动过期（默认 900 秒）防止死锁，消费完成后手动 DELETE 释放。sensitive:mac:schedule — ZSet（全局唯一）调度有序集合，score = 老化等待秒数（封顶 600 秒 × 权重）。ZREVRANGE 按等待时间从高到低获取候选 MAC，实现 "等最久的先调度" 策略。sensitive:mac:meta:<mac> — Hash（每个 MAC 一个）元数据，包含以下字段：字段说明pending_count队列积压任务数oldest_created_at最旧任务的创建时间，用于计算老化分数updated_at最后更新时间unschedulable_until退避到期时间戳（defer 指数退避时设置）二、Worker 管理相关 Keysensitive:worker:<hostname>:<pid> — Hash（每个 Worker 进程一个）Worker 资源注册与心跳，包含 CPU/GPU/内存利用率、活跃任务数、最后心跳时间等。Worker 启动时 HSET 注册，每 15 秒 心跳更新，TTL 过期（默认 30 秒）后自动清理。三、全局监控与过载保护 Keysensitive:global_pending — String（计数器）全局待处理任务总数。入队 INCR，出队 DECR，用于快速判断系统整体负载。sensitive:overloaded_macs — Set过载 MAC 集合。当单个 MAC 积压超过阈值（默认 200）时 SADD，恢复后 SREM。与 global_pending 一起用于 is_queue_overloaded() 过载判断。sensitive:watermark — String（JSON）队列水位快照，每 15 秒 由 publish_watermark 定时任务更新，包含 total_pending、mac_count、active_workers，TTL 为发布间隔 + 5 秒。四、死信与异常 Keysensitive:dead_letter — List死信队列。OCR 任务 defer 超过最大次数（默认 5 次）后移入此处，保留最近 10000 条供人工排查。abnormal:parsed_queue — List异常数据解析队列。decter 路由收到的异常数据 RPUSH 入队，限制上限 5000 条，TTL 20 秒兜底。五、Celery Broker 队列（由 Celery 管理）队列名用途sensitive.dispatch调度器队列，运行 dispatch_next_mac_sensitive_tasks / process_mac_sensitive_taskssensitive.ocrOCR 敏感检测子任务队列sensitive.text文本敏感检测子任务队列Key 关系图 ┌─────────────────────┐ │ sensitive:schedule │ ← ZSet, 按等待时间排序 └──────────┬──────────┘ │ ┌─────────────────────┼─────────────────────┐ ▼ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ queue:<mac_A> │ │ queue:<mac_B> │ │ queue:<mac_C> │ ← List, 任务数据 │ active:<mac_A> │ │ active:<mac_B> │ │ active:<mac_C> │ ← String, 运行锁 │ meta:<mac_A> │ │ meta:<mac_B> │ │ meta:<mac_C> │ ← Hash, 元数据 └──────────────────┘ └──────────────────┘ └──────────────────┘ │ ┌─────────────┴─────────────┐ ▼ ▼ ┌──────────────────┐ ┌──────────────────────┐ │ global_p…

基于mac的设备客户端任务队列调度服务
type: item
date: 2026-06-18T17:39:19+08:00
description: 一、概述基于mac的设备客户端任务队列调度服务是一个基于 Redis 的、按设备标识（MAC）维度隔离的异步任务调度中间件，核心解决以下问题：多租户公平调度：大量设备共享有限的 Worker 资源，需保证各设备的任务都能被及时处理资源感知调度：不同任务对 CPU/GPU 资源需求不同，需感知 Worker 资源状况动态分配稳定性自保：防止某设备异常发量冲垮系统，也防止系统过载后雪崩该服务的核心思想是：为每个设备维护一个独立逻辑队列，通过老化优先级计算各设备的调度得分，结合分布式互斥锁实现多 Worker 安全的公平调度。二、数据模型整个调度系统的状态全部存储在 Redis 中，核心数据结构分为五类：┌─────────────────────────────────────────────────────────────────┐ │ MAC 队列调度系统 │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 设备逻辑队列 │ 元数据 │ 调度集 │ 全局计数器 │ │ │ │ (List) │ (Hash) │ (ZSET) │ (String) │ │ │ │ │ │ │ │ │ │ │ task_mac_q:m1 │ meta:m1 │ schedule │ global_pend │ │ │ │ task_mac_q:m2 │ meta:m2 │ [m1:15.2] │ → 128 │ │ │ │ task_mac_q:m3 │ meta:m3 │ [m2:42.5] │ │ │ │ │ ... │ ... │ [m3:8.1] │ │ │ │ └────────────────┴──────────────┴───────────┴─────────────┘ │ │ │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 运行锁 │ 过载集合 │ 死信队列 │ │ │ │ (String) │ (Set) │ (List) │ │ │ │ │ │ │ │ │ │ active:m1 → TTL │ overloaded: │ dead_letter │ │ │ │ active:m2 → TTL │ {m3, m7} │ [{task1}, ...] │ │ │ └──────────────────┴─────────────────┴──────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘2.1 数据结构详表用途Redis 类型键格式值说明设备队列Listsensitive:mac:queue:{mac}JSON 序列化的任务 payload，按入队顺序排列设备元数据Hashsensitive:mac:meta:{mac}字段：pending_count(积压数)、oldest_created_at(最早任务时间)、unschedulable_until(退避截止时间)、updated_at(更新时间)调度集合ZSetsensitive:mac:schedulemember = mac，score = 老化调度分数全局计数器Stringsensitive:global_pending所有设备积压任务总数（原子计数器）运行锁Stringsensitive:mac:active:{mac}SET NX + TTL，防止多 Worker 同时调度同一设备过载集合Setsensitive:overloaded_macs积压超过阈值的设备集合死信队列Listsensitive:dead_letter超过最大退避次数的任务记录水位快照Stringsensitive:watermark定期发布的系统运行状态 JSON三、核心操作3.1 入队流程（Enqueue）客户端请求 │ ▼ 检查全局过载 ──> 过载 ──> 拒绝入队，返回 503 │ ▼ (正常) 构建任务 payload (含 task_id, task_type, mac, created_at 等) │ ▼ Redis Pipeline: ├─ RPUSH queue_key, serialized_payload (追加到队列尾部) ├─ HINCRBY meta_key pending_count 1 (递增积压计数器) └─ INCR sensitive:global_pending (递增全局计数器) │ ▼ 如果是该设备的第一个任务 └─ HSET…

分布式任务调度与资源管理系统设计
type: item
date: 2026-06-18T17:37:04+08:00
description: 一、系统概述本系统是一个面向内容检测场景的分布式异步任务调度与资源管理平台，核心提供三方面能力：任务调度：基于设备维度的逻辑队列 + 老化优先级调度，实现多租户公平调度与饥饿感知资源调度：Worker 节点资源注册与心跳，CPU/GPU/内存感知的任务分发稳定性保障：过载保护、背压机制、指数退避、死信队列、任务 TTL 过期等多层可靠性策略系统基于 Flask + Celery + Redis + PostgreSQL 技术栈构建，采用三层架构（Web 路由层 → 业务服务层 → 异步任务层），所有异步任务通过 Celery 分发至专用队列，由 Worker 集群消费。二、整体架构┌─────────────────────────────────────────────────────────┐ │ Web 路由层 (Flask) │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 检测结果上报 │ │ 图片处理任务 │ │ 文本处理任务 │ │ │ │ /api/decter │ │ /api/ocr/* │ │ /api/ocr/* │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ └─────────┼──────────────────┼──────────────────┼──────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────────────────────────┐ │ 服务层 (Services) │ │ ┌────────────────┐ ┌──────────────────────────────┐ │ │ │ 检测结果处理服务 │ │ 任务处理服务 │ │ │ │ (同步处理) │ │ (异步入队 + 过载检查) │ │ │ └────────────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌───────────────────────────────▼────────────────────┐ │ │ │ MAC 队列调度服务 │ │ │ │ 按设备维度逻辑队列 · 老化优先级排序 · 分布式互斥锁 │ │ │ │ 指数退避 · 死信 · 饥饿检测 · 动态批大小 │ │ │ └───────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 异步任务层 (Celery + Redis) │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 调度队列 │ │ 图片处理队列 │ │ 文本处理队列 │ │ │ │ dispatch │ │ ocr │ │ text │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────────────────────────────────────────────┐ │ │ │ Worker 集群 │ │ │ │ GPU Worker (消费所有队列) · CPU Worker (消费部分) │ │ │ │ Worker 注册/心跳 · 过载自检 · 自适应 CPU/GPU │ │ │ └──────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 数据持久层 (PostgreSQL) │ │ 任务记录 · 处理结果 · 统计报表 │ └─────────────────────────────────────────────────────────┘三、任务调度系统3.1 核心调度模型系统采用两级调度模型：第一级：全局调度器 (Celery Beat + dispatch Task) → 从 Sorted Set 中选取…

celery基于对象的逻辑栈分配设计
type: item
date: 2026-06-10T15:26:05+08:00
description: 运行逻辑图flowchart TD A[客户端请求 /text/handler 或 /calculate/handler] --> B[Router 参数校验/解密/取敏感词] B --> C[handlerService.text_handler / calculate_handler] C --> D[生成业务 task_id] D --> E[按 object 标准化 queue_object] E --> F[写入 Redis 逻辑队列 handler:object:queue:{queue_object}] F --> G{是否抢到该 object 的运行锁 handler:object:active:{queue_object}} G -- 否 --> H[结束提交 等待已有 dispatcher 继续消费] G -- 是 --> I[投递 Celery dispatcher 任务 dispatch_object_handler_tasks(object)] I --> J[Celery worker 固定消费 handler.dispatch] J --> K[dispatcher 启动] K --> L[循环处理最多 N 条 CELERY_object_DISPATCH_BATCH_SIZE] L --> M[从 Redis 队列 lpop 一条任务] M --> N{还有任务吗} N -- 否 --> O[释放 object 运行锁] O --> P[本轮 dispatcher 结束] N -- 是 --> Q{任务类型} Q -- text_handle --> R[_run_text_handler_job] Q -- calculate_handler --> S[_run_calculate_handler_job] R --> T[写 handlerDetectRecord processing] T --> U[执行敏感词匹配] U --> V{命中敏感词?} V -- 是 --> W[send_commend + write_log] V -- 否 --> X[正常结束] W --> Y[写 handlerDetectRecord success/failed] X --> Y S --> Z[写 calculateEngineRecord/handlerDetectRecord processing] Z --> AA[perform_calculate] AA --> AB{calculate 成功?} AB -- 否 --> AC[写 calculate failed + Detect failed] AB -- 是 --> AD[保存 calculate 成功结果] AD --> AE[执行敏感词匹配] AE --> AF{命中敏感词?} AF -- 是 --> AG[send_commend + write_log] AF -- 否 --> AH[正常结束] AG --> AI[写 handlerDetectRecord success/failed] AH --> AI Y --> AJ[刷新 object 锁 TTL] AI --> AJ AJ --> L L --> AK{本轮处理后 Redis 仍有剩余任务?} AK -- 否 --> O AK -- 是 --> AL[再次 apply_async 投递 dispatcher] AL --> P简化时序图sequenceDiagram participant Client as 客户端 participant Router as Router participant Service as handlerService participant Redis as Redis逻辑栈 participant Celery as Celery Dispatcher participant Job as 业务执行函数 participant DB as DB/日志/命令下发 Client->>Router: 提交 text/calculate 请求 Router->>Service: 调用 text_handler / calculate_handler Service->>Redis: rpush 到 object 对应逻辑队列 Service->>Redis: setnx active 锁 alt 抢锁成功 Service->>Celery: apply_async(dispatch_object_handler_tasks) else 抢锁失败 Service-->>Client: 返回 task_id，等待已有 dispatcher…

关于鲁班猫rk3568适配的linux系统使用新版本python的情况
type: item
date: 2026-04-29T17:13:48+08:00
description: 前言开发了一些程序需要放到3568上运行。但是开发环境是python3.12，而最新版本的rk3568适配的系统是ubt20，系统默认的版本是python3.8其次，为了规避linux系统使用python安装依赖与系统依赖冲突的问题。需要采用python虚拟环境来运行情况1 默认版本是3.8需要编译3.12从源码编译（最通用、最干净）这是 ARM 开发板上安装新版 Python 的标准做法，虽然编译时间稍长，但兼容性最好，也不会破坏系统自带的 python3。1.安装编译依赖（已安装过的可跳过）sudo apt update sudo apt install -y wget build-essential libreadline-dev libncursesw5-dev \ libssl-dev libsqlite3-dev tk-dev libgdbm-dev libc6-dev libbz2-dev \ libffi-dev zlib1g-dev2.下载 Python 3.12 源码去 Python 官网 找最新的 3.12 版本链接（例如 3.12.4）：wget https://www.python.org/ftp/python/3.12.4/Python-3.12.4.tgz tar -xf Python-3.12.4.tgz cd Python-3.12.43.配置、编译、安装（使用 altinstall 避免覆盖系统 python3）# 配置（可选 --enable-optimizations 会提升性能但编译时间翻倍） ./configure --enable-optimizations # 使用全部CPU核心编译，RK3568大概需要30-60分钟 make -j$(nproc) # 关键：用 altinstall 而不是 install sudo make altinstall4.验证python3.12 --version你会看到 Python 3.12.x，同时系统默认的 python3 依然是 3.8.2，互不冲突。情况2 安装python3.12-venv报错这里的处理很简单，编译版本的python环境已经带了venv工具，直接使用python3.12 -m venv venv工具既可

agent开发的memory导致上下文污染问题
type: item
date: 2026-04-24T09:51:00+08:00
description: 前言在开发AwiseOctopus的过程中，最初对于memory的设计直限于经验记忆的设计——设计大致如下：设定一个经验总结的llm，当agent完成整个任务时，交给这个agent总结成功经验并给予权重值，然后记入memory数据库，当agent执行同类任务时即可优先调用这个经验记忆的操作。想象美好，现实骨感。实际操作中，经验memory的注入反而导致了agent的上下文污染，产生了严重幻觉。例如：本来任务是操作A文件，但是由于历史经验操作的是B文件，注入后导致了经验错误，agent实际执行了B文件的操作。处理思路为了解决这一问题，以下记录了处理注入的简单实验过程。数据分层基本思路是，将经验记忆和当前上下文进行基本分层，基于信息提取优先级。让agent对于当前上下文的信息做出较高优先级的提取。再整合当前的经验和上下文。而非全量注入。首先优化一下记忆的读取，将一些关联度较低的经验过滤掉def search_experience(self, task_type, instruction, top_k=3): """搜索历史经验，根据 weight 分为成功和失败两类""" if not self.collection: return "" # 1. 向量检索最相关的 instruction results = self.collection.query( query_texts=[instruction], n_results=top_k * 2, where={"task_type": task_type} ) if not results['ids'] or not results['ids'][0]: return "" # 过滤距离大于 0.3 的结果 exp_ids = [] if 'distances' in results and results['distances']: for i, distance in enumerate(results['distances'][0]): if distance <= 0.38: exp_ids.append(results['ids'][0][i]) else: exp_ids = results['ids'][0] if not exp_ids: return ""对上下文做分层，将历史经验放置到system提示词而不是用户提示词。提示词做解释 # 搜索历史经验 hint = self.memory_manager.search_experience("thinking", user_request) if hint: yield ("RUNNING", "\n[思考Agent 经验记忆] 检索到相关历史经验，已注入上下文。") # 数据分层 for msg in self.messages: if msg["role"] == "system": msg["content"] += "\n" + ( "〖系统注入的历史经验（仅供参考，如与用户最新指令冲突，以用户指令为准）〗\n" f"{hint}\n" "〖经验参考结束〗" ) break else: self.messages.append({ "role": "system", "content": ( "〖系统注入的历史经验（仅供参考，如与用户最新指令冲突，以用户指令为准）〗\n" f"{hint}\n" "〖经验参考结束〗" ) })注: self.memory_manager是一个memory管理工具，会负责从向量数据库提取相关数据。整体结果：agent的智力明显提升，对于已经有经验的相关事件，操作速度和准确率得到了明显提升，出现需要操作A而去操作B的情况大幅下降。虽然概率下降了，但依然偶有找不到，或者用户操作目录过于模糊而导致操作失误的问题。操作工作区新增工作区机制，让agent的每个session的操作都默认限定在某个工作区(文件夹或目录)内。除非得到了用户的指定，否则操作不出工作区。整体结果：对于不确定，不明确的事项，当agent出现偏差操作时，由于有工作区兜底，操作行为风险降低。其次当把所需操作行为，操作对象放到工作区内，由于操作选项范围减少，agent操作更加精确，进一步减少了幻觉。

自定协议，串口数据处理粘包方法
type: item
date: 2026-03-06T11:07:54+08:00
description: 问题处理串口数据过程中，当前后有连续的数据的情况下，很有可能出现数据粘包的情况。如需要的数据是ef0001和de002但是数据传输和接收过快产生粘包得到数据ef0001de00处理思路(自定协议)再数据包前增加1位或者2位的数据长度帧，通过先获取长度帧判断需要取的数据位。参考代码from .FingerProtocol import * import serial import time from ._Types import * from colorama import Fore, Style __all__ = [ "FingerManager", ] class FingerManager: def __init__(self,serial_conn:serial.Serial) -> None: self.serial_conn = serial_conn def send_cmd(self, packet): """发送命令包""" self.serial_conn.write(packet) self.serial_conn.flush() def read_packet(self,timeout=2): """ 读取一个完整的数据包 1. 读取包头9字节 (Header+Addr+ID+Len) 2. 解析包长度 (大端) 3. 读取剩余数据 (Len指定长度) 1. 检查是否超时 2. 读取数据 3. 检查是否完整 """ data = b'' start_time = time.time() # 1. 读取包头9字节 (Header+Addr+ID+Len) while len(data) < 9: if time.time() - start_time > timeout: return None if self.serial_conn.in_waiting > 0: data += self.serial_conn.read(min(9 - len(data), self.serial_conn.in_waiting)) else: time.sleep(0.001) # 2. 解析包长度 (大端) pkg_len = (data[7] << 8) | data[8] # 3. 读取剩余数据 (Len指定长度) total_len = 9 + pkg_len while len(data) < total_len: if time.time() - start_time > timeout: return None if self.serial_conn.in_waiting > 0: data += self.serial_conn.read(min(total_len - len(data), self.serial_conn.in_waiting)) else: time.sleep(0.001) return data