在这篇文章中，不使用 LangChain 等现成框架，回归本质，用 Python 从零构建一个真正可落地的最小化 Agent 循环。我将实现以下核心功能：

工具调用（安全且可预测）

记忆机制（短期对话 + 长期笔记）

规划与重规划（简单实用，不搞花架子）

硬停止条件（防止 Agent 失控）

完整的追踪日志（方便调试和优化）

1、“AI Agent” 的本质逻辑

要从零构建 Agent，你只需要理解这一个核心公式：

Agent 是一个循环：

规划 → 行动（使用工具） → 观察 → 记忆 → 重新规划 → 停止

这就完了。

普通的聊天机器人是一次性问答；而 Agent 的核心能力在于自主决策下一步行动，并能通过工具持续交互，直到任务完成。

这里有一个行业内默认的“潜规则”：

优秀的 Agent 往往不追求“高智商”，而是追求“高可控性”。

因此，接下来的代码将重点聚焦于控制：结构化输出、工具安全沙箱、记忆压缩以及强制停止条件。

2、Agent 的架构设计

我将实现四大模块：

1) 工具系统

本质上就是封装好的 Python 函数，但必须包含以下元数据：

名称：调用时的标识。

描述：告诉 LLM 这个工具是干嘛的。

输入模式：定义参数的 JSON Schema。

2) 记忆系统

双层设计：

短期记忆：保存最近的对话上下文和工具执行结果（类似 RAM）。

长期记忆：将关键信息持久化到磁盘，需要时通过关键词检索（类似硬盘，这里暂不引入复杂的向量库，保持轻量）。

3) 规划系统

高层规划：由 3-6 个关键点组成的任务大纲。

下一步行动：当前这一步具体要做什么（调用工具或直接回答）。

4) 停止条件

这是区分“玩具 Demo”和“生产级 Agent”的关键：

硬限制：最大步数、最大工具调用次数、最大超时时间。

软限制：检测到重复动作、连续两步无进展时自动停止。

3、完整工作代码

这是一个纯 Python 实现的最小化 AI Agent（不依赖 LangChain）。

环境要求：

Python 3.10+

环境变量配置：

DEEPSEEK_API (必须)

DEEPSEEK_URL (可选，默认: https://api.deepseek.com )

DEEPSEEK_MODEL (可选，默认: deepseek-chat)

流程图：

"""
------------------------------------------------------------
基于原生 Python 实现的最小化 AI Agent 循环 (无 LangChain)。
- 工具系统：注册中心 + 安全执行
- 记忆系统：短期上下文 + 基于文件的长期笔记
- 规划系统：简易计划表 + 下一步行动
- 安全护栏：最大步数、工具调用限制、循环检测
- 调试支持：每一步的详细追踪日志
"""

from __future__ import annotations

import json
import os
import time
import re
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Tuple

import requests


# -----------------------------
# 工具函数
# -----------------------------
from dotenv import load_dotenv


def now_ms() -> int:
    """获取当前时间戳（毫秒）"""
    return int(time.time() * 1000)


def clamp_text(text: str, max_chars: int = 1800) -> str:
    """截断过长的文本，防止 Token 溢出"""
    text = text.strip()
    if len(text) <= max_chars:
        return text
    return text[:max_chars] + " ...[已截断]"


def safe_json_extract(text: str) -> Optional[Dict[str, Any]]:
    """
    安全地从模型回复中提取 JSON 对象。
    模型必须输出单个 JSON 对象，此函数尝试提取第一个 {...} 块。
    """
    text = text.strip()
    # 快速路径：纯 JSON
    if text.startswith("{") and text.endswith("}"):
        try:
            return json.loads(text)
        except Exception:
            pass

    # 正则提取第一个 {...} 代码块
    match = re.search(r"\{.*\}", text, flags=re.DOTALL)
    if not match:
        return None
    blob = match.group(0)
    try:
        return json.loads(blob)
    except Exception:
        return None


def simple_similarity(a: str, b: str) -> float:
    """
    微小的循环检测启发式算法：基于 token 重叠率。
    足以检测“反复执行相同动作”的死循环情况。
    """
    sa = set(a.lower().split())
    sb = set(b.lower().split())
    if not sa or not sb:
        return 0.0
    return len(sa & sb) / max(1, len(sa | sb))


# -----------------------------
# 工具系统
# -----------------------------

@dataclass
class Tool:
    name: str
    description: str
    schema: Dict[str, Any]
    fn: Callable[[Dict[str, Any]], Any]
    safe: bool = True  # 标记是否为危险工具（如写文件）


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: Dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"工具已注册: {tool.name}")
        self._tools[tool.name] = tool

    def get(self, name: str) -> Optional[Tool]:
        return self._tools.get(name)

    def as_prompt_block(self) -> str:
        """
        将工具列表转换为紧凑、易读的 Prompt 文本块。
        """
        lines = ["可用工具列表:"]
        for t in self._tools.values():
            safe_tag = "安全" if t.safe else "受限"
            lines.append(f"- {t.name} ({safe_tag}): {t.description}")
            lines.append(f"  参数定义: {json.dumps(t.schema, ensure_ascii=False)}")
        return "\n".join(lines)


