增删改查、统计平均分、排名、及格率、导出文件——数据管理程序的标准套路
一、场景:期末了,成绩单一团乱
假设你是老师(或者家长,要管理孩子的各科成绩):
- 语文85,数学92,英语78——记在纸上
- 想算全班平均分?拿计算器一个个加
- 想知道谁进步了?翻上次的记录
- 领导要一份”及格率统计表”?重新算一遍再誊抄
一件纯机械的事,耗掉一晚上。
今天我们做一个成绩管理器:录入成绩、查看名单、算平均分、找最高最低、排名、算及格率、导出成文件。
增删改查、统计平均分、排名、及格率、导出文件——数据管理程序的标准套路
假设你是老师(或者家长,要管理孩子的各科成绩):
一件纯机械的事,耗掉一晚上。
今天我们做一个成绩管理器:录入成绩、查看名单、算平均分、找最高最低、排名、算及格率、导出成文件。
项目地址: https://github.com/huggingface/smolagents
Stars: 29.2k | Commits: 1,054 | 协议: Apache 2.0
核心理念: “Agents that think in code” — 用代码片段而非 JSON 工具调用来表达 Agent 动作
SmolAgents 是 HuggingFace 推出的轻量级 Agent 框架,核心逻辑仅约 1,000 行代码(agents.py)。其最大特点是 Code Agent 模式——LLM 生成 Python 代码作为动作,而非传统的 JSON 工具调用字典。研究表明这种方式比传统工具调用减少 30% 的步骤数,并在困难基准上达到更高性能。
SmolAgents 采用清晰的抽象基类层次结构:MultiStepAgent 为抽象基类,CodeAgent 和 ToolCallingAgent 为两个具体实现。
# agents.py — 基类定义
class MultiStepAgent(ABC):
"""
Agent class that solves the given task step by step, using the ReAct framework:
While the objective is not reached, the agent will perform a cycle of action
(given by the LLM) and observation (obtained from the environment).
"""
def __init__(
self,
tools: list[Tool],
model: Model,
prompt_templates: PromptTemplates | None = None,
instructions: str | None = None,
max_steps: int = 20,
add_base_tools: bool = False,
verbosity_level: LogLevel = LogLevel.INFO,
managed_agents: list | None = None,
step_callbacks: list[Callable] | dict[Type[MemoryStep], Callable | list[Callable]] | None = None,
planning_interval: int | None = None,
name: str | None = None,
description: str | None = None,
provide_run_summary: bool = False,
final_answer_checks: list[Callable] | None = None,
return_full_result: bool = False,
logger: AgentLogger | None = None,
):
Agent 注册表采用白名单机制,防止反序列化时的任意代码执行:
Stars: 29.4k | Forks: 4.7k | License: MIT
创建时间: 2025-03-11 | Python: 3.10+
定位: 轻量级、强大的多 Agent 工作流框架,Provider-agnostic(支持 OpenAI Responses/Chat Completions API 及 100+ 其他 LLM)
OpenAI Agents SDK 的核心是一个 dataclass 驱动的 Agent 模型。它采用两层继承结构:AgentBase 定义共享字段,Agent 扩展完整功能。
@dataclass
class AgentBase(Generic[TContext]):
"""Base class for `Agent` and `RealtimeAgent`."""
name: str
handoff_description: str | None = None
tools: list[Tool] = field(default_factory=list)
mcp_servers: list[MCPServer] = field(default_factory=list)
mcp_config: MCPConfig = field(default_factory=lambda: MCPConfig())
Agent 类在 AgentBase 之上增加了完整的 Agent 行为配置:
class Agent(AgentBase, Generic[TContext]):
instructions: (
str
| Callable[
[RunContextWrapper[TContext], Agent[TContext]],
MaybeAwaitable[str],
]
| None
) = None
prompt: Prompt | DynamicPromptFunction | None = None
handoffs: list[Agent[Any] | Handoff[TContext, Any]] = field(default_factory=list)
model: str | Model | None = None
model_settings: ModelSettings = field(default_factory=get_default_model_settings)
input_guardrails: list[InputGuardrail[TContext]] = field(default_factory=list)
output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list)
output_type: type[Any] | AgentOutputSchemaBase | None = None
hooks: AgentHooks[TContext] | None = None
tool_use_behavior: (
Literal["run_llm_again", "stop_on_first_tool"] | StopAtTools | ToolsToFinalOutputFunction
) = "run_llm_again"
reset_tool_choice: bool = True
关键设计特点:
源码版本:microsoft/autogen (main branch, 2025)
框架定位:微软研究院出品的多智能体AI应用框架,基于消息传递与事件驱动架构
当前状态:已进入维护模式,由 Microsoft Agent Framework (MAF) 接替
AutoGen 采用三层分层架构,每一层有明确的职责边界,上层构建在下层之上:
autogen-core):消息传递、事件驱动 Agent、本地/分布式运行时autogen-agentchat):面向快速原型的高级 API,支持群聊等多 Agent 模式autogen-ext):LLM 客户端、代码执行等扩展能力从 README 中可以看到架构概览:
The autogen framework uses a layered and extensible design. Layers have clearly
divided responsibilities and build on top of layers below. This design enables
you to use the framework at different levels of abstraction, from high-level
APIs to low-level components.
- Core API implements message passing, event-driven agents, and local and
distributed runtime for flexibility and power.
- AgentChat API implements a simpler but opinionated API for rapid prototyping.
- Extensions API enables first- and third-party extensions continuously
expanding framework capabilities.
这种分层设计使得用户可以从不同抽象层级切入——直接用 Core API 构建底层消息驱动系统,或者用 AgentChat API 快速搭建多 Agent 协作原型。
基于 crewAIInc/crewAI 源码(2025年最新版本),10维度深度剖析
GitHub: https://github.com/crewAIInc/crewAI | Stars: 58,000+ | License: MIT
CrewAI 是一个角色扮演式多Agent协作框架,核心设计理念是将AI Agent组织成”船员”(Crew)来协作完成复杂任务。架构分为两大范式:
源码目录结构(lib/crewai/src/crewai/):
crewai/
├── agent/ # Agent核心实现
├── agents/ # Agent构建器、缓存、执行器
├── crews/ # Crew输出等辅助类
├── flow/ # Flow框架(DSL、运行时、定义)
├── tasks/ # Task子类(条件任务等)
├── tools/ # 工具系统(BaseTool、AgentTools)
├── memory/ # 统一记忆系统
├── knowledge/ # 知识源和知识存储
├── events/ # 事件总线和监听器
├── llms/ # LLM提供者适配层
├── mcp/ # MCP协议集成
├── a2a/ # A2A协议集成
├── security/ # 安全配置和指纹
├── state/ # 检查点和运行时状态
├── crew.py # Crew核心类(2490行)
├── task.py # Task核心类(1566行)
├── process.py # 流程枚举
└── llm.py # LLM封装
所有Agent的根基类,使用 Pydantic BaseModel + ABC 多重继承:
项目: langchain-ai/langgraph
Stars: 41,548 | Forks: 7,018 | License: MIT
定位: 面向长时间运行、有状态 Agent 的底层编排框架
灵感来源: Google Pregel、Apache Beam、NetworkX
LangGraph 的核心是一个 Bulk Synchronous Parallel (BSP) 执行引擎,称为 Pregel。每个执行步骤分为三个阶段:
# pregel/main.py — Pregel 类的文档字符串(摘录)
class Pregel(PregelProtocol[StateT, ContextT, InputT, OutputT]):
"""Pregel manages the runtime behavior for LangGraph applications.
Pregel combines **actors** and **channels** into a single application.
**Actors** read data from channels and write data to channels.
Pregel organizes the execution of the application into multiple steps,
following the **Pregel Algorithm**/**Bulk Synchronous Parallel** model.
Each step consists of three phases:
- **Plan**: Determine which **actors** to execute in this step.
- **Execution**: Execute all selected **actors** in parallel.
- **Update**: Update the channels with the values written by the **actors**.
"""
这种设计意味着:同一轮执行中,节点写入的 Channel 值对其他节点不可见,只有进入下一轮时才生效。这从根本上避免了并发写入的竞争条件,是 LangGraph 保证确定性执行的关键。
Channel 是 LangGraph 的核心抽象,所有节点间通信都通过 Channel 进行。BaseChannel 定义了统一的接口:
基于 langchain-ai/langchain 源码分析,覆盖 langchain-core 与 langchain 两个核心包
LangChain 的核心设计哲学是 “一切皆 Runnable”。所有组件——LLM、Prompt、OutputParser、Tool、Retriever——都继承自同一个泛型基类 Runnable[Input, Output],从而获得统一的调用接口和组合能力。
class Runnable(ABC, Generic[Input, Output]):
"""A unit of work that can be invoked, batched, streamed, transformed and composed."""
name: str | None
"""The name of the `Runnable`. Used for debugging and tracing."""
@abstractmethod
def invoke(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Output:
"""Transform a single input into an output."""
async def ainvoke(self, input: Input, config: RunnableConfig | None = None, **kwargs: Any) -> Output:
return await run_in_executor(config, self.invoke, input, config, **kwargs)
关键设计决策:
| 管道运算符实现声明式链式组合,RunnableSequence、RunnableParallel 是核心组合原语invoke/ainvoke、stream/astream、batch/abatch,异步默认通过线程池桥接同步实现def __or__(self, other) -> RunnableSerializable[Input, Any]:
"""Compose this Runnable with another object to create a RunnableSequence."""
return RunnableSequence(self, coerce_to_runnable(other))
def pipe(self, *others, name=None) -> RunnableSerializable[Input, Other]:
"""Equivalent to RunnableSequence(self, *others) or self | others[0] | ..."""
return RunnableSequence(self, *others, name=name)
LangChain 的 Agent 采用经典的 ReAct(Reasoning + Acting) 模式。核心循环定义在 langchain_core/agents.py 中,通过两个关键数据结构驱动:
try/except,给程序装安全网
上一篇我们给通讯录加上了文件保存功能,程序已经很好用了。但你有没有遇到过这种情况:
请输入选择:5
请输入年龄:abc
Traceback (most recent call last):
File "通讯录.py", line 15, in
年龄 = int("abc")
ValueError: invalid literal for int() with base 10: 'abc'
程序直接崩溃了。红色的报错信息看得人一头雾水。
这就像一个不会游泳的人跳进了深水区——没有救生圈,直接沉底。
用抽屉类比理解文件读写
上一篇我们学会了用列表和字典来组织数据——通讯录、成绩单、购物清单,都可以用程序来管理了。但你有没有发现一个问题?
每次运行程序,你辛辛苦苦输入了10个联系人,结果一关程序,全没了。
下次再打开,又得从头输入。
把代码打包,随时复用
上一篇我们学了循环,程序终于能”重复干活”了。但你有没有遇到这种情况:同样的代码块在程序里出现了三四遍,每次只改几个数字?复制粘贴倒是能用,可一旦要改逻辑,就得改三四处——漏了一处就是bug。
函数(Function)就是解决这个问题的。它的核心思想特别简单:把一段代码打包,取个名字,以后想用就用。
想象你做番茄炒蛋做得特别好,朋友问你怎么做。你不会每次都从头想”先打蛋还是先切番茄”,而是把步骤整理成一份菜谱: