The user wants a technical article about "智能体" (AI Agents). Need to produce JSON with title, summary, tags, markdown. 1500-3000 characters (字). At least one runnable code example. Structure: background/problem, design, implementation steps, code, notes, summary. Tags 3-5, prefer common tech tags like AIGC, Python, etc.
I should write a concrete, runnable Python example of a minimal agent loop with tool calling — but must not fabricate APIs. I can use a self-contained example that doesn't depend on a specific LLM SDK, e.g., implement a ReAct-style loop with a pluggable LLM interface, and provide a mock LLM for testing so it's runnable without API keys. Or use OpenAI-compatible HTTP API via requests — that's stable and real. Better: write a minimal agent framework with tool registry in Python, using an OpenAI-compatible chat completions endpoint via requests (widely supported, e.g. DashScope compatible mode). But to be safe and runnable, provide a MockLLM so the code runs offline, and note how to swap in a real model.
Let me design a good article:
Title: 从零实现一个可运行的智能体循环：工具调用、状态管理与失败重试
Sections:
- 背景/问题: LLM 单轮问答的局限; 智能体 = 模型 + 工具 + 循环 + 记忆; 工程上的问题: 循环终止、工具schema、错误处理、幂等、成本控制、可观测性。
- 方案设计:
- 核心循环: 输入 -> 规划 -> 调用工具 -> 观察 -> 直到终止
- 组件: LLM 客户端抽象、工具注册表(JSON Schema)、执行器(超时/重试/权限)、上下文/记忆裁剪、轨迹记录
- 状态机图说明
- 实现步骤:
- step1 定义工具注册表 + schema
- step2 实现 LLM 抽象 (Mock + OpenAI 兼容)
- step3 实现 agent loop with max_steps, 重复调用检测
- step4 可观测性 (trace)
- 代码: complete runnable Python single file. Use only stdlib. Mock LLM that returns tool calls in a fixed format.
- 注意事项: prompt injection, 工具权限最小化, 幂等与副作用, 上下文膨胀, 终止条件, 成本, 评测
- 总结
Let me write the code carefully so it actually runs.
Design:
# agent.py
from __future__ import annotations
import json, re, time, uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol
Tool registry:
@dataclass
class Tool:
name: str
description: str
parameters: dict # JSON Schema
func: Callable[..., Any]
timeout_s: float = 5.0
class ToolRegistry:
def __init__(self):
self._tools = {
}
def register(self, tool): ...
def schema_text(self) -> str: ...
def call(self, name, args) -> str # returns string observation, catches exceptions
Guard: limit args, catch Exception, return error string.
LLM interface:
class LLM(Protocol):
def chat(self, messages: list[dict]) -> str: ...
Mock LLM: parse messages, if no observation yet -> output tool call for calculator; after observation -> final answer. Simple heuristic.
Action parsing: use a simple format:
Thought / Action block:
<tool_call>{"name": "...", "arguments": {...}}</tool_call>
Or final answer <final>...</final>.
Simpler: define protocol in the system prompt: model must output either JSON with {"type":"tool_call", ...} or {"type":"final", "answer": ...}. Use JSON parsing with fallback.
MockLLM: examines last message; if role == "user" -> plan tool call; if role == "tool" -> produce final answer that includes observation.
Let me write:
class MockLLM:
"""离线可运行的假模型，仅用于验证 Agent 循环，不产生真实语言能力。"""
def __init__(self):
self.turn = 0
def chat(self, messages):
self.turn += 1
tool_msgs = [m for m in messages if m["role"] == "tool"]
if not tool_msgs:
return json.dumps({
"type": "tool_call",
"name": "calculator",
"arguments": {
"expression": "(128 * 3 + 40) / 4"},
"thought": "需要先算出人均成本"
}, ensure_ascii=False)
last = tool_msgs[-1]["content"]
return json.dumps({
"type": "final",
"answer": f"计算结果为 {last}，这就是人均成本。"
}, ensure_ascii=False)
Wait — the calculator tool executes safely. Use ast-based safe eval, not eval. That's better practice and demonstrates a note about安全。
Calculator tool:
import ast, operator
_OPS = {
ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod, ast.USub: operator.neg, ast.UAdd: operator.pos}
def safe_eval(expr: str) -> float:
def _eval(node):
if isinstance(node, ast.Expression): return _eval(node.body)
if isinstance(node, ast.Constant):
if isinstance(node.value, (int, float)): return node.value
raise ValueError("只允许数字常量")
if isinstance(node, ast.BinOp):
op = _OPS.get(type(node.op))
if op is None: raise ValueError("不支持的运算符")
return op(_eval(node.left), _eval(node.right))
if isinstance(node, ast.UnaryOp):
op = _OPS.get(type(node.op))
...
raise ValueError("不支持的表达式")
return _eval(ast.parse(expr, mode="eval"))
Agent loop:
@dataclass
class AgentResult:
answer: str
steps: int
trace: list[dict] = field(default_factory=list)
class Agent:
def __init__(self, llm, registry, max_steps=6, max_repeat=2, verbose=True):
...
def run(self, task: str) -> AgentResult:
messages = [{
"role":"system","content": SYSTEM_PROMPT + self.registry.schema_text()},
{
"role":"user","content": task}]
seen = {
}
for step in range(1, self.max_steps+1):
raw = self.llm.chat(messages)
action = parse_action(raw) # raises on invalid
...
Handle: invalid JSON -> append a "observation" telling model format error and continue (retry). That's a real design point.
Signature for repetition detection: name + json.dumps(args, sort_keys=True).
If repeated > max_repeat -> stop with fallback.
Also record trace with timing.
parse_action:
def parse_action(raw: str) -> dict:
text = raw.strip()
# 容忍 ```json fence
if text.startswith("```"):
text = re.sub(r"^```[a-zA-Z]*\n?|\n?```$", "", text).strip()
try:
data = json.loads(text)
except json.JSONDecodeError as e:
raise ActionFormatError(f"输出不是合法 JSON: {e}")
if not isinstance(data, dict) or data.get("type") not in ("tool_call","final"):
raise ActionFormatError("缺少 type 字段或取值非法")
...
Now the main: build registry with calculator + a "get_weather" mock? One tool is enough, but two shows registry. Keep calculator plus a word_count? Let's keep calculator only but mention registering more. Actually add a second tool now returning fixed date? Might be unnecessary. I'll add two: calculator and unit_lookup? Keep it simple: calculator + read_kv? I'll just do calculator, plus show registering another tool in "扩展" snippet.
Hmm, better to demonstrate registry with two tools so schema listing is meaningful. Add current_time that returns time — that's fine and deterministic-ish. But MockLLM only uses calculator. That's fine.
Also need OpenAI-compatible real LLM adapter:
class OpenAICompatLLM:
def __init__(self, base_url, api_key, model, timeout=60):
...
def chat(self, messages):
import urllib.request
payload = {
"model": self.model, "messages": messages, "temperature": 0}
req = urllib.request.Request(self.base_url.rstrip("/") + "/chat/completions",
data=json.dumps(payload).encode(), headers={
...})
with urllib.request.urlopen(req, timeout=self.timeout) as r:
data = json.loads(r.read())
return data["choices"][0]["message"]["content"]
That's real OpenAI-compatible API. Note: uses only stdlib urllib. Good.
Add retry wrapper for LLM call with exponential backoff.
Also context trimming: keep system + last N messages.
Let me keep the code compact but complete, ~120 lines. Word count: article 1500-3000 字. Chinese chars count. Code doesn't count much toward 字 but fine.
Now write markdown.
I need to double check the agent loop message handling: after tool call, append assistant message with raw content, then append tool message:
messages.append({
"role":"assistant","content":raw})
messages.append({
"role":"tool","name":name,"content":observation})
Note: OpenAI API requires tool_call_id for role=tool when using native function calling, but here we're using a text protocol, so role "tool" with content is fine for our own prompt format — but if using OpenAICompatLLM, sending role "tool" without tool_call_id may error. Safer: use role "user" with prefix "OBSERVATION:" ... Hmm. Actually many compatible APIs will reject role "tool" without tool_call_id. To be safe, use role "user" content "[OBSERVATION] ..." — that's a common ReAct-style trick and works everywhere. I'll do that and note it in 注意事项.
So MockLLM checks for messages with role "user" and content starting with "[OBSERVATION]".
Let me restructure MockLLM:
def chat(self, messages):
obs = [m["content"] for m in messages if m["role"] == "user" and m["content"].startswith("[OBSERVATION]")]
if not obs:
return json.dumps({
...tool_call...})
return json.dumps({
"type":"final","answer": f"..."})
Good.
Agent loop handling of format errors: append assistant raw + user message "[OBSERVATION] 上一步输出无法解析：... 请只输出 JSON。" and continue (counts as step).
Also need: first turn messages list. Final answer extraction.
Let me write code fully:
"""mini_agent.py —— 一个只依赖标准库的最小智能体循环"""
from __future__ import annotations
import ast
import json
import operator
import re
import time
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol
# ---------- 1. 工具层 ----------
class ToolError(Exception):
pass
@dataclass
class Tool:
name: str
description: str
parameters: dict[str, Any]
func: Callable[..., Any]
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 schema_text(self) -> str:
items = [
{
"name": t.name, "description": t.description, "parameters": t.parameters}
for t in self._tools.values()
]
return json.dumps(items, ensure_ascii=False, indent=2)
def call(self, name: str, args: dict[str, Any]) -> str:
tool = self._tools.get(name)
if tool is None:
return f"ERROR: 未注册的工具 {name}"
if not isinstance(args, dict):
return "ERROR: arguments 必须是 JSON 对象"
try:
result = tool.func(**args)
text = str(result)
return text[:2000] # 观察结果截断，避免上下文膨胀
except TypeError as e:
return f"ERROR: 参数不匹配 {e}"
except Exception as e:
return f"ERROR: {type(e).__name__}: {e}"
safe_eval:
_BIN_OPS = {
ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
ast.Div: operator.truediv, ast.Mod: operator.mod, ast.Pow: operator.pow,
ast.FloorDiv: operator.floordiv}
_UNARY_OPS = {
ast.USub: operator.neg, ast.UAdd: operator.pos}
def safe_eval(expr: str) -> float:
"""只允许数字与四则运算，不使用 eval。"""
def _walk(node: ast.AST) -> float:
if isinstance(node, ast.Expression):
return _walk(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 _BIN_OPS:
return _BIN_OPS[type(node.op)](_walk(node.left), _walk(node.right))
if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
return _UNARY_OPS[type(node.op)](_walk(node.operand))
raise ValueError(f"表达式含不允许的语法: {ast.dump(node)[:60]}")
return float(_walk(ast.parse(expr, mode="eval")))
Then tool defs:
def build_registry() -> ToolRegistry:
reg = ToolRegistry()
reg.register(Tool(
name="calculator",
description="计算只含数字和 + - * / % ** 的算术表达式，返回数值字符串",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string", "description": "如 (128*3+40)/4"}},
"required": ["expression"]},
func=lambda expression: safe_eval(expression),
))
reg.register(Tool(
name="now",
description="返回当前时间的 ISO8601 字符串，无参数",
parameters={
"type": "object", "properties": {
}, "required": []},
func=lambda: time.strftime("%Y-%m-%dT%H:%M:%S"),
))
return reg
Wait: time.strftime with lambda no args - fine.
LLM layer:
class LLM(Protocol):
def chat(self, messages: list[dict[str, str]]) -> str: ...
class OpenAICompatLLM:
"""任意 OpenAI 兼容的 /chat/completions 端点，例如本地 vLLM、Ollama 的兼容口。"""
def __init__(self, base_url: str, api_key: str, model: str, timeout: float = 60.0):
self.url = base_url.rstrip("/") + "/chat/completions"
self.api_key = api_key
self.model = model
self.timeout = timeout
def chat(self, messages):
payload = {
"model": self.model, "messages": messages, "temperature": 0.0}
req = urllib.request.Request(
self.url,
data=json.dumps(payload).encode("utf-8"),
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"},
method="POST",
)
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
data = json.loads(resp.read().decode("utf-8"))
return data["choices"][0]["message"]["content"] or ""
MockLLM as described.
Parser:
class ActionFormatError(Exception):
pass
_FENCE = re.compile(r"^```[a-zA-Z]*\s*|\s*```$")
def parse_action(raw: str) -> dict[str, Any]:
text = _FENCE.sub("", raw.strip())
try:
data = json.loads(text)
except json.JSONDecodeError as exc:
raise ActionFormatError(f"输出不是合法 JSON（{exc}）") from exc
if not isinstance(data, dict):
raise ActionFormatError("输出必须是 JSON 对象")
kind = data.get("type")
if kind == "final":
if not isinstance(data.get("answer"), str):
raise ActionFormatError("final 缺少字符串字段 answer")
elif kind == "tool_call":
if not isinstance(data.get("name"), str) or not isinstance(data.get("arguments", {
}), dict):
raise ActionFormatError("tool_call 需要字符串 name 与对象 arguments")
else:
raise ActionFormatError("type 只能是 tool_call 或 final")
return data
Agent:
```python
SYSTEM_PROMPT = """你是一个会使用工具的智能体。每一轮只能输出一个 JSON 对象，不要输出 JSON 之外的任何文字。
可选工具（JSON Schema）：
{tools}
输出格式二选一：
1) 调用工具：{
{"type":"tool_call","name":"<工具名>","arguments":{
{...}},"thought":"<一句话>"
2) 给出答案：{
{"type":"final","answer":"<最终答复>"}}
规则：需要外部计算或事实时必须调用工具；收到以 [OBSERVATION] 开头的消息即为工具结果，据此决定下一步。"""
@dataclass
class Step:
index: int
raw: str
action: dict[str, Any] | None
observation: str | None
elapsed_ms: int
@dataclass
class AgentResult:
answer: str
steps: int
trace: list[Step] = field(default
