惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

月光博客
月光博客
有赞技术团队
有赞技术团队
S
SegmentFault 最新的问题
宝玉的分享
宝玉的分享
量子位
小众软件
小众软件
The Cloudflare Blog
奇客Solidot–传递最新科技情报
奇客Solidot–传递最新科技情报
OSCHINA 社区最新新闻
OSCHINA 社区最新新闻
大猫的无限游戏
大猫的无限游戏
C
Check Point Blog
G
Google Developers Blog
博客园 - 叶小钗
H
Help Net Security
Jina AI
Jina AI
Y
Y Combinator Blog
Last Week in AI
Last Week in AI
GbyAI
GbyAI
让小产品的独立变现更简单 - ezindie.com
让小产品的独立变现更简单 - ezindie.com
Apple Machine Learning Research
Apple Machine Learning Research
MyScale Blog
MyScale Blog
T
Tailwind CSS Blog
钛媒体:引领未来商业与生活新知
钛媒体:引领未来商业与生活新知
Vercel News
Vercel News

又耳笔记

用AI挣钱之AI建站:二维码生成器工具 用AI挣钱之AI建站:https://markdowntopdf.top Cloudflare Workers实战番外一:用Static Assets托管网站,告别传统方案 Cloudflare Workers实战(五):不止JavaScript,拥抱Python与Rust Cloudflare Workers实战(四):托管和分发静态文件 Cloudflare Workers实战(三):实现认证、重定向与缓存 Cloudflare Workers实战(二):动态修改后端响应 Cloudflare Workers实战(一):随心所欲操作客户端请求 Cloudflare workers不完全指南 使用n8n创作短篇小说 水果风波:信任的代价 用Python将PDF文件转换成图片 rust网络框架Pingora源码阅读3 rust网络框架Pingora源码阅读2 rust网络框架Pingora源码阅读1 Pingora快速入门教程1之总览 Rust模板引擎askama快速入门引擎 使用Loco快速搭建自己的后台系统 用Rust发一封图文并茂的邮件 Rust小项目: 写一个简单的恶意流量阻断器 Rust小项目: 写一个简单的网页爬虫 Rust命令行库Clap快速入门教程 Rust小项目:用Rust写一个端口扫描器 Rust真全栈开发快速入门 rust声明宏快速入门教程 Rust文本处理快速入门教程 RUST web框架axum快速入门教程6之测试 白嫖免费的Rust在线运行时shuttle RUST web框架axum快速入门教程5之中间件 用Rust来做以太坊开发5之事件日志及签名
小龙虾(OpenClaw)源码分析13:用Python实现一个迷你版(仅命...
About 又耳宁 关于技术与股票的一些零碎想法 · 2026-04-18 · via 又耳笔记

前面12篇主要是“读懂OpenClaw”,这一篇番外咱们换个姿势:
自己用Python做一个迷你版,只保留命令行交互。

先说清楚边界,避免预期过高:

  • 只做CLI,不做Web UI
  • 单会话,不做多渠道接入
  • 支持/new/history/exit
  • 用JSONL保存会话历史

这篇的目标不是替代OpenClaw,而是让你亲手把核心思想跑通。

项目结构

mini_openclaw/
├── main.py
├── session.py
├── agent.py
└── requirements.txt

requirements.txt内容:

session.py

import json
from pathlib import Path
from typing import Dict, List


class SessionStore:
    """会话存储,负责历史消息的加载、追加和重置。"""

    def __init__(self, file_path: str) -> None:
        """初始化会话存储对象。"""
        self.file_path = Path(file_path)
        self.file_path.parent.mkdir(parents=True, exist_ok=True)
        if not self.file_path.exists():
            self.file_path.touch()

    def load_messages(self) -> List[Dict[str, str]]:
        """加载历史消息,返回消息列表。"""
        messages: List[Dict[str, str]] = []
        with self.file_path.open("r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                messages.append(json.loads(line))
        return messages

    def append_message(self, role: str, content: str) -> None:
        """追加单条消息到JSONL文件。"""
        record = {"role": role, "content": content}
        with self.file_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    def reset(self) -> None:
        """重置会话,清空历史文件。"""
        with self.file_path.open("w", encoding="utf-8"):
            pass

agent.py

import time
from typing import Dict, List

from openai import OpenAI


class MiniAgent:
    """迷你Agent,负责调用模型并返回回复。"""

    def __init__(self, model: str, max_retries: int = 2) -> None:
        """初始化Agent,默认带简单重试。"""
        self.client = OpenAI()
        self.model = model
        self.max_retries = max_retries

    def chat(self, messages: List[Dict[str, str]]) -> str:
        """调用模型进行对话,失败时进行有限次重试。"""
        last_error: Exception | None = None
        for attempt in range(self.max_retries + 1):
            try:
                response = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    temperature=0.3,
                )
                return response.choices[0].message.content or ""
            except Exception as e:
                last_error = e
                if attempt == self.max_retries:
                    break
                print(f"[提示] 第{attempt + 1}次请求失败,准备重试: {e}")
                time.sleep(1.5)
        raise RuntimeError(f"模型调用失败: {last_error}")

main.py

from pathlib import Path
from typing import Dict, List

from agent import MiniAgent
from session import SessionStore


SYSTEM_PROMPT = (
    "你是一个中文技术助手,回答要准确、可执行、简洁。"
    "如果不确定,请先说明假设。"
)


def build_context(history: List[Dict[str, str]], limit: int = 12) -> List[Dict[str, str]]:
    """组装上下文,保留最近N条历史,避免上下文无限膨胀。"""
    trimmed = history[-limit:]
    return [{"role": "system", "content": SYSTEM_PROMPT}, *trimmed]


def print_help() -> None:
    """打印命令帮助。"""
    print("可用命令:")
    print("  /new      重置当前会话")
    print("  /history  查看历史条数")
    print("  /exit     退出程序")


def run() -> None:
    """运行命令行主循环。"""
    data_file = Path(".mini_openclaw/session.jsonl")
    store = SessionStore(str(data_file))
    agent = MiniAgent(model="gpt-4o-mini")

    print("Mini OpenClaw 启动成功(仅CLI模式)")
    print_help()

    while True:
        user_input = input("\n你: ").strip()
        if not user_input:
            continue
        if user_input == "/exit":
            print("助手: 再见,欢迎下次继续调试。")
            break
        if user_input == "/new":
            store.reset()
            print("助手: 会话已重置。")
            continue
        if user_input == "/history":
            print(f"助手: 当前历史消息条数: {len(store.load_messages())}")
            continue

        store.append_message("user", user_input)
        history = store.load_messages()
        context = build_context(history)

        try:
            answer = agent.chat(context)
        except Exception as e:
            print(f"助手: 请求失败,请稍后重试。错误信息: {e}")
            continue

        answer = answer.strip() or "(模型返回了空内容)"
        store.append_message("assistant", answer)
        print(f"助手: {answer}")


if __name__ == "__main__":
    run()

运行方式

export OPENAI_API_KEY="你的key"
python3 main.py

示例交互:

你: 解释一下什么是事件循环
助手: ...
你: /history
助手: 当前历史消息条数: 2
你: /new
助手: 会话已重置。

有意省略了很多“工程硬骨头”:

  • 多渠道接入
  • 并发队列与调度
  • 插件与工具体系
  • 鉴权和安全策略
  • 可观测性和运维能力

但核心思想是一致的:
输入 -> 会话上下文 -> 模型回合 -> 输出 -> 历史落盘

总结

如果你能把这一版跑通,再回头看OpenClaw源码,会更容易理解为什么它需要那么多工程模块。
因为大型系统本质上不是“更聪明的聊天”,而是“更稳的运行”。

系列到这里就告一段落啦,后续如果你希望,我可以继续出:

  • 迷你版升级2:加入工具调用
  • 迷你版升级3:加入多会话和简单HTTP接口
  • 迷你版升级4:接一个真实渠道(例如Telegram)