# -----------------------------
# 记忆系统
# -----------------------------

@dataclass
class MemoryNote:
    ts_ms: int
    text: str
    tags: List[str] = field(default_factory=list)


class LongTermMemory:
    """
    简单的长期记忆：
    - 将紧凑的笔记存储在 JSONL 文件中
    - 基于关键词重叠进行检索（简单、快速、无需向量库）
    """

    def __init__(self, path: str = "agent_memory.jsonl") -> None:
        self.path = path
        if not os.path.exists(self.path):
            with open(self.path, "w", encoding="utf-8") as f:
                f.write("")

    def add(self, text: str, tags: Optional[List[str]] = None) -> None:
        note = MemoryNote(ts_ms=now_ms(), text=text.strip(), tags=tags or [])
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(note.__dict__, ensure_ascii=False) + "\n")

    def search(self, query: str, k: int = 5) -> List[MemoryNote]:
        query_tokens = set(query.lower().split())
        scored: List[Tuple[float, MemoryNote]] = []
        with open(self.path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                    note = MemoryNote(**obj)
                    tokens = set(note.text.lower().split())
                    score = len(tokens & query_tokens)
                    if score > 0:
                        scored.append((float(score), note))
                except Exception:
                    continue

        scored.sort(key=lambda x: x[0], reverse=True)
        return [n for _, n in scored[:k]]


class ShortTermMemory:
    """
    保存最近的 N 条消息（已压缩）。
    """

    def __init__(self, max_items: int = 18) -> None:
        self.max_items = max_items
        self.items: List[Dict[str, str]] = []

    def add(self, role: str, content: str) -> None:
        self.items.append({"role": role, "content": content})
        if len(self.items) > self.max_items:
            self.items = self.items[-self.max_items:]


# -----------------------------
# LLM 客户端 (DeepSeek)
# -----------------------------
load_dotenv()

class LLMClient:
    def __init__(self) -> None:
        # 适配国内环境，优先读取 DeepSeek 配置
        self.api_key = os.environ['DEEPSEEK_API'].strip()

        if not self.api_key:
            raise RuntimeError("缺少 API Key，请设置环境变量 DEEPSEEK_API_KEY")

        # DeepSeek 兼容 OpenAI 接口格式
        self.base_url = os.environ['DEEPSEEK_URL'].strip()
        if not self.base_url.endswith("/v1"):
            # 兼容用户可能输入不带 v1 的情况
            self.base_url = self.base_url.rstrip("/") + "/v1"

        self.model = os.environ['DEEPSEEK_MODEL'].strip()

    def chat(self, messages: List[Dict[str, str]], temperature: float = 0.2) -> str:
        url = f"{self.base_url}/chat/completions"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        payload = {
            "model": self.model,
            "temperature": temperature,
            "messages": messages,
        }
        try:
            r = requests.post(url, headers=headers, json=payload, timeout=60)
            r.raise_for_status()
            data = r.json()
            return data["choices"][0]["message"]["content"]
        except Exception as e:
            return f"LLM 调用错误: {str(e)}"


# -----------------------------
# Agent 核心逻辑
# -----------------------------

@dataclass
class AgentConfig:
    max_steps: int = 10  # 最大思考步数
    max_tool_calls: int = 6  # 最大工具调用次数
    max_seconds: int = 35  # 最大运行时间（秒）
    allow_restricted_tools: bool = False  # 是否允许使用危险工具


@dataclass
class AgentState:
    step: int = 0
    tool_calls: int = 0
    started_ms: int = field(default_factory=now_ms)
    high_level_plan: List[str] = field(default_factory=list)
    last_actions: List[str] = field(default_factory=list)  # 用于检测循环
    trace: List[Dict[str, Any]] = field(default_factory=list)  # 调试追踪


class ScratchAgent:
    """
    最小化 Agent 实现：
    - 请求模型生成计划 + 下一步行动
    - 按需执行工具
    - 存储记忆笔记
    - 安全停止
    """

    def __init__(self, llm: LLMClient, tools: ToolRegistry, ltm: LongTermMemory, cfg: AgentConfig) -> None:
        self.llm = llm
        self.tools = tools
        self.ltm = ltm
        self.cfg = cfg
        self.stm = ShortTermMemory(max_items=18)

    def _time_left(self, state: AgentState) -> int:
        elapsed = now_ms() - state.started_ms
        return max(0, self.cfg.max_seconds * 1000 - elapsed)

    def _should_stop(self, state: AgentState) -> Optional[str]:
        """检查是否满足停止条件"""
        if state.step >= self.cfg.max_steps:
            return "已达到最大步数限制"
        if state.tool_calls >= self.cfg.max_tool_calls:
            return "已达到最大工具调用次数"
        if self._time_left(state) <= 0:
            return "已达到最大运行时长"

        # 循环检测：如果连续执行非常相似的动作
        if len(state.last_actions) >= 3:
            a, b, c = state.last_actions[-3:]
            if simple_similarity(a, b) > 0.85 and simple_similarity(b, c) > 0.85:
                return "检测到重复动作循环"

        return None

    def _compact_memory_summary(self) -> str:
        """
        将短期记忆压缩为 Agent 可以携带的紧凑摘要。
        """
        # 仅保留最后 8 条，并压缩换行符
        tail = self.stm.items[-8:]
        lines = []
        for it in tail:
            role = it["role"]
            content = clamp_text(it["content"], 260).replace("\n", " ")
            lines.append(f"{role}: {content}")
        return "\n".join(lines)

    def _build_system_prompt(self, user_goal: str) -> str:
        """
        最核心的部分：严格的输出格式 + 清晰的行为规范。
        """
        return f"""
你是一个严谨的 AI Agent。你的任务是安全、高效地完成用户的目标。

用户目标:
{user_goal}

规则:
- 你必须且只能回复一个 JSON 对象，不要包含任何其他废话。
- 每一步只能选择一个动作。
- 如果信息充足，直接给出最终答案。
- 保持计划简短扼要。
- 避免死循环。如果卡住了，说明缺失了什么信息，然后尝试结束任务。

输出 JSON 格式定义:
{{
  "type": "plan" | "tool_call" | "final",
  "plan": ["..."] (当 type="plan" 时必填, 高层计划列表),
  "next": "...一句话描述下一步..." (当 type="plan" 时必填),
  "tool": "工具名称" (当 type="tool_call" 时必填),
  "args": {{...}} (当 type="tool_call" 时必填, 工具参数),
  "answer": "...最终答案..." (当 type="final" 时必填),
  "memory_note": "...需要长期存储的简短笔记..." (可选)
}}

{self.tools.as_prompt_block()}
""".strip()

    def _model_step(self, user_goal: str, state: AgentState) -> Dict[str, Any]:
        # 检索长期记忆
        ltm_hits = self.ltm.search(user_goal, k=4)
        ltm_block = "\n".join([f"- {clamp_text(n.text, 240)}" for n in ltm_hits]) or "- (无历史记忆)"

        system = self._build_system_prompt(user_goal)
        context = f"""
短期记忆 (摘要):
{self._compact_memory_summary()}

长期记忆 (相关历史):
{ltm_block}

当前计划:
{state.high_level_plan if state.high_level_plan else "(尚未制定)"}

当前步骤: {state.step}
已用工具调用: {state.tool_calls}
剩余时间: {self._time_left(state) / 1000:.1f}s
""".strip()

        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": context},
        ]

        raw = self.llm.chat(messages, temperature=0.2)
        obj = safe_json_extract(raw)
        if not obj:
            # 容错：如果解析失败，强制结束以防乱跑
            return {"type": "final", "answer": clamp_text(raw, 900)}
        return obj

    def _run_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
        tool = self.tools.get(name)
        if not tool:
            return {"ok": False, "error": f"未知工具: {name}", "data": None, "latency_ms": 0}

        if (not tool.safe) and (not self.cfg.allow_restricted_tools):
            return {"ok": False, "error": f"工具受限: {name}", "data": None, "latency_ms": 0}

        t0 = now_ms()
        try:
            out = tool.fn(args)
            return {"ok": True, "error": None, "data": out, "latency_ms": now_ms() - t0}
        except Exception as e:
            return {"ok": False, "error": str(e), "data": None, "latency_ms": now_ms() - t0}

    def run(self, user_goal: str) -> str:
        state = AgentState()
        self.stm.add("user", user_goal)

        while True:
            stop_reason = self._should_stop(state)
            if stop_reason:
                final = f"安全停止: {stop_reason}。\n\n建议下一步：澄清缺失信息或缩小任务范围。"
                self.stm.add("assistant", final)
                return final

            state.step += 1

            decision = self._model_step(user_goal, state)
            dtype = decision.get("type", "").strip()

            trace_item: Dict[str, Any] = {
                "step": state.step,
                "decision": decision,
                "tool_result": None,
            }

            # 存储记忆笔记 (如果模型提供了)
            mem_note = (decision.get("memory_note") or "").strip()
            if mem_note:
                self.ltm.add(mem_note, tags=["agent_note"])

            if dtype == "plan":
                plan = decision.get("plan") or []
                if isinstance(plan, list) and plan:
                    state.high_level_plan = [str(x)[:140] for x in plan][:6]
                nxt = str(decision.get("next") or "").strip()
                state.last_actions.append("plan:" + nxt)
                self.stm.add("assistant", f"PLAN: {state.high_level_plan}\nNEXT: {nxt}")
                state.trace.append(trace_item)
                continue

            if dtype == "tool_call":
                tool_name = str(decision.get("tool") or "").strip()
                args = decision.get("args") or {}
                if not isinstance(args, dict):
                    args = {}

                state.tool_calls += 1
                action_sig = f"tool:{tool_name} args:{json.dumps(args, sort_keys=True)}"
                state.last_actions.append(action_sig)

                result = self._run_tool(tool_name, args)
                trace_item["tool_result"] = result
                state.trace.append(trace_item)

                obs = {
                    "tool": tool_name,
                    "ok": result["ok"],
                    "error": result["error"],
                    "data": clamp_text(json.dumps(result["data"], ensure_ascii=False), 1200) if result["ok"] else None,
                    "latency_ms": result["latency_ms"],
                }
                self.stm.add("assistant", f"TOOL_OBSERVATION: {json.dumps(obs, ensure_ascii=False)}")
                continue

            # 最终答案
            answer = str(decision.get("answer") or "").strip()
            if not answer:
                answer = "任务结束。"
            self.stm.add("assistant", answer)
            return answer


