DengQN的 RSS 预览

从一个能聊天的脚本，到可扩展的个人 Agent

2026-08-19 11:50:05

我做个人 Agent 的时候，很快就遇到了一个看起来有点尴尬的问题：让模型回复一句话并不难，难的是让它长期稳定地工作。

它要记得自己在哪个会话里，知道什么时候该调用工具，能把某个操作流程交给 Skill，还要能换模型、换存储、接前端。更麻烦的是，模型“思考”过程里的工具消息很多，真正应该写入会话历史的内容却很少。如果这些东西全塞进一个 chat() 函数，代码大概能跑，但第二次改需求时就会开始还债。

我用 Python 3.11 做个个人助手 Agent。它基于deepagents和langgraph，用 Flask 暴露 REST API，默认接 DeepSeek。项目没有数据库，线程历史直接落本地 JSON 文件。

这篇文章记录一下我采用的结构，以及几个实现过程中比较容易被忽略的边界。

我想要的“个人助手”是什么

目标其实很克制：小而全，能跑起来，方便继续加能力，也方便自己读懂。

所以它暂时没有引入复杂的用户体系、消息队列和数据库。一次对话通过 thread_id 标识，服务端保存这个线程的 user/assistant 消息；Agent 负责推理和工具循环；Flask 负责把这些能力变成 HTTP 接口。

这里有一个很重要的取舍：Agent 图本身不负责跨请求记忆。每次请求到来时，应用层从历史文件读取消息，组装成 LangChain 的 messages，再交给 Agent。

这样做的好处是边界很直观。历史怎么存、要不要截断、未来是否换成 Redis，属于应用层的问题；模型是否调用工具、工具循环跑几轮，属于 Agent 层的问题

deepagents 省掉了什么

如果自己实现 Agent，核心循环大概是：模型输出 tool_calls，程序执行工具，把 ToolMessage 放回消息列表，再次调用模型，直到模型不再请求工具。

deepagents 已经把这套 ReAct 式循环放进 create_deep_agent() 里了。只需要提供模型、工具、backend、Skills 和 system prompt：

agent = create_deep_agent(
    model=model,
    backend=backends.LocalShellBackend(),
    tools=list(tools),
    middleware=[
        SkillsMiddleware(
            backend=FilesystemBackend("."),
            sources=[agent_cfg.skills_dir],
        ),
        SummarizationMiddleware(
            model=model,
            backend=backend,
            trigger=("tokens", 3000),
            keep=("messages", 20),
            trim_tokens_to_summarize=4000,
        ),
    ],
    system_prompt=agent_cfg.system_prompt,
    skills=[agent_cfg.skills_dir],
    debug=False,
)

这里的 debug=False 不是为了安静。LangGraph 的 debug 输出通常是一大坨结构化 repr，人在控制台里很难看。我单独写了 trace_pretty.py，从 stream_mode="updates" 的事件里提取思考内容、工具调用、工具结果和最终回复，整理成可读的 trace。

调用 Agent 时，API 层使用的是 stream：

reply = None
for update in agent.stream(
    {"messages": messages},
    stream_mode="updates",
):
    print_update(update)
    candidate = last_assistant_text(update)
    if candidate is not None:
        reply = candidate

这个流式调用目前只用于服务端控制台，HTTP 接口仍然等完整回复后一次性返回。

Tools 和 Skills：代码能力与操作规程

我一开始也想把所有能力都写成 Python 函数。后来发现，有些能力的核心在于一套“什么时候使用、要先做什么、禁止做什么”的流程。

因此项目把能力分成两类。

Tool 是可执行的 Python 函数，适合稳定的 API 封装、计算和确定性逻辑。比如通用的 http_request，由 LangChain 直接调用，函数的 docstring 会成为模型选择工具时看到的描述，参数类型 hint 会转成 JSON Schema。

Skill 是 Markdown 文件，适合描述操作规程。以 time_now 为例，Skill 可以要求模型先执行 date 或 time /t 获取真实时间，禁止凭记忆回答。它本身没有把“当前时间”硬编码成 Python 工具，而是用 SKILL.md 告诉 Agent 该怎么完成任务。

一个 Skill 的骨架大概是这样：

---
name: time_now
description: Use when the user asks for the current time or date.
---


