源码级别解析 · ~1000行核心代码 · CodeAgent · 多智能体协作
2026-05-07 | 每日技术深度解读
smol = small, 精简到极致的智能体实现
代码作为行动表达,比文本指令更精确
CodeAgent的核心循环: 生成代码→执行→观察→重复
选择取决于安全需求和任务复杂度
| 特性 | CodeAgent | ToolCallingAgent |
|---|---|---|
| 行动表达 | Python代码 | 结构化JSON |
| 灵活性 | 极高(任意Python) | 受限(预定义工具) |
| 安全性 | 需沙箱 | 天然安全 |
| 多步推理 | 代码循环/变量 | 多次工具调用 |
| 适用场景 | 数据分析、编程 | 信息检索、API调用 |
| 代码行数 | ~500行 | ~300行 |
from smolagents import CodeAgent, WebSearchTool, InferenceClientModel
model = InferenceClientModel()
agent = CodeAgent(
tools=[WebSearchTool()],
model=model,
stream_outputs=True
)
agent.run("How many seconds for a leopard at full speed to run through Pont des Arts?")
smolagents的设计目标: 几行代码即可运行强大智能体
class Agent(ABC):
"""所有智能体的基类"""
model: Model
tools: list[BaseTool]
memory: AgentMemory
max_steps: int = 20
@abstractmethod
def run(self, task: str) -> str: ...
class ToolCallingAgent(Agent):
"""传统工具调用智能体"""
def step(self) -> ActionOutput:
# LLM生成ChatMessageToolCall
# 解析并执行工具调用
...
class CodeAgent(Agent):
"""代码执行智能体 (核心创新)"""
python_executor: PythonExecutor
def step(self) -> ActionOutput:
# LLM生成Python代码
# 通过python_executor执行
...
Agent基类 → ToolCallingAgent / CodeAgent 两种实现
class CodeAgent(Agent):
def step(self) -> ActionOutput:
# 1. 构建消息: system_prompt + memory + tools
messages = self.memory.get_serialized_messages()
# 2. LLM生成响应 (包含Python代码)
chat_message = self.model(messages)
# 3. 从响应中提取代码块
code_action = extract_code_from_text(chat_message.content)
# 4. 在沙箱中执行代码
observation, is_final = self.python_executor(
code_action,
tools=self.tools,
authorized_imports=self.authorized_imports
)
# 5. 记录到memory
self.memory.step(ActionStep(
tool_calls=[],
observations=[observation]
))
return ActionOutput(observation, is_final)
CodeAgent每一步: 生成代码→提取→执行→记录
def extract_code_from_text(text: str) -> str:
"""从LLM输出中提取Python代码块"""
# 匹配 ```python ... ``` 代码块
code_blocks = []
# 尝试解析markdown代码块
pattern = r"```(?:python)?\s*\n(.*?)```"
matches = re.findall(pattern, text, re.DOTALL)
if matches:
return matches[-1] # 取最后一个代码块
# 回退: 如果没有代码块标记,
# 尝试解析整个文本为代码
return parse_code_blobs(text)
def fix_final_answer_code(code: str) -> str:
"""确保代码调用final_answer工具"""
if 'final_answer' not in code:
code += f"\nfinal_answer({code_result})"
return code
从LLM输出中智能提取可执行代码
6种沙箱执行后端,从本地到云端全覆盖
生产环境推荐Docker或E2B沙箱
class LocalPythonExecutor(PythonExecutor):
"""本地Python代码执行器"""
BASE_BUILTIN_MODULES = [
'math', 'random', 'datetime', 're',
'collections', 'itertools', 'statistics'
]
def __init__(self, authorized_imports, tools):
self.authorized_imports = authorized_imports
self.tools = {t.name: t for t in tools}
self.state = {} # 跨步骤共享变量
def __call__(self, code, tools, authorized_imports):
# 1. 注入工具为函数
for tool_name, tool in self.tools.items():
self.state[tool_name] = tool
# 2. 控制import: 只允许授权模块
exec(code, {
'__builtins__': {
**{m: __import__(m) for m in authorized_imports},
'__import__': restricted_import,
}
}, self.state)
return self._get_observation()
通过受限的exec()实现安全本地执行
class DockerExecutor:
"""Docker容器沙箱执行器"""
def __init__(self, image="python:3.11-slim"):
self.image = image
self.container = None
def execute(self, code, tools, authorized_imports):
if not self.container:
self.container = docker_client.containers.run(
self.image, detach=True,
mem_limit="512m", cpu_period=100000,
network_disabled=True # 安全: 禁用网络
)
# 序列化工具并发送到容器
exec_result = self.container.exec_run(
cmd=["python", "-c", code],
workdir="/workspace"
)
return exec_result.output.decode()
def close(self):
if self.container:
self.container.stop()
self.container.remove()
Docker提供进程和网络隔离,适合生产环境
工具是智能体能力的扩展点
class Tool(BaseTool):
"""智能体工具基类"""
name: str # 工具名称
description: str # 工具描述 (LLM看到)
inputs: dict # 输入JSON Schema
output_type: str # 输出类型
output_schema: dict # 输出JSON Schema
is_initialized: bool = False
@abstractmethod
def forward(self, *args, **kwargs) -> Any:
"""工具核心逻辑,子类实现"""
...
def __call__(self, *args, **kwargs) -> Any:
"""延迟初始化 + 执行"""
if not self.is_initialized:
self.setup() # 首次调用时初始化
self.is_initialized = True
return self.forward(*args, **kwargs)
def setup(self):
"""可选: 加载模型等昂贵操作"""
pass
def to_dict(self) -> dict:
"""序列化为字典 (Hub分享用)"""
return {
'name': self.name,
'description': self.description,
'inputs': self.inputs,
'output_type': self.output_type
}
延迟初始化模式: setup()只在首次调用时执行
from smolagents import Tool
class WeatherTool(Tool):
name = "get_weather"
description = "获取指定城市的天气信息"
inputs = {
'city': {
'type': 'string',
'description': '城市名称'
}
}
output_type = "string"
def forward(self, city: str) -> str:
# 调用天气API
response = requests.get(
f"https://api.weather.com/v1/{city}"
)
data = response.json()
return f"{city}: {data['temp']}°C, {data['condition']}"
# CodeAgent中,工具直接作为函数调用:
# result = get_weather("Beijing")
CodeAgent中工具=函数,直接在代码中调用
tool-agnostic设计理念
# 从MCP服务器导入工具
from smolagents import ToolCollection
tools = ToolCollection.from_mcp(
server_url="http://localhost:3000"
)
# 从LangChain导入工具
from smolagents import Tool
tool = Tool.from_langchain(
langchain_tool # 任何LangChain BaseTool
)
# 从HuggingFace Hub加载工具
tool = Tool.from_hub("user/tool_name")
# 将Hub Space作为工具
tool = Tool.from_space(
"user/space_name",
"这个工具可以做XXX"
)
# 发布工具到Hub
my_tool.push_to_hub("user/my_weather_tool")
一个框架打通多个工具生态
| 工具名 | 功能 | 输入 | 输出 |
|---|---|---|---|
| WebSearchTool | DuckDuckGo网络搜索 | 查询字符串 | 搜索结果 |
| VisitWebTool | 访问网页获取内容 | URL | 页面文本 |
| FileManagementTool | 文件读写管理 | 文件路径+操作 | 文件内容 |
| FinalAnswerTool | 终止智能体循环 | 最终答案 | 终止信号 |
| PythonTool | 执行Python代码 | 代码字符串 | 执行结果 |
Memory是ReAct循环的核心组件
class MemoryStep(TypedDict):
"""所有记忆步骤的基类"""
timestamp: float
class TaskStep(MemoryStep):
"""用户任务"""
task: str
class ActionStep(MemoryStep):
"""智能体行动"""
tool_calls: list[ToolCall]
observations: list[str]
class PlanningStep(MemoryStep):
"""规划步骤"""
plan: str
class FinalAnswerStep(MemoryStep):
"""最终答案"""
answer: str
class AgentMemory:
def __init__(self):
self.steps: list[MemoryStep] = []
self.system_prompt: str = ""
def get_serialized_messages(self) -> list[dict]:
"""将记忆序列化为LLM消息格式"""
...
类型化的记忆步骤,便于回溯和分析
MODEL_REGISTRY注册机制,插件式扩展
class Model(ABC):
"""模型抽象基类"""
@abstractmethod
def chat(self, messages, **kwargs) -> ChatMessage:
...
nclass InferenceClientModel(Model):
"""HuggingFace推理端点 (免费!)"""
def __init__(self, model_id=None, provider=None):
self.client = InferenceClient(
model=model_id, provider=provider
)
def chat(self, messages, **kwargs) -> ChatMessage:
response = self.client.chat_completion(
messages=messages, **kwargs
)
return ChatMessage.from_dict(response)
class LiteLLMModel(Model):
"""100+ LLM提供商"""
def chat(self, messages, **kwargs):
import litellm
response = litellm.completion(
model=self.model_id,
messages=messages,
**kwargs
)
return ChatMessage.from_openai(response)
LiteLLM一个库接入几乎所有LLM
| 模型类 | 提供商 | 需要API Key | 特色 |
|---|---|---|---|
| InferenceClientModel | HF Inference | 可选(有免费) | 免费推理、HF生态 |
| LiteLLMModel | 100+提供商 | 是 | 最广泛的覆盖 |
| OpenAIModel | OpenAI/兼容 | 是 | GPT系列 |
| TransformersModel | 本地模型 | 否 | 离线、隐私 |
| AzureOpenAIModel | Azure | 是 | 企业合规 |
| AmazonBedrockModel | AWS | 是 | AWS生态集成 |
多智能体协作的核心机制
from smolagents import CodeAgent, ManagedAgent
# 子智能体1: 搜索专家
web_agent = CodeAgent(
tools=[WebSearchTool()],
model=model,
name="search_agent",
description="搜索网络获取信息"
)
# 子智能体2: 数据分析专家
data_agent = CodeAgent(
tools=[PythonTool()],
model=model,
name="data_agent",
description="分析数据和生成图表"
)
# 主智能体: 协调者
manager = CodeAgent(
tools=[
ManagedAgent(web_agent), # 作为工具!
ManagedAgent(data_agent), # 作为工具!
],
model=model,
)
# 主智能体在代码中调用子智能体:
# search_result = search_agent("AI最新进展")
# analysis = data_agent(search_result)
ManagedAgent将Agent→Tool,实现递归委托
Manager通过代码调用委托任务给子智能体
Plan-and-Execute模式提升可靠性
class PlanningPromptTemplate(TypedDict):
initial_plan: str
update_plan_pre_messages: str
update_plan_post_messages: str
DEFAULT_PLANNING_PROMPT = PlanningPromptTemplate(
initial_plan="""You should first make a plan and
clarify your plan. Then carry out the plan step
by step. You can revise your plan based on new
information you discover along the way.""",
update_plan_pre_messages="""Here is the plan you
have been following so far:
{plan}
Recent steps:
{memory}""",
update_plan_post_messages="""Now, based on the
above, please update your plan. Keep it simple:
just state the remaining steps."""
)
Jinja2模板驱动的规划系统
所有提示词都可通过模板定制
from jinja2 import Template, StrictUndefined
def populate_template(template: str, variables: dict) -> str:
"""用变量填充Jinja2模板"""
compiled = Template(template, undefined=StrictUndefined)
try:
return compiled.render(**variables)
except Exception as e:
raise Exception(
f"Template rendering error: {type(e).__name__}: {e}"
)
# 示例: 系统提示词模板
SYSTEM_PROMPT = """
You are a helpful assistant.
You have access to these tools: {{ tool_names }}.
To use a tool, write Python code calling the tool as a function.
Always end with final_answer(your_result).
"""
rendered = populate_template(SYSTEM_PROMPT, {
'tool_names': 'web_search, visit_webpage'
})
StrictUndefined防止未定义变量导致的静默错误
一键分享,社区驱动增长
# 1. 发布智能体到Hub
agent.push_to_hub("my-user/my-coder-agent")
# → 自动创建repo,上传代码、工具定义、配置
# 2. 从Hub加载智能体
from smolagents import CodeAgent
agent = CodeAgent.from_hub("my-user/my-coder-agent")
agent.run("帮我写一个快速排序")
# 3. 发布自定义工具
my_tool.push_to_hub("my-user/weather-tool")
# 4. 从Hub加载工具
from smolagents import Tool
tool = Tool.from_hub("my-user/weather-tool")
# 5. push_to_hub内部实现
# → create_repo() + upload_folder()
# → 包含: agent_config.yaml, tools.py, README.md
# → 自动创建Gradio Space demo
基于huggingface_hub的完整分享机制
无需写代码,命令行直接用
# 通用智能体
smolagent "Plan a trip to Tokyo" \
--model-type "InferenceClientModel" \
--model-id "Qwen/Qwen3-Next-80B-A3B-Thinking" \
--imports pandas numpy \
--tools web_search
# 交互模式 (无参数启动)
smolagent
# → 引导选择: Agent类型 → 工具 → 模型 → 输入任务
# 网页浏览智能体
webagent "go to amazon.com, find best-selling books" \
--model-type "LiteLLMModel" \
--model-id "gpt-4o"
# 使用本地模型 (完全离线)
smolagent "分析这个CSV文件" \
--model-type "TransformersModel" \
--model-id "Qwen/Qwen3-Next-80B-A3B-Thinking" \
--imports pandas
CLI让智能体像命令行工具一样简单
模态无关设计,支持任意输入输出类型
# agent_types.py
class AgentImage:
"""智能体图像类型"""
def __init__(self, path_or_url: str):
self.path = path_or_url
def to_pil(self) -> "PIL.Image.Image":
return Image.open(self.path)
class AgentAudio:
"""智能体音频类型"""
def __init__(self, path: str):
self.path = path
def handle_agent_input_types(**kwargs):
"""自动将输入转换为模型可处理格式"""
for key, value in kwargs.items():
if isinstance(value, AgentImage):
kwargs[key] = value.to_pil()
elif isinstance(value, AgentAudio):
kwargs[key] = load_audio(value.path)
return kwargs
ndef handle_agent_output_types(output):
"""自动包装模型输出"""
if isinstance(output, PIL.Image.Image):
return AgentImage.from_pil(output)
return output
类型系统确保多模态输入输出的正确处理
完整的可观测性支持
@dataclass
class RunResult:
"""智能体运行的完整结果"""
output: Any | None # 最终输出
state: Literal["success", "max_steps_error"] # 运行状态
steps: list[dict] # 所有记忆步骤
token_usage: TokenUsage | None # Token消耗
timing: Timing # 时间统计
nclass TokenUsage(TypedDict):
input_tokens: int
output_tokens: int
total_tokens: int
class Timing(TypedDict):
start_time: float
end_time: float
duration: float
n# 使用示例
result = agent.run("分析数据")
print(f"状态: {result.state}")
print(f"Token: {result.token_usage['total_tokens']}")
print(f"耗时: {result.timing['duration']:.2f}s")
print(f"步骤数: {len(result.steps)}")
RunResult提供运行全貌,便于调试和优化
精确的错误分类便于定位和恢复
# 启用流式输出
agent = CodeAgent(
model=model,
tools=[WebSearchTool()],
stream_outputs=True # 关键!
)
# 源码: stream_outputs处理
if self.stream_outputs:
with Live(
Group(
Panel(Markdown(accumulated_output)),
Rule(),
),
console=console,
refresh_per_second=4,
) as live:
for delta in model.stream_chat(messages):
# 实时更新Rich Live显示
accumulated_output += delta.content
live.update(
Panel(Markdown(accumulated_output))
)
# 使用Rich库实现漂亮的终端实时输出
Rich Live提供流畅的终端流式体验
极简模块划分,每个文件职责明确
| 模块 | 行数 | 依赖 | 核心类 |
|---|---|---|---|
| agents.py | ~1000 | memory,tools,models | Agent,CodeAgent,ToolCallingAgent |
| tools.py | ~500 | huggingface_hub | Tool,ToolCollection |
| models.py | ~300 | litellm,openai | Model,LiteLLMModel |
| memory.py | ~200 | 无 | AgentMemory,MemoryStep |
| local_python_executor.py | ~300 | 无 | LocalPythonExecutor |
| remote_executors.py | ~400 | docker,e2b,modal | DockerExecutor,E2BExecutor |
| monitoring.py | ~200 | rich | AgentLogger,Monitor |
与LangChain等重量级框架形成鲜明对比
| 特性 | smolagents | LangChain | CrewAI | Agno |
|---|---|---|---|---|
| 核心代码 | ~1000行 | ~50000行 | ~15000行 | ~5000行 |
| 代码智能体 | ✅ 一等公民 | ❌ | ❌ | ❌ |
| 沙箱执行 | 6种后端 | 无内置 | 无内置 | 无内置 |
| Hub集成 | ✅ 原生 | ❌ | ❌ | ❌ |
| CLI工具 | ✅ 2个 | ❌ | ❌ | ❌ |
| 学习曲线 | 极低 | 高 | 中 | 低 |
| 多智能体 | ManagedAgent | LangGraph | ✅ 核心 | ✅ 核心 |
最适合需要灵活代码执行的场景
极简设计意味着某些高级功能需要自行实现
from smolagents import CodeAgent, InferenceClientModel
data_analyst = CodeAgent(
model=InferenceClientModel(
model_id="deepseek-ai/DeepSeek-R1",
provider="together"
),
tools=[], # 纯代码智能体,不需要外部工具
authorized_imports=["pandas", "matplotlib", "numpy", "scipy"],
max_steps=10,
)
result = data_analyst.run(
"""加载 iris.csv 数据集:
1. 计算每种花的平均花瓣长度
2. 找出花瓣最长的种类
3. 生成散点图保存为 scatter.png
"""
)
# CodeAgent会生成类似这样的代码:
# import pandas as pd
# df = pd.read_csv('iris.csv')
# means = df.groupby('species')['petal_length'].mean()
# max_species = means.idxmax()
# import matplotlib.pyplot as plt
# df.plot.scatter(x='petal_length', y='petal_width')
# plt.savefig('scatter.png')
# final_answer(f"最长花瓣: {max_species}, 均值: {means[max_species]:.2f}")
纯代码智能体: 不需要工具,直接写Python
感谢阅读!
访问 https://atcfu.com/ai-articles/smolagents-code-agents/ 回顾本文