# -----------------------------
# 内置工具 (安全且实用)
# -----------------------------

def tool_calc(args: Dict[str, Any]) -> Any:
    """安全的计算器"""
    expr = str(args.get("expression", "")).strip()
    if not expr:
        raise ValueError("缺少表达式")
    # 仅允许数字和基本运算符
    if not re.fullmatch(r"[0-9\.\+\-\*\/\(\)\s]+", expr):
        raise ValueError("表达式包含非法字符")
    return eval(expr, {"__builtins__": {}}, {})


def tool_summarize(args: Dict[str, Any]) -> Any:
    """简单的文本摘要工具"""
    text = str(args.get("text", "")).strip()
    max_lines = int(args.get("max_lines", 6))
    text = clamp_text(text, 3000)
    lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
    # 启发式摘要：取前几行 + 关键点（这里简单取前N行）
    out = []
    for ln in lines[:max_lines]:
        out.append(ln[:180])
    return out


def tool_read_file(args: Dict[str, Any]) -> Any:
    """读取本地文本文件"""
    path = str(args.get("path", "")).strip()
    if not path:
        raise ValueError("缺少路径参数")
    # 禁止父目录遍历
    if ".." in path.replace("\\", "/"):
        raise ValueError("禁止访问父目录")
    with open(path, "r", encoding="utf-8") as f:
        return clamp_text(f.read(), 6000)


