很多人说的「Agent」，在工程上往往可以拆成四件事：记住刚才说了什么、让模型能调用你写的函数、对话太长时怎么处理、要不要跨会话记住用户。下面用同一套技术栈（OpenAI 兼容的 Chat Completions + 可选本地向量模型）说明原理，并配上可直接对照实现的代码片段。

---

1. 先统一心智：模型本身不替你「记着聊天」

每次调用接口，模型只处理你这一次传过去的文字。所以要自己维护一个列表，按时间顺序把「系统设定、用户话、模型话、工具结果」放进去，下一轮原样再发给模型——这就是最常见的短期记忆。

技术点：messages 是一个列表，元素是字典，至少包含 role 和 content。常见角色：

role	含义
system	开发者写的人设、规则
user	用户输入
assistant	模型的回复（有时还带「我要调哪个工具」的结构）
tool	你执行工具后返回给模型的结果

代码示例：初始化并追加一轮对话

messages = [
    {"role": "system", "content": "你是一个简洁、准确的助手。"},
]

user_text = input("你：")
messages.append({"role": "user", "content": user_text})

# 下面会调用模型；得到回复后再 append assistant
# messages.append({"role": "assistant", "content": reply})

要点：谁说了什么，就按顺序往 messages 里堆；请求时把整个列表传给 chat.completions.create(..., messages=messages)。

---

2. 工具调用（Function Calling）：模型「点菜」，你「做菜」

模型不能直接打开你电脑上的文件、不能写你的数据库，所以要约定一套协议：模型在回复里声明要调用的函数名和参数（JSON），你的程序真的去执行对应 Python 函数，再把返回值以固定格式塞回 messages，再调一次模型——直到某一轮模型不再要工具，只输出最终给用户看的文字。

技术点：

1. 给模型一份「菜单」：函数名、说明、参数 JSON Schema（很多 SDK 里叫 tools）。
2. 给程序一份「真函数表」：名字 → 可调用对象（如字典映射）。
3. 若返回里带有 tool_calls，对每一个调用：解析参数 → 执行函数 → 追加一条 role: "tool" 的消息，且必须带上 tool_call_id，和模型那条 assistant 里的 id 对上。

代码示例：菜单 + 真函数表

import json

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取本地文本文件的全部内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "文件路径"},
                },
                "required": ["file_path"],
            },
        },
    },
]


def read_file(file_path: str) -> str:
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()


TOOL_MAP = {
    "read_file": read_file,
}

代码示例：内层循环——直到不再出现工具调用

# 假设已有 client、MODEL、messages、TOOLS
while True:
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
    )
    message = response.choices[0].message

    if getattr(message, "tool_calls", None):
        # 必须把「带 tool_calls 的 assistant」留在上下文里
        messages.append(message)
        for tool_call in message.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            result = TOOL_MAP[name](**args)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": name,
                    "content": str(result),
                }
            )
        # 继续下一轮：让模型根据工具结果再推理
        continue

    # 没有工具调用，本轮回复就是给用户看的正文
    reply = message.content or ""
    break

messages.append({"role": "assistant", "content": reply})

为什么要 while True 套一层？
一次工具调用往往不是终点：模型可能根据结果再调别的工具，或再调同一个工具。内层循环表达的是：「只要模型还想调工具，就执行并再问，直到它只说话」。

---

3. 外层循环：终端 / Web 都是同一套路

上面处理的是「用户已经说了一句」之后的事。整体再包一层：读输入 →（可选压缩上下文）→ 把用户话放进 messages → 跑上一节的内层循环 → 打印或返回回复。

代码示例：外层 + 内层骨架

messages = [{"role": "system", "content": "你是一个助手。"}]

while True:
    user_input = input("你：").strip()
    if not user_input:
        continue

    messages.append({"role": "user", "content": user_input})

    reply = ""
    while True:
        response = client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS, tool_choice="auto"
        )
        message = response.choices[0].message
        if getattr(message, "tool_calls", None):
            messages.append(message)
            for tool_call in message.tool_calls:
                name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)
                result = TOOL_MAP[name](**args)
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "name": name,
                        "content": str(result),
                    }
                )
            continue
        reply = message.content or ""
        break

    messages.append({"role": "assistant", "content": reply})
    print("助手：", reply)

---

4. 对话太长怎么办：压缩（用模型总结模型看过的历史）

messages 越长，费用和延迟越高，还可能超出上下文窗口。常见做法是：当序列化后的长度超过阈值，再起一次 Chat 请求，让模型把旧对话压缩成一小段摘要，然后扔掉明细，只保留「原来的 system」+「一条承载摘要的 system（或 user）」。

技术点：

• 先把「要长期保留的人设」那条 system 抠出来，只对其余消息算长度或做摘要。
• 摘要失败时要有保底（例如只保留最近几条），避免程序崩掉。

代码示例：按 JSON 字符串长度决定是否压缩

import json