# time_now


## Instructions


1. 调用命令行获取本机当前时间。
2. 把命令输出简洁地回复给用户。


## Rules


- 禁止猜测时间。

模型能不能正确选择 Skill，很大程度上取决于 frontmatter 里的 description 和正文里的约束写得是否清楚。这一点比我想象中更像写 API 文档：描述含糊，调用就会漂。

两者的判断标准可以简单一点：有稳定 Python API，就写 Tool；需要模型阅读说明后自主选择执行手段，就写 Skill。如果 Skill 最终还是要调用外部 API，可以让它复用 http_request，这样网络能力和业务流程各自保持一份。

不过 LocalShellBackend 必须认真看待。它让 Agent 能在本机执行命令，个人电脑上做实验很方便，部署到生产环境就需要沙箱、权限限制和更明确的工作目录。Agent 的“能做什么”永远不只取决于 prompt，也取决于 backend 给了它多大的手。

两种历史，不要混为一谈

会话历史的实现很简单：一个 thread_id 对应 data/<thread_id>.json，文件里只保存用户消息、助手回复和 UTC 时间戳。

{
  "thread_id": "550e8400-e29b-41d4-a716-446655440000",
  "messages": [
    {
      "role": "user",
      "content": "你好",
      "ts": "2026-08-11T01:30:00+00:00"
    },
    {
      "role": "assistant",
      "content": "你好！有什么可以帮你的？",
      "ts": "2026-08-11T01:30:02+00:00"
    }
  ]
}

工具调用中间消息没有写进去。API 只在 Agent 得到最终的、没有 tool_calls 的 AIMessage 后，追加一轮 user/assistant 记录。

每次请求的消息组装顺序是：

messages = []
if system_prompt:
    messages.append(SystemMessage(content=system_prompt))
for item in history.get_messages(thread_id):
    if item["role"] == "user":
        messages.append(HumanMessage(content=item["content"]))
    elif item["role"] == "assistant":
        messages.append(AIMessage(content=item["content"]))
messages.append(HumanMessage(content=user_message))

这里至少有两层上下文：应用层的 JSON 负责跨请求保存完整对话；Agent 内部的 SummarizationMiddleware 负责单次请求里模型、工具、工具结果不断循环时的上下文压缩。

