大模型 Agent：从 Function Calling 到记忆增强的工程实现

原创

IT大佬 jzit-top

发布于 2026-09-28 16:33:44

500

大模型 Agent 的本质，是让 LLM 从"文本生成器"升级为"决策中枢"：它不再只输出答案，而是输出动作，由外部执行器落地，再把结果回灌，形成闭环。专业实现的核心不在提示词，而在三件事：原生工具调用、分层记忆、可评估的循环。

一、Agent 的最小构成

一个可用的 Agent 包含四个部件：

LLM：推理与决策，输出结构化动作。
Tools：可执行的函数集合，带 JSON Schema 描述。
Memory：短期上下文 + 长期向量库。
Loop：观察→决策→执行→回灌，带终止条件。

其中最容易做错的是工具调用方式。用提示词让模型"输出 JSON 再解析"是脆弱方案，现代模型已原生支持 Function Calling，应优先使用。

二、原生 Function Calling

工具用 JSON Schema 声明，模型返回 tool_calls，由代码执行后回灌。

import os, json, subprocess
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

TOOLS_SPEC = [
    {
        "type": "function",
        "function": {
            "name": "run_sql",
            "description": "执行只读SQL查询，返回JSON结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SELECT语句"}
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_docs",
            "description": "在内部知识库中检索文档片段",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "top_k": {"type": "integer", "default": 3},
                },
                "required": ["query"],
            },
        },
    },
]

执行器必须做安全校验，绝不能直接执行模型生成的任意命令：

def run_sql(query: str) -> str:
    q = query.strip().lower()
    if not q.startswith("select"):
        return "拒绝：仅允许 SELECT 查询"
    for kw in ("drop", "delete", "update", "insert", ";"):
        if kw in q.replace("select", "", 1):
            return f"拒绝：包含危险关键字 {kw}"
    # 实际执行走只读连接、超时、行数上限
    return json.dumps({"rows": [], "note": "示例未连接真实库"})

def search_docs(query: str, top_k: int = 3) -> str:
    # 接入向量库，此处返回模拟结果
    return json.dumps([{"title": f"doc_{i}", "score": 0.9 - i * 0.1}
                       for i in range(top_k)], ensure_ascii=False)

DISPATCH = {"run_sql": run_sql, "search_docs": search_docs}

三、分层记忆

短期记忆是对话历史，长期记忆需要向量检索。关键在于何时写入、何时召回。

import numpy as np
from openai import OpenAI
client = OpenAI()

def embed(text: str) -> list[float]:
    r = client.embeddings.create(model="text-embedding-3-small", input=text)
    return r.data[0].embedding

class VectorMemory:
    def __init__(self, path="memory.jsonl"):
        self.path = path
        self.items = []          # [(vec, text)]
        self._load()

    def _load(self):
        try:
            with open(self.path, encoding="utf-8") as f:
                for line in f:
                    d = json.loads(line)
                    self.items.append((np.array(d["vec"]), d["text"]))
        except FileNotFoundError:
            pass

    def add(self, text: str):
        vec = np.array(embed(text))
        self.items.append((vec, text))
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps({"vec": vec.tolist(), "text": text},
                               ensure_ascii=False) + "\n")

    def search(self, query: str, top_k=3) -> list[str]:
        if not self.items:
            return []
        q = np.array(embed(query))
        # 余弦相似度
        sims = [(q @ v / (np.linalg.norm(q) * np.linalg.norm(v) + 1e-9), t)
                for v, t in self.items]
        sims.sort(reverse=True)
        return [t for _, t in sims[:top_k]]

召回时机很重要：不是每轮都检索，而是在模型判断需要外部知识时触发。可以用一个轻量分类器或直接让模型在决策阶段调用 search_docs。

四、主循环：带记忆的 Agent

SYS = """你是任务型助手。规则：
1. 需要外部信息时调用工具，不要臆测
2. 同一工具连续调用两次且参数相同，必须换策略
3. 信息足够时直接回答，不再调用工具
"""

def agent(question: str, memory: VectorMemory, max_steps=6) -> str:
    # 召回相关长期记忆，注入上下文
    recalled = memory.search(question)
    context = "\n".join(f"- {r}" for r in recalled) or "无"

    messages = [
        {"role": "system", "content": SYS},
        {"role": "system", "content": f"相关历史记忆：\n{context}"},
        {"role": "user", "content": question},
    ]

    last_call = None
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o", messages=messages,
            tools=TOOLS_SPEC, tool_choice="auto", temperature=0,
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            memory.add(f"Q: {question}\nA: {msg.content}")   # 写入长期记忆
            return msg.content

        for call in msg.tool_calls:
            name = call.function.name
            args = json.loads(call.function.arguments)

            # 死循环防护
            signature = (name, json.dumps(args, sort_keys=True))
            if signature == last_call:
                messages.append({"role": "tool", "tool_call_id": call.id,
                                 "content": "重复调用，请更换策略或直接回答"})
                continue
            last_call = signature

            fn = DISPATCH.get(name)
            try:
                result = fn(**args) if fn else f"未知工具 {name}"
            except Exception as e:
                result = f"执行异常：{e}"

            messages.append({"role": "tool", "tool_call_id": call.id,
                             "content": result[:2000]})

    return "达到步数上限，未完成"

mem = VectorMemory()
print(agent("帮我查一下上季度的销售数据并总结", mem))

五、评估与护栏

Agent 的评估不能只看单轮准确率，要看任务完成率、步数、工具调用正确率、成本。

def evaluate(cases, memory):
    stats = {"success": 0, "steps": 0, "cost_tokens": 0}
    for c in cases:
        out = agent(c["question"], memory)
        ok = all(k in out for k in c["must_contain"])
        stats["success"] += int(ok)
    stats["accuracy"] = stats["success"] / len(cases)
    return stats

工程护栏清单：

工具白名单：只注册必要函数，破坏性操作一律禁止。
参数校验：Schema 之外再加业务校验（如 SQL 只读）。
步数与成本上限：超限即停，防止失控循环。
幂等设计：写操作带唯一键，重试不产生副作用。
可观测：每步记录 thought、tool、args、result，支持回放。
人工审批：高风险动作（转账、删除、发布）走确认流程。

结语

大模型 Agent 的专业实现，不是把提示词写得更花哨，而是用原生 Function Calling 保证决策可靠、用分层记忆突破上下文限制、用评估与护栏让自主可控。当 Agent 能稳定完成多步任务且成本可预测，它才从演示走向生产。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

大模型部署

agent