1. 什么是AI Agent

AI Agent 可以理解为一个「能自主完成任务的智能体」。它不像普通聊天机器人那样只会根据你的一句话回一句话,而是能够理解目标、制定计划、调用外部工具、观察执行结果,并在多轮循环中逼近最终目标。

一句话概括:AI Agent = 大语言模型(LLM)+ 工具调用 + 记忆 + 规划循环。

举个直观例子:你问普通的聊天助手「北京今天热不热」,它可能凭记忆猜一个答案;而一个具备工具调用能力的 Agent 会真正调用天气查询工具,把北京的实时温度取回来,再结合数据给出回答。

2. AI Agent 的核心架构

一个典型的 AI Agent 通常包含以下四个核心模块:

  • 大语言模型(LLM):负责理解任务、推理和决策,是 Agent 的大脑。
  • 工具(Tools):Agent 可以调用的外部能力,例如搜索、计算、查天气、读写数据库等。
  • 记忆(Memory):保存多轮对话上下文,让 Agent 记得之前说过什么、做过什么。
  • 循环控制(Loop):把「思考 → 调用工具 → 观察结果 → 继续思考」串起来的执行流程。

下面的流程图展示了 Agent 的典型执行链路:

flowchart LR
    A[用户输入] --> B[LLM 推理]
    B --> C{需要调用工具?}
    C -->|是| D[执行工具]
    D --> E[拿到工具结果]
    E --> B
    C -->|否| F[输出最终答案]

其中「需要调用工具吗」这一步,现代大模型可以通过 Function Calling(函数调用)能力自动完成:模型不只返回文字,还会返回一个结构化的工具调用请求,开发者执行完工具后再把结果交还给模型继续推理。

3. 环境准备与最小可用 Agent

本文以 Python 为主,使用 OpenAI 提供的 Chat Completions 接口演示。只要你的模型服务兼容 OpenAI 协议,以下代码基本都能直接跑通。

先安装依赖:

pip install openai

设置环境变量:

export OPENAI_API_KEY="你的 API Key"

下面先实现一个最简 Agent。它只做三件事:接收用户输入、调用 LLM、返回结果,同时把对话记录保存在内存里。

import os
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)
class SimpleAgent:
"""只有“接收输入 -> 调用 LLM -> 返回输出”的最简 Agent。"""
def __init__(self, model: str = "gpt-4o-mini"):
    self.model = model
    self.messages = []
def run(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
response = client.chat.completions.create(
model=self.model,
messages=self.messages,
)
reply = response.choices[0].message.content
self.messages.append({"role": "assistant", "content": reply})
return reply
if name == "main":
agent = SimpleAgent()
while True:
user_input = input("你:")
if user_input.lower() in ("exit", "quit"):
break
reply = agent.run(user_input)
print(f"Agent:{reply}")

这个版本已经能聊天,但它没有任何「手脚」,不能查资料、不能算数、不能操作外部系统。接下来我们给它装上工具。

4. 给 Agent 赋予工具调用能力

实现工具调用最常见的方式是 ReAct 模式:让模型先「推理」该用什么工具,再「行动」调用工具,然后「观察」结果,循环往复直到能给出最终答案。借助 Function Calling,我们不必自己解析模型的文字,只要把工具声明传给接口即可。

第一步,定义两个工具函数:查天气和算数。

def get_weather(city: str) -> str:
    """模拟天气查询,真实项目中可替换为高德、OpenWeatherMap 等 API。"""
    weather_data = {
        "北京": "晴,气温 22~30℃,空气质量优",
        "上海": "多云转小雨,气温 24~29℃",
        "深圳": "晴间多云,气温 26~33℃",
    }
    return weather_data.get(city, f"暂时没有 {city} 的天气数据")
def calculate(expression: str) -> str:
"""执行简单数学计算,生产环境建议使用安全的表达式解析库。"""
try:
result = eval(expression, {"builtins": {}}, {})
return f"计算结果为:{result}"
except Exception as exc:
return f"计算失败:{exc}"

第二步,用 JSON 描述这些工具,让模型知道何时可以调用、需要哪些参数。

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气,参数为城市中文名称",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,例如:北京",
                    }
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学表达式计算,支持加减乘除和括号",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式,例如:2*(3+4)",
                    }
                },
                "required": ["expression"],
            },
        },
    },
]
TOOL_FUNCTIONS = {
"get_weather": get_weather,
"calculate": calculate,
}

