1. 为什么需要 AI Agent

传统聊天机器人主要做“生成文字”：你问一句，它根据上下文答一句，不会主动查数据、调接口，也不会自己拆分任务。AI Agent 则是由大语言模型驱动的自动化系统，它的核心特征是：

  • 目标导向：能理解用户目标，并拆成多个步骤。
  • 工具调用：可以查询天气、执行计算、检索文档、操作数据库等。
  • 环境反馈：观察工具返回结果，再决定下一步。
  • 持续迭代：像 Thought → Action → Observation 一样循环，直到完成任务。

一句话概括：AI Agent = 大模型 + 工具 + 记忆 + 执行循环 + 环境反馈。

对工程师来说，入门 AI Agent 不以“写 Prompt”为终点，而是要掌握如何设计模型可调用的工具、如何写稳定的 Agent 循环、如何处理上下文与记忆，以及如何做安全护栏。

2. 核心组成与运行流程

一个典型 AI Agent 的组成如下：

用户目标

LLM 大脑

规划与拆解

调用工具

环境 / API / 数据库

观察结果

最终答案

我们通常需要实现五个部分：

  1. 模型入口：调用 LLM，拿到文本或工具调用请求。
  2. 工具注册表：声明工具名称、参数 Schema、执行函数。
  3. Agent 循环：发现工具调用就执行工具，把结果塞回上下文。
  4. 记忆系统：保存多轮对话、文档片段或用户偏好。
  5. 安全与观测：记录日志、限制危险操作、人工确认。

下面从零开始，写一个可以运行的最小 AI Agent。我们使用 Python 和 OpenAI 兼容接口，兼容 OpenAI、DeepSeek、Qwen 等大多数模型服务。

3. 开发环境准备

推荐环境：Python 3.10+。

依赖如下：

pip install openai python-dotenv numpy

在项目根目录创建 .env：

OPENAI_API_KEY=你的密钥
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4.1-mini

如果你使用 DeepSeek，可以改成：

OPENAI_BASE_URL=https://api.deepseek.com/v1
OPENAI_MODEL=deepseek-chat

加载配置并初始化客户端：

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)

MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini")

4. 先理解模型对话：单轮生成

在开始 Agent 之前，先确认模型可正常调用：

def chat_once(user_text: str) -> str:
    messages = [
        {"role": "system", "content": "你是一个专业的 AI 助手，回答简洁准确。"},
        {"role": "user", "content": user_text},
    ]
    resp = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        temperature=0.1,
    )
    return resp.choices[0].message.content

print(chat_once("用两句话说明什么是 AI Agent。"))

这一步跑通后，就可以进入真正的 Agent 开发。

5. 完整可运行 Agent：ReAct 工具调用循环

下面实现一个核心函数：模型说“我要调用工具”，我们就执行工具；模型继续观察结果，直到给出最终答案。

import json

SYSTEM_PROMPT = """你是一个 AI Agent。你可以使用提供的工具完成用户请求。
如果当前信息不足以回答，请先调用工具；工具返回结果后继续推理。
只有当你已经获得足够信息时，才直接回复最终答案，不要编造未经验证的数据。"""

def call_llm(messages, tools):
    resp = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=tools,
        temperature=0,
    )
    return resp.choices[0].message

def run_agent_from_messages(messages, tools, execute_tool, max_steps=5):
    for _ in range(max_steps):
        msg = call_llm(messages, tools)
        messages.append(msg.model_dump(exclude_none=True))

        # 没有工具调用，说明模型已经给出最终答案
        if not msg.tool_calls:
            return msg.content, messages

        for tool_call in msg.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments or "{}")
            result = execute_tool(name, args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

    return "已达到最大步数，任务未完成。", messages

def run_agent(user_query, tools, execute_tool, max_steps=5):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_query},
    ]
    return run_agent_from_messages(messages, tools, execute_tool, max_steps)

这个 run_agent 就是一个最小但完整的 Agent 循环。它可以自动处理一次或多次工具调用。

6. 工具开发：注册表与输入校验

接下来实现两个工具：calculator 和 get_weather。

import ast
import operator

ALLOWED_BINARY_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
}

ALLOWED_UNARY_OPS = {
    ast.UAdd: operator.pos,
    ast.USub: operator.neg,
}