def build_tools() -> ToolRegistry:
    reg = ToolRegistry()

    reg.register(Tool(
        name="calc",
        description="安全地计算数学表达式 (数字 + - * / 括号)",
        schema={"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]},
        fn=tool_calc,
        safe=True,
    ))

    reg.register(Tool(
        name="summarize",
        description="从长文本中生成简短的要点摘要",
        schema={"type": "object", "properties": {"text": {"type": "string"}, "max_lines": {"type": "integer"}},
                "required": ["text"]},
        fn=tool_summarize,
        safe=True,
    ))

    reg.register(Tool(
        name="read_file",
        description="读取本地文本文件 (禁止父目录遍历)",
        schema={"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]},
        fn=tool_read_file,
        safe=False,  # 文件读取默认视为受限操作
    ))

    return reg


# -----------------------------
# 演示 / Demo
# -----------------------------

if __name__ == "__main__":
    # 初始化 LLM (DeepSeek)
    llm = LLMClient()

    # 构建工具集
    tools = build_tools()

    # 初始化记忆
    ltm = LongTermMemory(path="agent_memory.jsonl")

    # Agent 配置
    cfg = AgentConfig(
        max_steps=10,
        max_tool_calls=6,
        max_seconds=35,
        allow_restricted_tools=False,
    )

    # 实例化 Agent
    agent = ScratchAgent(llm=llm, tools=tools, ltm=ltm, cfg=cfg)

    # 执行任务
    goal = (
        "制定一个简短的旅游计划，目的地是北京，计划包括整个费用，路线，酒店，吃饭，风景名胜古迹等。"
    )

    print("-" * 50)
    print("Agent 开始运行...")
    print("-" * 50)
    result = agent.run(goal)
    print("-" * 50)
    print("最终结果:")
    print(result)

问题：

# 执行任务goal = (    "制定一个简短的旅游计划，目的地是北京，计划包括整个费用，路线，酒店，吃饭，风景名胜古迹等。")

回答：

--------------------------------------------------
Agent 开始运行...
--------------------------------------------------
--------------------------------------------------
最终结果:
**北京3天2晚经典游计划**

**总费用估算：约 1800-2500 元/人** (经济型，不含往返大交通)

**路线与景点：**
- **第1天：** 天安门广场 -> 故宫 -> 景山公园 (俯瞰故宫全景) -> 王府井步行街 (晚餐)
- **第2天：** 八达岭长城 (上午) -> 鸟巢、水立方 (外观，下午) -> 什刹海/后海 (晚餐与夜景)
- **第3天：** 颐和园 (上午) -> 圆明园 (下午部分区域) -> 返程

**酒店：** 选择地铁沿线经济型酒店或连锁酒店 (如如家、汉庭)，约 300-400 元/晚。建议住在东城区、西城区或海淀区地铁站附近。

**吃饭：**
- 每日餐饮预算约 100-150 元/人，可品尝北京烤鸭、炸酱面、卤煮、涮羊肉等。
- 推荐地点：王府井小吃街、前门大街、牛街、以及各大商圈。

**费用明细 (人均估算)：**
- 住宿：2晚 * 350元 = 700元
- 餐饮：3天 * 120元 = 360元
- 门票 (故宫60，长城40，颐和园30，景山2，圆明园10)：约 142元
- 市内交通 (地铁、公交)：约 50元
- 其他 (零食、纪念品)：约 100元

**总计：约 1352元 (基础) + 弹性 = 1800-2500元**

**提示：** 提前在线预订门票，使用地铁APP，避开早晚高峰。

Process finished with exit code 0

4、为什么这个“从零实现”的 Agent 真的有效？

以下是该设计的关键优势：

1) 强制结构化 JSON 输出

当输出不可控时，Agent 就会崩溃。在这个实现中，模型被严格要求进行三选一：

• plan（制定计划）
• tool_call（调用工具）
• final（完成任务）

这看似简单的约束，消除了 80% 的 Agent 幻觉和逻辑混乱。

2) 将工具输出视为“观察结果”

Agent 不会凭空“猜测”世界状态。它必须执行一个动作，获得真实的反馈，然后基于这些反馈更新记忆。这种闭环机制保证了逻辑的严密性。

3) 具备真正的“硬停止”条件

大多数入门教程会忽略这一点，但这恰恰是生产环境的关键。 如果你的 Agent 永远不知道何时停下来，那它就不是一个助手，而是一个失控的后台进程（甚至是一笔巨额的 API 账单）。通过限制步数、时间和检测重复动作，我确保了系统的底线安全。

4) 紧凑的记忆设计

我主动修剪了短期记忆，并将长期记忆设计为基于文件的快照存储。这有效防止了上下文窗口的无限膨胀——这是导致 Agent 性能下降和响应延迟增加的隐形杀手。