摘要中间件触发后，会把旧消息交给摘要模型，必要时把详细历史 offload 到 backend，再用摘要替换上下文中的旧内容。它不会修改 data/*.json。所以它解决的是“这一次 Agent 运行太长”的问题，解决不了“这个线程聊了半年，下一次请求一开始就把所有历史都塞给模型”的问题。

这也是当前实现里一个刻意保留的扩展点：如果以后长会话变多，需要在应用层增加历史截断或持久化摘要。

API 只暴露稳定的结果

聊天请求大致是：

POST /au/api/threads/{thread_id}/chat
Authorization: Bearer dsk-...
Content-Type: application/json


{"message": "现在几点？"}

响应只给最终内容以及可选的 usage 信息。中间的 reasoning、tool call 和 tool result 留在服务端 trace 里，避免把 Agent 内部执行细节直接变成客户端契约。

从零做一遍，顺序比想象中重要

先用 init_chat_model 调通一次模型，再加一个工具循环；接着引入 thread_id 和历史文件；确认多轮对话可用后，再包 Flask API，补上鉴权、Skills、trace 和上下文摘要。

这样排查问题会容易很多。服务启动但聊天 500，先看 LLM 密钥；模型编造时间，先看工具或 Skill 是否真的注册；多轮失忆，检查历史有没有拼回 messages；Ollama 连接失败，确认服务和模型是否启动。Agent 项目最怕所有组件一起接入，结果只得到一个模糊的“模型不太聪明”。

现在仍然是一个偏个人使用的实现。

【本文经AI润色】

【折腾】Tun设备和使用方式

2026-01-22 09:30:49

TUN设备简述和使用方式

之前用easytier被公司检测出来，遂起意自己搓一个IP转TCP的工具，用于在异地组网

首先，什么是TUN设备

TUN（Network TUNnel）设备是一种虚拟网络设备，工作在OSI模型的第三层（网络层）。与TAP设备（工作在第二层，处理以太网帧）不同，TUN设备处理的是IP数据包。

你就知道是操作系统开的一个口子，让你在用户空间程序和内核空间网络协议栈之间进行数据交互。

数据流转过程

发送流程（用户空间 -> 内核 -> 网络）

复制代码应用程序 ↓ 系统调用（socket） ↓ 内核网络协议栈 ↓ 路由查找（匹配到TUN设备） ↓ IP数据包封装 ↓ 写入TUN设备文件

接收流程（网络 -> 内核 -> 用户空间）

复制代码 TUN设备接收数据包 ↓ 内核网络协议栈处理 ↓ 路由查找/转发 ↓ 发送到网络或本地应用

详细数据流转示例

假设有一个TUN设备 tun0，配置如下：

IP地址：10.0.0.1/24
点对点对端：10.0.0.2

发送数据包

应用程序发送数据 bash 复制代码 ping 10.0.0.2
内核处理创建ICMP Echo Request数据包源IP：10.0.0.1，目标IP：10.0.0.2 查找路由表，发现10.0.0.2通过tun0可达
写入TUN设备内核将IP数据包写入tun0的文件描述符用户空间程序通过read()读取数据包
用户空间处理读取完整的IP数据包（包含IP头和载荷）根据目标IP进行转发或其他处理

接收数据包

用户空间构造数据包构造完整的IP数据包包含IP头、协议类型、载荷等
写入TUN设备通过write()将数据包写入tun0文件描述符
内核处理内核接收数据包根据IP头信息进行路由如果目标IP是本机，传递给相应协议栈如果需要转发，查找路由表并转发

Go语言相关包介绍

github.com/songgao/water

这是一个流行的TUN/TAP设备操作库

安装

go get github.com/songgao/water

基本使用

package main
import (
	"fmt"
	"log"
	"net"
	"os/exec"
	"github.com/songgao/water"
	"github.com/vishvananda/netlink"
)
func setupTunDevice(ip string, mask int) *water.Interface {
	// 创建 TUN 设备
	ifce, err := water.New(water.Config{
		DeviceType: water.TUN,
	})
	if err != nil {
		log.Fatalln("create tun dev failed", err.Error())
	}
	// 解析 IP 地址
	addr, err := netlink.ParseAddr(fmt.Sprintf("%s/%d", ip, mask))
	if err != nil {
		log.Fatalln("parse addr failed", err.Error())
	}
	// 获取 TUN 设备
	tunDev, err := netlink.LinkByName(ifce.Name())
	if err != nil {
		log.Fatalln("get tun dev failed: ", err.Error())
	}
	// 添加 IP 地址
	if err = netlink.AddrAdd(tunDev, addr); err != nil {
		log.Fatalln("add address to tun dev failed", err.Error())
	}
	// 启用设备
	cmd := exec.Command("ip", "link", "set", ifce.Name(), "up")
	if err = cmd.Run(); err != nil {
		log.Fatal(err)
	}
	return ifce
}

实际项目使用示例

参考 client/client.go 中的实际使用方式：

package main import ( "fmt" "log" "net" "os/exec" "git.dengqn.com/dqn/gotun/config" "github.com/songgao/water" "github.com/vishvananda/netlink" ) type Client struct { tunDev *water.Interface } func initTunDev() *water.Interface { // 创建TUN设备 ifce, err := water.New(water.Config{ DeviceType: water.TUN, PlatformSpecificParams: water.PlatformSpecificParams{}, }) if err != nil { log.Fatalln("create tun dev failed", err.Error()) } // 配置IP地址 addr, _ := netlink.ParseAddr(fmt.Sprintf( "%s/%d", config.INSTANCE.Client.Ip, config.INSTANCE.Client.Mask), ) // 获取TUN设备 if tunDev, err := netlink.LinkByName(ifce.Name()); err != nil { log.Fatalln("get tun dev failed: ", err.Error()) } else { // 添加IP地址到TUN设备 if err = netlink.AddrAdd(tunDev, addr); err != nil { log.Fatalln("add address to tun dev failed", err.Error()) } else { // 启用TUN设备 cmd := exec.Command("ip", "link", "set", ifce.Name(), "up") err = cmd.Run() if err != nil { log.Fatal(err) } } } return ifce }

数据读写示例

package main import ( "log" "net" "github.com/songgao/water" ) func TunDataToServer(tunDev *water.Interface, tcpConn net.Conn) { buf := make([]byte, 1500) for { n, err := tunDev.Read(buf) if err != nil { log.Printf("Error reading from TUN: %v", err) return } // 将TUN设备读取的数据发送到TCP连接 _, err = tcpConn.Write(buf[:n]) if err != nil { log.Printf("Error writing to TCP: %v", err) return } } } func ServerDataToTun(tcpConn net.Conn, tunDev *water.Interface) error { buf := make([]byte, 1500) for { n, err := tcpConn.Read(buf) if err != nil { return err } // 将TCP连接读取的数据写入TUN设备 _, err = tunDev.Write(buf[:n]) if err != nil { log.Printf("Error writing to TUN: %v", err) return err } } }

github.com/vishvananda/netlink

用于网络设备配置和路由管理的库，常与water配合使用。

安装

bash 复制代码 go get github.com/vishvananda/netlink

配置TUN设备

package main

import (
	"fmt"
	"log"
	"net"
	"os/exec"

	"github.com/songgao/water"
	"github.com/vishvananda/netlink"
)

func setupTunDevice(ip string, mask int) *water.Interface {
	// 创建 TUN 设备
	ifce, err := water.New(water.Config{
		DeviceType: water.TUN,
	})
	if err != nil {
		log.Fatalln("create tun dev failed", err.Error())
	}

	// 解析 IP 地址
	addr, err := netlink.ParseAddr(fmt.Sprintf("%s/%d", ip, mask))
	if err != nil {
		log.Fatalln("parse addr failed", err.Error())
	}

	// 获取 TUN 设备
	tunDev, err := netlink.LinkByName(ifce.Name())
	if err != nil {
		log.Fatalln("get tun dev failed: ", err.Error())
	}

	// 添加 IP 地址
	if err = netlink.AddrAdd(tunDev, addr); err != nil {
		log.Fatalln("add address to tun dev failed", err.Error())
	}

	// 启用设备
	cmd := exec.Command("ip", "link", "set", ifce.Name(), "up")
	if err = cmd.Run(); err != nil {
		log.Fatal(err)
	}

	return ifce
}

配置路由

package main import ( "log" "net" "github.com/vishvananda/netlink" ) func setupRoute(tunName string, network string, mask int) error { // 获取TUN设备 tunDev, err := netlink.LinkByName(tunName) if err != nil { return err } // 添加路由 route := &netlink.Route{ LinkIndex: tunDev.Attrs().Index, Scope: netlink.SCOPE_UNIVERSE, Dst: &net.IPNet{ IP: net.ParseIP(network), Mask: net.CIDRMask(mask, 32), }, } if err := netlink.RouteAdd(route); err != nil { return err } log.Printf("Route added: %s/%d via %s", network, mask, tunName) return nil }

完整示例

简单的TUN设备客户端

package main import ( "fmt" "log" "net" "os/exec" "time" "github.com/songgao/water" "github.com/vishvananda/netlink" ) type Client struct { tcpConn net.Conn tunDev *water.Interface } func (c *Client) Init(serverAddr string, localIP string, mask int) error { c.tunDev = initTunDev(localIP, mask) conn, err := net.Dial("tcp", serverAddr) if err != nil { return fmt.Errorf("connect to server failed: %w", err) } c.tcpConn = conn return nil } func (c *Client) Run() error { go c.tunToTcp() return c.tcpToTun() } func (c *Client) tunToTcp() { buf := make([]byte, 1500) for { n, err := c.tunDev.Read(buf) if err != nil { log.Printf("Error reading from TUN: %v", err) return } _, err = c.tcpConn.Write(buf[:n]) if err != nil { log.Printf("Error writing to TCP: %v", err) return } } } func (c *Client) tcpToTun() error { buf := make([]byte, 1500) for { n, err := c.tcpConn.Read(buf) if err != nil { return err } _, err = c.tunDev.Write(buf[:n]) if err != nil { log.Printf("Error writing to TUN: %v", err) return err } } } func initTunDev(ip string, mask int) *water.Interface { ifce, err := water.New(water.Config{ DeviceType: water.TUN, }) if err != nil { log.Fatalln("create tun dev failed", err.Error()) } addr, _ := netlink.ParseAddr(fmt.Sprintf("%s/%d", ip, mask)) if tunDev, err := netlink.LinkByName(ifce.Name()); err != nil { log.Fatalln("get tun dev failed: ", err.Error()) } else { if err = netlink.AddrAdd(tunDev, addr); err != nil { log.Fatalln("add address to tun dev failed", err.Error()) } else { cmd := exec.Command("ip", "link", "set", ifce.Name(), "up") err = cmd.Run() if err != nil { log.Fatal(err) } } } return ifce } func main() { client := &Client{} err := client.Init("server.example.com:8080", "10.0.0.2", 24) if err != nil { log.Fatal(err) } for { err := client.Run() if err != nil { log.Printf("Connection error: %v", err) time.Sleep(5 * time.Second) err = client.Init("server.example.com:8080", "10.0.0.2", 24) if err != nil { log.Fatal(err) } } } }

【生活】青秀山花卉摄影记录

2025-03-24 07:41:12

先是去看了黄花风铃。

南宁行道树的一种，特别是东葛路一路上全是。但是在青秀山是直接开了两个山谷，还是比较震撼的。

然后是第一次看到的樱花，之前觉得也就和桃花差不多，实际上不太一样，桃花更“热烈”些。

适马30 1.4 裁切出来的，糊完了

然后是莫奈主题花园

在上传这些文件的时候，我minio的读写异常地高，不知为何

【清单】2024站点A-Z

2024-12-18 03:39:14

A

Aspose

https://products.aspose.com/

工作中用这个库维护了一套Excel、word的导出导入工具。可以说是能力比较强大的库了。比apache的poi要强。氪金就是牛。

Adoptium

https://mirrors.tuna.tsinghua.edu.cn/Adoptium/

最喜欢用的openjdk

B

Bilibili

https://www.bilibili.com/

目前来讲，国内比较好的视频平台。由于广告、画质、正版受害者等比较影响体验的问题，已经不续费了。

Baidu

https://www.baidu.com/

用百度没啥问题，主要是省事，记得装个插件把csdn纳入过滤

C

cesium

https://cesium.com/

工作原因，需要实现一些GIS场景，这个算最成熟的框架了，国内很多产品都是基于这个实现的。

cloudflare

https://www.cloudflare.com/

网络活菩萨，永远的神

DEV

https://dev.to/

基本就是RSS订阅一些新闻，看的其实不多

`E`

F

FnOS

https://fnos.net/

横空出世的国产Nas系统，跑了个在本地的pve用起来还行，目前还在快速迭代上升期。

follow

https://follow.is/

目前来讲最好的RSS阅读器，其实现了一个平台，可以用RSS把孤立的博客聚合起来。之前我有想过，能否实现一个聚合平台，定义一个通用协议，或者直接使用Nostr协议，不同的博客可以接入，后来不了了之。

foreverblog

https://www.foreverblog.cn/

十年之约，把博客接入了。

G

Github

Github.com

没思路就上去看看吧，说不定人家做好了。

Goproxy

https://goproxy.cn/

很快，很好，七牛可以说是相当慷慨了

H

## HostHatch

hosthatch.com

vps, 适合做大盘鸡跑pt。

HelloGithub

https://hellogithub.com/

Github项目推荐，已经加到RSS订阅，有时候会有一些有趣的项目

Hutool

https://hutool.cn/

如果你是java开发，可以用一下。

i

ip.sb

`curl ip.sb` 可以得到你的公网出口ip

J

## Jetbrains

jetbrains.com

Idea 依然是Java开发最好的选择，每年都会订阅。同系列的其他应用也很好用。

L

M

mirror

https://mirrors.tuna.tsinghua.edu.cn/

https://mirrors.ustc.edu.cn/

清华和中科大的源经常使用，特别是Linux发行版镜像和软件源。

N

npm mirror

https://www.npmmirror.com/

和软件源类似，js包仓库

Naive UI

https://www.naiveui.com/

水平很高。Vue3库，博客部分用到了这个。

O

openai

https://openai.com/

24年最火AI公司

P

1Panel

https://1panel.cn

新兴宝塔竞品，够用，如果你的同事不太会Linux，可以试一下。

Parsec

https://parsec.app

用过最好的远程工具，丝滑。目前需要配置代理用于登录账号，隧道打通后不用代理。

Q

R

RainYun

https://www.rainyun.com/

雨云，除了硬盘小点，其他还行。

Rust

反复入门，目前可以写一些简单的东西了，学习曲线比较陡峭。

https://course.rs/first-try/sth-you-should-not-do.html 跟着这个教程学的

https://rustcc.cn/ 社区，还算活跃的社区

S

T

开往

travellings.cn

U

V

V2ex

https://www.v2ex.com/

程序员很多，话题比较水深火热

Vue

https://vuejs.org/

开始一个前端项目，Vue3是个好选择

W

Wolai

https://www.wolai.com/

Notion 国内平替，更低的订阅价格，差不多的功能

X

Y

Z

【折腾】Gin支持mp4、mp3等文件断点加载

2024-11-28 08:26:15

本来mp4文件是直接nginx代理的，但是最近把图片音视频迁移到家里的minio里了，nginx不好直接代理到，索性实现个接口

直接上代码, 看注释得了。

复制代码

func videoServ(ctx *gin.Context) {
	filepath := ctx.Param("filepath")
	key := strings.Join([]string{config.CONFIG_INTANCE.DataDir, "videos", filepath}, "/")
	// 渠道object
	obj, err := storage.ReadFromMC(key)
	if err != nil {
		ctx.AbortWithError(http.StatusInternalServerError, err)
		return
	}
	// 取到文件大小
	fileInfo, err := obj.Stat()
	fileSize := fileInfo.Size
	// 获取浏览器的请求头
	rangeHeader := ctx.GetHeader("Range")

	// 空的就按全部大小来写入
	if rangeHeader == "" {
		// No Range header, send full file
		ctx.Header("Content-Length", strconv.FormatInt(fileInfo.Size, 10))
		ctx.Header("Accept-Ranges", "bytes")
		ctx.Header("Content-Type", "video/mp4")
		ctx.DataFromReader(http.StatusOK, fileInfo.Size, "video/mp4", obj, nil)
		return
	}

	// 解析Range头
	rangeSpec := strings.TrimPrefix(rangeHeader, "bytes=")
	startEnd := strings.SplitN(rangeSpec, "-", 2)
	// 获取开始值
	start, err := strconv.ParseInt(startEnd[0], 10, 64)
	if err != nil || start < 0 {
		ctx.JSON(http.StatusRequestedRangeNotSatisfiable, gin.H{"error": "Invalid start byte range"})
		return
	}
	// 获取结束值
	var end int64
	if len(startEnd) == 2 && startEnd[1] != "" {
		end, err = strconv.ParseInt(startEnd[1], 10, 64)
		// 超了就按最大
		if err != nil || end > fileSize-1 {
			end = fileSize - 1
		}
	} else {
		// 没有 也是最大
		end = fileSize - 1
	}
	// 值不对
	if start > end || start > fileSize-1 {
		ctx.JSON(http.StatusRequestedRangeNotSatisfiable, gin.H{"error": "Requested range not satisfiable"})
		return
	}

	// 设置响应头
	contentLength := end - start + 1
	// bytes start-end/total
	ctx.Header("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, fileSize))
	// acccept 头
	ctx.Header("Accept-Ranges", "bytes")
	ctx.Header("Content-Length", strconv.FormatInt(contentLength, 10))
	ctx.Header("Content-Type", "video/mp4")
	// 206 code
	ctx.Status(http.StatusPartialContent)

	// 跳到开始的地方
	_, err = obj.Seek(start, io.SeekStart)
	if err != nil {
		ctx.JSON(http.StatusInternalServerError, gin.H{"error": "Could not seek to start byte"})
		return
	}
	// 写入lenght长度
	io.CopyN(ctx.Writer, obj, contentLength)
	obj.Close()
}

【生活】中秋海边摄影记录

2024-09-26 15:53:38

去了个人少的海滩，阳光把沙子晒得烫脚。

大海有个比较好的地方元素比较少，稍微构图就好了。

偶尔也试试正方形构图

适马30mm 1.4 下的人像，非常给力。

DengQN修改