def safe_eval(expr: str):
    def _eval(node):
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in ALLOWED_BINARY_OPS:
            left = _eval(node.left)
            right = _eval(node.right)
            return ALLOWED_BINARY_OPS[type(node.op)](left, right)
        if isinstance(node, ast.UnaryOp) and type(node.op) in ALLOWED_UNARY_OPS:
            operand = _eval(node.operand)
            return ALLOWED_UNARY_OPS[type(node.op)](operand)
        raise ValueError("不支持的表达式")

    return _eval(ast.parse(expr, mode="eval"))

def execute_tool(name: str, args: dict) -> str:
    if name == "calculator":
        try:
            value = safe_eval(args["expression"])
            return f"计算结果为：{value}"
        except Exception as exc:
            return f"计算失败：{exc}"

    if name == "get_weather":
        city = args.get("city", "未知城市")
        return f"{city} 今天晴，气温 22~29℃，空气质量优。"

    return f"未知工具：{name}"

这里用 ast 解析数学表达式，而不是直接使用 eval，可以避免执行任意代码。生产环境中，工具输入必须做严格校验。

接下来声明工具的 JSON Schema：

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "计算一个数学表达式，支持 + - * / ** 和括号。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "要计算的数学表达式，例如 123 * 456",
                    }
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市今天的天气。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名，例如 北京",
                    }
                },
                "required": ["city"],
            },
        },
    },
]

运行一次多工具配合的任务：

answer, _ = run_agent(
    "请计算 123 * 456 - 789，并查看北京天气。",
    TOOLS,
    execute_tool,
)
print(answer)

模型会先调用 calculator，再调用 get_weather，最后汇总回答。

7. 多轮记忆：让 Agent 记得上下文

前面的 run_agent 每次都从头开始，无法记住上一轮。我们加一个简单的记忆类。

class ConversationMemory:
    def __init__(self, max_messages: int = 20):
        self.max_messages = max_messages
        self.messages = []

    def add(self, message: dict):
        self.messages.append(message)
        self._trim()

    def _trim(self):
        if len(self.messages) <= self.max_messages:
            return
        system = [m for m in self.messages if m["role"] == "system"]
        non_system = [m for m in self.messages if m["role"] != "system"]
        keep_num = self.max_messages - len(system)
        self.messages = system + non_system[-keep_num:]

def run_agent_with_memory(user_query, tools, execute_tool, memory):
    memory.add({"role": "user", "content": user_query})
    answer, messages = run_agent_from_messages(
        memory.messages,
        tools,
        execute_tool,
    )
    memory.messages = messages
    return answer

使用方式：

memory = ConversationMemory(max_messages=20)
memory.messages.append({"role": "system", "content": SYSTEM_PROMPT})

print(run_agent_with_memory("北京天气如何？", TOOLS, execute_tool, memory))
print(run_agent_with_memory("那上海呢？", TOOLS, execute_tool, memory))

第二轮中，模型可以从上下文里知道“那”指的是天气，继续保持对话。

8. 接入知识库：一个轻量 RAG 工具

真实 Agent 往往需要查资料。我们可以用一个最简单的向量检索实现 RAG。

import numpy as np

def embed_texts(texts):
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return np.array([d.embedding for d in resp.data])

class SimpleRAG:
    def __init__(self):
        self.docs = []
        self.vectors = None

    def add_documents(self, docs):
        if not docs:
            return
        self.docs = docs
        self.vectors = embed_texts(docs)

    def search(self, query, top_k=3):
        if self.vectors is None:
            return []
        qv = embed_texts([query])[0]
        norm_docs = np.linalg.norm(self.vectors, axis=1)
        norm_q = np.linalg.norm(qv)
        scores = self.vectors @ qv / (norm_docs * norm_q + 1e-8)
        idx = np.argsort(scores)[::-1][:top_k]
        return [self.docs[i] for i in idx]

再把它包装成工具：

rag = SimpleRAG()
rag.add_documents([
    "AI Agent 是由大语言模型驱动的自动化系统，可以调用工具并迭代完成任务。",
    "ReAct 模式将推理与行动交替进行，适合实现可解释的 Agent。",
    "工具调用允许模型请求外部系统执行计算、查询天气或检索文档。",
])