def compact_context(messages: list, client, model: str, max_chars: int = 10_000) -> list:
    system_msg = messages[0] if messages and messages[0]["role"] == "system" else None
    rest = messages[1:] if system_msg else messages

    serializable = [
        m.model_dump() if hasattr(m, "model_dump") else m for m in rest
    ]
    blob = json.dumps(serializable, ensure_ascii=False)
    if len(blob) <= max_chars:
        return messages

    prompt = (
        "请将以下对话压缩为不超过 300 字的摘要，保留事实与结论，省略工具调用的细节：\n"
        + blob
    )
    try:
        r = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
        )
        summary = r.choices[0].message.content or ""
        new_messages = []
        if system_msg:
            new_messages.append(system_msg)
        new_messages.append(
            {"role": "system", "content": f"此前对话摘要：{summary}"}
        )
        return new_messages
    except Exception:
        return messages[-5:]  # 保底：只留最近几条

取舍：压缩省 token、省窗口，但细节不可恢复；和「清空列表」不同——清空是硬重置，连人设是否保留都要自己设计。

---

5. 跨会话「长期记忆」：向量检索是最小闭环

短期记忆只在内存里的 messages 里；关掉进程就没了。若要让模型「过几天还记得用户的偏好」，需要持久化。教学里常用极简做法：

1. 每条记忆是一段文本，落盘（例如 JSON 里一个字符串列表）。
2. 用句子向量模型把每条文本变成向量，和文本存在一起（下标对齐）。
3. 用户提问时，把问题也编成向量，和所有记忆向量算相似度，取最相关的几条拼成字符串，通过工具交给模型，或你在发请求前写进 system/user。

技术点：sentence-transformers 或任意 Embedding API；相似度可用归一化后的点积（等价于余弦相似度，当向量已归一化时）。

代码示例：保存与检索（CPU 可跑的小模型示例）

import json
from pathlib import Path

import numpy as np
from sentence_transformers import SentenceTransformer

DB_PATH = Path("memory_db.json")
model = SentenceTransformer("all-MiniLM-L6-v2")


def load_db():
    if not DB_PATH.exists():
        return {"docs": [], "embeddings": []}
    with open(DB_PATH, "r", encoding="utf-8") as f:
        return json.load(f)


def save_db(data):
    with open(DB_PATH, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def save_memory(text: str) -> None:
    data = load_db()
    data["docs"].append(text)
    vec = model.encode(text).astype(np.float32).tolist()
    data["embeddings"].append(vec)
    save_db(data)


def query_memory(query: str, top_k: int = 3, min_score: float = 0.5) -> str:
    data = load_db()
    docs = data["docs"]
    embs = data["embeddings"]
    if not docs:
        return ""
    q = model.encode(query).astype(np.float32)
    mat = np.array(embs, dtype=np.float32)
    # 余弦相似度：点积 / (||q|| * ||行||)
    denom = np.linalg.norm(q) * np.linalg.norm(mat, axis=1) + 1e-8
    scores = np.dot(mat, q) / denom
    idx = np.argsort(scores)[::-1][:top_k]
    hits = [docs[i] for i in idx if scores[i] >= min_score]
    return "\n".join(hits)

把 save_memory / query_memory 再包装成 第 2 节那种 TOOLS + TOOL_MAP，模型就能在对话里自己决定「什么时候记、什么时候查」。这就是迷你版 RAG 思路：检索 → 把结果塞进上下文 → 再生成。

---

6. 和「OpenAI 兼容网关」怎么接

很多国内服务提供与 OpenAI SDK 相同的 base_url + api_key 用法：实例化客户端时换掉地址和密钥即可，chat.completions.create 与 tools 参数形态一致。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["YOUR_API_KEY"],
    base_url=os.environ["YOUR_BASE_URL"],  # 兼容网关地址
)
MODEL = "你在网关上实际可用的模型名"

注意：密钥和地址应来自环境变量或密钥管理，不要写进仓库。

---

7. 安全与工程上别忽略的几句话

• 若工具包含任意路径读写，等价于把文件系统的一部分交给模型——只适合可信环境；对外必须白名单、沙箱或专用存储层。
• read_image_as_base64 这类工具只是把文件变成字符串；多模态还要按你所用网关的文档，把图片以正确字段放进 messages（不同厂商字段名不同）。
• 工具执行失败时，把错误信息也作为 tool 的 content 返回，模型才能改口或换策略。

---

8. 小结（背这几句就够串起来）

1. 短期记忆 = 自己维护 messages，每轮原样提交。
2. 工具 = 模型声明调用 → 你执行 → role=tool + tool_call_id 写回 → 再请求，内层循环直到不再调用。
3. 太长 = 阈值触发二次请求做摘要，用摘要换明细。
4. 长期记忆 = 文本落盘 + 向量相似度检索，结果再进入 messages 或工具返回。

掌握以上四块，再去看各种 Agent 框架，你会清楚：框架多半是在替你管理同样的消息流、同样的工具循环、同样的记忆与观测；协议层仍是这一套。

---

本文侧重原理与实现套路，代码为示意级片段，可按你的网关与业务替换模型名、工具与安全策略。