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

我做个人 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润色】