def retrieve_knowledge(query: str) -> str:
    docs = rag.search(query, top_k=3)
    return "\n---\n".join(docs)

如果你使用的 OPENAI_BASE_URL 不支持 text-embedding-3-small，请替换成该平台提供的 embedding 模型名。

9. 工程化建议

入门之后，建议从以下几个方面继续提升：

  • 输入校验：所有工具参数用 Pydantic 或 JSON Schema 严格校验。
  • 日志与追踪：记录每一步模型决策、工具入参、工具结果，便于排查。
  • 错误与重试：工具调用失败时返回可理解错误，并允许模型重试。
  • 危险操作护栏：删除数据、发送消息、执行代码等操作需要人工确认。
  • 成本控制：设置最大步数、Token 上限和单次对话超时。
  • 评估体系：建立任务集，评估 Agent 的完成率、调用准确率和回复质量。

10. 实战项目：从天气助手到个人知识库 Agent

建议按以下顺序做三个项目：

  1. 天气 + 计算助手：完成上文的 run_agent 和两个工具。
  2. 多轮任务 Agent：加入记忆，让 Agent 能追问、澄清和连续执行任务。
  3. 个人知识库 Agent：把你的笔记、文档导入 SimpleRAG，让 Agent 先检索再回答。

一个简单的目录结构如下：

ai-agent-demo/
├── .env
├── main.py
├── agent.py
├── tools.py
├── memory.py
└── rag.py

main.py 可以这样串联：

from dotenv import load_dotenv
from openai import OpenAI
from agent import SYSTEM_PROMPT, run_agent_with_memory
from tools import TOOLS, execute_tool
from memory import ConversationMemory

load_dotenv()

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)

memory = ConversationMemory(max_messages=30)
memory.messages.append({"role": "system", "content": SYSTEM_PROMPT})

while True:
    user_input = input("你：")
    if user_input.lower() in {"exit", "quit", "q"}:
        break
    answer = run_agent_with_memory(user_input, TOOLS, execute_tool, memory)
    print("Agent：", answer)

11. 常见问题与避坑

  • 模型不调用工具怎么办？
    检查工具描述是否足够清晰，参数名和说明是否明确；“手写 ReAct JSON”容易解析失败，优先使用平台原生 function calling。

  • Agent 输出大量无效工具调用？
    设置 max_steps，工具失败时返回具体错误，而不是把原始异常直接送回模型。

  • 多轮对话丢失关键信息？
    使用记忆窗口时，要保留下 system 消息；如果上下文过长，可以引入摘要记忆或长期记忆。

  • RAG 检索不准确？
    先检查文档切片是否合适，再考虑加入重排序、元数据过滤和查询改写。

  • 代码执行安全性？
    不要在模型可触达的环境中直接运行任意 Python、SQL 或 Shell；建议放在沙箱中，或干脆禁止危险工具。

12. 总结

学习 AI Agent 工程师的关键不是“调好几个 Prompt”，而是掌握：

  • 用 function calling 设计工具调用；
  • 用循环实现稳定的 Agent 执行；
  • 用记忆系统维护上下文；
  • 用 RAG 增强知识能力；
  • 用校验、日志和人工确认保证安全。

把上面的代码完整跑通，你已经拥有一个可扩展的 AI Agent 骨架。后续可以逐步替换成 LangChain、LangGraph 等框架，也可以基于工作流引擎做多 Agent 协作。

# 学习 # 人工智能 # android studio # 前端
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

AI Agent编排：复杂工作流的自动化实现

在深入探讨Agent编排之前，我们首先需要明确什么是AI Agent。AI Agent是一个能够感知环境、做出决策并采取行动的自主实体。感知模块：接收和处理环境信息推理/决策模块：基于感知信息进行思考和决策行动模块：执行具体的操作（如调用工具、生成文本）记忆模块：存储历史信息和知识通信模块：与其他Agent或系统进行交互Agent编排（Agent Orchestration）是指对多个Agent的

avatar AI Agent技术社区
cover

90%的人认为RAG和Agent分开做都不难，合在一起就废？可写简历:RAG+Agent合体全链路方案

avatar AI Agent技术社区
cover

阶跃星辰 StepAudio 3：五款语音模型一次放出，全双工对话拿下两个全球第一

avatar AI Agent技术社区