第三步,实现带循环控制的核心 Agent。当模型返回工具调用时,我们执行对应函数,再把结果追加回上下文,直到模型给出最终文本。

import json
class ToolAgent:
def init(self, model: str = "gpt-4o-mini"):
self.model = model
self.messages = []
def _call_llm(self):
    response = client.chat.completions.create(
        model=self.model,
        messages=self.messages,
        tools=TOOLS,
    )
    return response.choices[0].message
def run(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
# 最多循环 8 次,避免工具调用陷入死循环
for _ in range(8):
    message = self._call_llm()
# 如果模型没有返回工具调用,说明已经得到最终答案
if not message.tool_calls:
    reply = message.content
    self.messages.append({"role": "assistant", "content": reply})
    return reply
把模型的工具调用请求写回上下文
self.messages.append(message)
for tool_call in message.tool_calls:
name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result = TOOL_FUNCTIONSname
self.messages.append(
    {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result,
    }
)
return "工具调用次数过多,已停止。"</code></pre>
现在测试一下。你问「北京今天天气怎么样,顺便帮我算一下 (23 + 17) * 3 等于多少」,Agent 会先调用天气工具、再调用计算工具,最后整合两个结果回答你。
5. 加入记忆与多轮对话
上面的 Agent 已经把对话记录保存在 self.messages 中,程序运行期间它能记住上下文。但一旦进程结束,记忆就丢失了。我们把它持久化到本地 JSON 文件,让 Agent 重启后仍能记得历史。
import json
import os
MEMORY_FILE = "agent_memory.json"
def save_messages(messages):
with open(MEMORY_FILE, "w", encoding="utf-8") as f:
json.dump(messages, f, ensure_ascii=False, indent=2)
def load_messages():
if not os.path.exists(MEMORY_FILE):
return []
with open(MEMORY_FILE, "r", encoding="utf-8") as f:
return json.load(f)
把记忆加载逻辑接入 Agent 的构造函数即可:
class MemoryAgent(ToolAgent):
def init(self, model: str = "gpt-4o-mini"):
super().init(model)
self.messages = load_messages()
def run(self, user_input: str) -> str:
reply = super().run(user_input)
save_messages(self.messages)
return reply</code></pre>
这样每次对话结束都会自动落盘。更复杂的项目可以把记忆放进 SQLite、向量数据库或 Redis,实现摘要记忆、长期记忆和检索增强。
6. 完整实战:构建一个个人助理 Agent
下面我们把前面的知识串起来,做一个可交互的个人助理。它具备五个能力:查天气、算数、当前时间、添加待办、查看待办。待办清单保存在内存中,方便扩展为数据库存储。
import json
import os
from datetime import datetime
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)
---------- 工具实现 ----------
def get_weather(city: str) -> str:
weather_data = {
"北京": "晴,气温 22~30℃,空气质量优",
"上海": "多云转小雨,气温 24~29℃",
"深圳": "晴间多云,气温 26~33℃",
}
return weather_data.get(city, f"暂时没有 {city} 的天气数据")
def calculate(expression: str) -> str:
try:
result = eval(expression, {"builtins": {}}, {})
return f"计算结果为:{result}"
except Exception as exc:
return f"计算失败:{exc}"
def get_current_time() -> str:
now = datetime.now()
return now.strftime("%Y-%m-%d %H:%M:%S")
TODO_LIST = []
def add_todo(item: str) -> str:
TODO_LIST.append(item)
return f"已添加待办:{item}"
def list_todos() -> str:
if not TODO_LIST:
return "当前没有待办事项"
lines = [f"{index + 1}. {item}" for index, item in enumerate(TODO_LIST)]
return "\n".join(lines)
TOOL_FUNCTIONS = {
"get_weather": get_weather,
"calculate": calculate,
"get_current_time": get_current_time,
"add_todo": add_todo,
"list_todos": list_todos,
}
---------- 工具声明 ----------
def build_tool_schema(name, description, properties, required):
return {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
},
}
TOOLS = [
build_tool_schema(
"get_weather",
"查询指定城市的实时天气",
{"city": {"type": "string", "description": "城市名称"}},
["city"],
),
build_tool_schema(
"calculate",
"执行数学表达式计算",
{"expression": {"type": "string", "description": "数学表达式,如 2*(3+4)"}},
["expression"],
),
build_tool_schema("get_current_time", "获取当前系统时间", {}, []),
build_tool_schema(
"add_todo",
"添加一条待办事项",
{"item": {"type": "string", "description": "待办内容"}},
["item"],
),
build_tool_schema("list_todos", "查看所有待办事项", {}, []),
]
---------- Agent ----------
class AssistantAgent:
def init(self, model: str = "gpt-4o-mini"):
self.model = model
self.messages = []
def _call_llm(self):
response = client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=TOOLS,
)
return response.choices[0].message
def run(self, user_input: str) -&gt; str:
self.messages.append({"role": "user", "content": user_input})
for _ in range(8):
message = self._call_llm()
if not message.tool_calls:
reply = message.content
self.messages.append({"role": "assistant", "content": reply})
return reply
self.messages.append(message)
for tool_call in message.tool_calls:
name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result = TOOL_FUNCTIONSname
self.messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
}
)
return "工具调用次数过多,已停止。"
if name == "main":
agent = AssistantAgent()
print("个人助理已启动,输入 exit 退出。")
while True:
user_input = input("你:")
if user_input.lower() in ("exit", "quit"):
break
reply = agent.run(user_input)
print(f"助理:{reply}")
运行后可以连续测试多轮:
你:现在几点了?
助理:现在是 2026-09-02 19:10:14。
你:帮我记一下,明天下午三点开会。
助理:已添加待办:明天下午三点开会。
你:深圳天气如何?顺便列一下我的待办。
助理:深圳今天晴间多云,气温 26~33℃。你的待办如下:
明天下午三点开会
这段演示完整展示了 Agent 的多工具组合能力:它先理解任务、拆解成多个工具调用、执行后汇总,最后用自然语言给出结论。
7. 常见问题与进阶方向
在从 0 到 1 完成后,实际生产环境还会遇到一些典型问题:
安全:示例中的 eval 只用于演示,生产环境必须替换为安全的表达式解析方案,例如 ast 白名单解析。
上下文长度:对话太长会超出模型上下文窗口,可以引入摘要记忆、滑动窗口或向量检索。
工具可靠性:工具要返回结构化、可读的结果,失败时要给出明确错误信息,方便模型自我纠正。
并行调用:模型一次可能返回多个工具调用,示例中已按顺序处理,生产环境可用异步并发提升速度。
可观测性:给每一步推理和工具调用打日志,便于定位是模型判断错误还是工具执行错误。
进一步进阶可以尝试:接入检索增强生成(RAG)、使用 LangChain 或 LlamaIndex 等框架做编排、给 Agent 加反思机制、实现多 Agent 协作,以及把 Agent 封装成 API 服务供业务系统调用。
8. 总结
开发一个 AI Agent 的核心并不神秘,关键链路可以归纳为四步:定义目标、为模型声明可用工具、实现「推理 → 调用 → 观察」循环、沉淀记忆。本文从最小可用版本出发,逐步加入工具调用、持久化记忆,最终完成了一个可交互的个人助理。
建议你把示例代码完整跑一遍,再根据自己的业务场景替换工具实现。看懂这个最小闭环之后,你已经站在了 AI Agent 开发的起点上。
Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