2026 年 Python 开源生态中的热门项目，涵盖 AI 代理框架如 LangGraph、CrewAI 和 smolagents，以及数据工程工具包括高性能 DataFrame 库 Polars、嵌入式分析数据库 DuckDB 和数据编排平台 Dagster。此外还提及了统一 LLM API 网关 LiteLLM 和高性能 API 框架 FastAPI。文章通过架构图示和代码示例展示了这些工具的核心功能与应用场景，为开发者提供了选型参考和技术栈建议，强调 Rust 加速 Python 及 AI 代理成为标配的趋势。
2026 年的 Python 生态正在被 AI 代理（AI Agent）和数据工程工具重新定义。本文精选 GitHub 上最具影响力的开源项目，涵盖 AI 代理框架、数据管道工具、向量数据库客户端等关键领域，附带代码示例与架构解析。
┌─────────────────────────────────────────────────────────────────────┐
│ 2026 Python 开源热门方向 │
├──────────────────┬──────────────────┬───────────────────────────────┤
│ AI 代理框架 │ 数据工具链 │ 基础设施与编排 │
├──────────────────┼──────────────────┼───────────────────────────────┤
│ LangGraph │ Polars │ Dagster │
│ CrewAI │ DuckDB │ Prefect │
│ AutoGen │ ibis-project │ Modal │
│ PydanticAI │ Airflow 3.0 │ BentoML │
│ OpenAI Agents SDK│ LanceDB │ FastAPI │
│ smolagents │ Delta Lake │ LiteLLM │
└──────────────────┴──────────────────┴───────────────────────────────┘
GitHub: langchain-ai/langgraph | ⭐ 55k+
LangGraph 将 AI 代理建模为有向图（Directed Graph），支持循环、分支、人工介入等复杂控制流，是目前最成熟的代理编排框架。
┌──────────────── LangGraph 核心架构 ────────────────┐
│
│
┌─────────┐ ┌──────────┐ ┌───────────┐
│ 用户输入 │───▶│ 路由节点 │───▶│ Agent 节点│
└─────────┘ └────┬─────┘ └─────┬─────┘
│
│
┌────────┼────────┐
│
▼ ▼ ▼ ▼
│
┌────────┐┌────────┐┌────────┐┌────────┐
│ 搜索工具 ││代码执行 ││数据库 ││LLM 推理 │
└────────┘└────────┘└────────┘└────────┘
│
│
│
└────────┴────────┴────────┘
│
│
▼
│
┌─────────────┐
│ 条件分支 │◀─── 循环回路上一步 │
│ 继续或结束 │
└──────┬──────┘
│
▼
│
┌─────────────┐
│ 最终输出 │
└─────────────┘
└─────────────────────────────────────────────────────┘
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator
# 定义状态
class ResearchState(TypedDict):
messages: Annotated[list, operator.add]
research_topic: str
findings: list[str]
iteration: int
# 定义工具
def search_web(query: str) -> str:
"""模拟网络搜索"""
return f"搜索结果：关于 '{query}' 的最新研究发现..."
def analyze_paper(paper_url: str) -> str:
"""分析论文内容"""
return f"论文分析：{paper_url} 的核心结论是..."
# 构建图
def create_research_agent():
llm = ChatOpenAI(model="gpt-4o")
# 节点 1: 规划研究步骤
def plan_research(state: ResearchState) -> dict:
prompt = f"为以下主题制定研究计划：{state['research_topic']}"
response = llm.invoke(prompt)
return {: [response]}
() -> :
topic = state[]
results = search_web(topic)
{: [results], : state.get(, ) + }
() -> :
all_findings = .join(state[])
prompt =
response = llm.invoke(prompt)
{: [response]}
() -> :
state.get(, ) >= :
graph = StateGraph(ResearchState)
graph.add_node(, plan_research)
graph.add_node(, execute_search)
graph.add_node(, synthesize)
graph.set_entry_point()
graph.add_edge(, )
graph.add_conditional_edges(, should_continue)
graph.add_edge(, END)
graph.()
agent = create_research_agent()
result = agent.invoke({: [], : , : [], : })
(result[][-].content)
GitHub: crewAIInc/crewAI | ⭐ 30k+
CrewAI 的核心理念是让多个 AI 代理像团队一样协作，每个代理有明确的角色、目标和工具。
┌──────────────── CrewAI 多代理协作模型 ────────────────┐
│
│
┌──────────┐
│ 任务输入 │
└─────┬────┘
│
▼
│
┌───────────┐ ┌───────────┐ ┌───────────────┐
│ 研究员代理 │──▶│ 编写者代理 │──▶│ 审核者代理 │
│ Role: 研究 │ │ Role: 撰写 │ │ Role: 质量控制 │
│ Tools: 搜索 │ │ Tools: 无 │ │ Tools: 评估 │
└───────────┘ └───────────┘ └───────┬───────┘
│
│
│
┌────────────┴────────┐
│
│
▼ ▼
│
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
# 定义工具
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
# 定义代理
researcher = Agent(
role="高级技术研究员",
goal="深入研究给定主题，收集最新、最权威的信息",
backstory="""你是一位拥有 10 年经验的技术研究员，擅长从海量信息中
提取关键洞察，对 AI 和数据领域有深刻理解。""",
tools=[search_tool, scrape_tool],
verbose=True,
llm="gpt-4o"
)
writer = Agent(
role="技术内容撰写专家",
goal="将研究结论转化为清晰、有深度的技术文章",
backstory="""你是一位资深技术作家，曾为多家顶级科技媒体撰稿。
你擅长用通俗易懂的语言解释复杂的技术概念。""",
verbose=True,
llm="gpt-4o"
)
reviewer = Agent(
role="内容质量审核员",
goal="确保文章的技术准确性、逻辑连贯性和可读性",
backstory="""你是一位严格的技术编辑，对事实准确性和逻辑严谨性
有极高的标准。你会仔细核查每一个技术细节。""",
verbose=True,
llm="gpt-4o"
)
# 定义任务
research_task = Task(
description="""
研究 {topic} 的最新进展，包括：
1. 核心技术原理和架构
2. 主要开源项目和工具
3. 业界最佳实践和案例
4. 未来发展趋势
""",
expected_output="一份包含 5 个以上关键发现的研究报告",
agent=researcher
)
writing_task = Task(
description="""
基于研究报告，撰写一篇技术博客文章，要求：
1. 标题吸引人，开头有冲击力
2. 包含代码示例和架构图
3. 对比分析不同方案的优劣
4. 给出明确的实践建议
""",
expected_output="一篇 2000 字以上的 Markdown 格式技术文章",
agent=writer
)
review_task = Task(
description="""
审核文章的：
1. 技术准确性 — 所有技术概念是否正确
2. 逻辑连贯性 — 文章结构是否合理
3. 代码质量 — 示例代码是否能正常运行
4. 可读性 — 目标读者是否能理解
""",
expected_output="审核通过的文章终稿 + 修改说明",
agent=reviewer
)
# 组建团队并运行
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, writing_task, review_task],
process=Process.sequential # 顺序执行
)
result = crew.kickoff(inputs={"topic": "2026 年 Python AI Agent 开发实践"})
(result)
GitHub: huggingface/smolagents | ⭐ 15k+
smolagents 主打极简主义，整个框架核心仅几千行代码，适合快速原型和嵌入式场景。
from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel
# 3 行代码创建一个能搜索并执行代码的代理
agent = CodeAgent(
tools=[DuckDuckGoSearchTool()],
model=HfApiModel("Qwen/Qwen2.5-72B-Instruct"),
additional_authorized_imports=["pandas", "numpy", "matplotlib"]
)
result = agent.run("搜索 2026 年 GitHub 上星标最多的 Python 项目，然后用 pandas 创建 DataFrame 并按星标数排序")
print(result)
GitHub: pola-rs/polars | ⭐ 32k+
Polars 基于 Rust 编写，采用惰性求值（Lazy Evaluation）和多线程并行，在大多数基准测试中比 pandas 快 5-30 倍。
┌───────────────── Polars vs Pandas 性能对比 ──────────────────┐
│
│
操作：读取 5GB CSV → 过滤 → 分组聚合 → 排序
│
│
Pandas (单线程) ████████████████████████████ 48s
Polars (eager) ████████ 11s
Polars (lazy) ████ 6.2s
DuckDB ███ 4.8s
│
│
0s 10s 20s 30s 40s 50s
└───────────────────────────────────────────────────────────────┘
import polars as pl
# 惰性读取 + 链式操作（自动优化查询计划）
result = (
pl.scan_csv("data/orders_2026.csv") # 惰性读取
.filter(pl.col("amount") > 100) # 谓词下推
.with_columns(
pl.col("created_at").str.to_datetime("%Y-%m-%d %H:%M:%S").dt.month().alias("month"),
(pl.col("amount") * pl.col("tax_rate")).alias("tax"),
pl.col("user_id").hash(seed=42).alias("user_hash") # 差分隐私
)
.group_by(["month", "category"])
.agg(
pl.col("amount").sum().alias("total_amount"),
pl.col("amount").mean().alias("avg_amount"),
pl.col("order_id").n_unique().alias("order_count"),
pl.col("user_id").n_unique().alias("unique_users"),
)
.sort("total_amount", descending=True)
.head(20)
.collect() # 在此触发实际计算
)
print(result)
from langchain_openai import ChatOpenAI
import polars as pl
class DataAnalysisAgent:
def __init__(self, df: pl.DataFrame):
self.df = df
self.llm = ChatOpenAI(model="gpt-4o")
self.schema = df.schema
self.head = df.head(5).to_pandas().to_string()
def analyze(self, question: str) -> pl.DataFrame:
"""将自然语言问题转换为 Polars 查询"""
prompt = f"""
数据框架 schema: {self.schema}
数据预览：{self.head}
用户问题：{question}
请生成 Polars 代码来回答这个问题。
只输出可执行的 Python 代码，不要解释。
"""
code = self.llm.invoke(prompt).content
# 清理代码块标记
code = code.replace("```python", "").replace("```", "").strip()
# 安全执行
local_vars = {"df": self.df, "pl": pl}
exec(code, {"__builtins__": {}}, local_vars)
return local_vars.get("result", pl.DataFrame())
# 使用
df = pl.read_csv("data/sales_2026.csv")
agent = DataAnalysisAgent(df)
result = agent.analyze("每月销售额最高的三个产品类别是什么？")
print(result)
GitHub: duckdb/duckdb | ⭐ 28k+
DuckDB 被称为"分析领域的 SQLite"，支持直接查询 Parquet、CSV、JSON 等文件，无需导入数据。
import duckdb
# 直接查询 Parquet 文件（无需加载到内存）
result = duckdb.sql("""
WITH monthly_stats AS (
SELECT DATE_TRUNC('month', created_at) AS month,
category,
SUM(amount) AS total_sales,
COUNT(*) AS order_count,
AVG(amount) AS avg_order_value
FROM read_parquet('s3://data-lake/orders/*.parquet')
WHERE year(created_at) = 2026 AND status = 'completed'
GROUP BY ALL
)
SELECT category, month, total_sales, order_count,
-- 环比增长率
(total_sales - LAG(total_sales) OVER (
PARTITION BY category ORDER BY month
)) / LAG(total_sales) OVER (
PARTITION BY category ORDER BY month
) AS mom_growth
FROM monthly_stats
ORDER BY total_sales DESC
LIMIT 20
""")
# 结果直接转 Polars DataFrame
df = result.pl()
print(df)
# 或者导出为 Parquet
result.write_parquet("output/monthly_sales.parquet")
GitHub: dagster-io/dagster | ⭐ 14k+
Dagster 3.0 将数据管道定义为软件定义资产（Software-Defined Assets），天然支持增量计算和血缘追踪。
┌──────────────── Dagster 数据管道血缘图 ────────────────┐
│
│
┌────────────┐ ┌──────────────┐ ┌────────────┐
│ raw_events │────▶│ cleaned_data │───▶│ user_table│
└────────────┘ └──────┬───────┘ └─────┬──────┘
│
│
│
▼ ▼
│
┌──────────────┐ ┌────────────┐
│ feature_store│ │ order_table│
└──────┬───────┘ └─────┬──────┘
│
│
│
└────────┬───────────┘
│
▼
│
┌──────────────┐
│ ml_training │
└──────┬───────┘
│
▼
│
┌──────────────┐
│ model_registry│
└──────────────┘
└─────────────────────────────────────────────────────────┘
from dagster import (
asset, AssetExecutionContext, MaterializeResult, MetadataValue, Config, Definitions
)
import polars as pl
import duckdb
class DataConfig(Config):
date_range_start: str = "2026-01-01"
date_range_end: str = "2026-03-30"
@asset(
description="原始用户行为日志",
compute_kind="polars",
group_name="ingestion"
)
def raw_events(context: AssetExecutionContext) -> pl.DataFrame:
"""从数据湖读取原始事件数据"""
df = pl.scan_parquet("data/events/*.parquet").collect()
context.log.info(f"读取 {len(df)} 条原始事件")
return df
@asset(
description="清洗后的用户特征数据",
compute_kind="polars",
group_name="processing"
)
def cleaned_data(context: AssetExecutionContext, raw_events: pl.DataFrame) -> pl.DataFrame:
"""数据清洗与特征工程"""
cleaned = (
raw_events
.filter(pl.col("event_type").is_not_null())
.with_columns(
pl.col("timestamp").str.to_datetime().alias("event_time"),
pl.col("user_id").cast(pl.Int64),
)
.with_columns(
pl.col("event_time").dt.hour().alias("hour"),
pl.col("event_time").dt.day_of_week().alias(),
)
.drop_nulls(subset=[, ])
)
context.log.info()
cleaned
() -> MaterializeResult:
result = duckdb.sql().pl()
result.write_parquet()
MaterializeResult(
metadata={: (result), : MetadataValue.md(result.head().to_pandas().to_markdown())}
)
defs = Definitions(assets=[raw_events, cleaned_data, feature_store])
GitHub: BerriAI/litellm | ⭐ 20k+
一个 API 调用所有大模型，支持 100+ 提供商的统一接口。
from litellm import completion
import os
# 统一接口，切换模型只需改一行
models_to_try = ["openai/gpt-4o", "anthropic/claude-sonnet-4-6", "google/gemini-2.5-pro", "deepseek/deepseek-chat"]
for model in models_to_try:
response = completion(
model=model,
messages=[{"role": "user", "content": "用一句话解释量子计算"}],
temperature=0.3,
)
print(f"[{model}] {response.choices[0].message.content}\n")
GitHub: fastapi/fastapi | ⭐ 85k+
2026 年 FastAPI 已成为 AI 服务部署的事实标准。
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import AsyncGenerator
import asyncio
app = FastAPI(title="AI Agent Service", version="2.0")
class ChatRequest(BaseModel):
message: str
model: str = "gpt-4o"
stream: bool = False
class ChatResponse(BaseModel):
reply: str
model: str
tokens_used: int
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
"""同步聊天接口"""
from litellm import completion
response = completion(
model=request.model,
messages=[{"role": "user", "content": request.message}],
)
return ChatResponse(
reply=response.choices[0].message.content,
model=request.model,
tokens_used=response.usage.total_tokens
)
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest) -> AsyncGenerator[str, None]:
"""SSE 流式响应"""
litellm completion
response = completion(
model=request.model,
messages=[{: , : request.message}],
stream=,
)
chunk response:
content = chunk.choices[].delta.content
content:
┌────────────────────────────────────────────────────────────────────┐
│ 选型决策树 │
├────────────────────────────────────────────────────────────────────┤
│
│
Q1: 你需要什么？ │
│
│
├── AI 代理开发 ────────────────────────────────────────────── │
│ ├── 需要复杂状态/循环？ ────▶ LangGraph │
│ ├── 多代理协作？ ──────────▶ CrewAI │
│ ├── 极简/嵌入式？ ────────▶ smolagents │
│ └── OpenAI 生态绑定？ ───▶ OpenAI Agents SDK │
│
│
├── 数据处理 ─────────────────────────────────────────────── │
│ ├── 单机大数据处理？ ─────▶ Polars + DuckDB │
│ ├── SQL 分析为主？ ──────▶ DuckDB │
│ ├── 需要类型安全？ ──────▶ Polars (强类型) │
│ └── 从 pandas 迁移？ ───▶ Polars (API 相似) │
│
│
├── 数据管道编排 ─────────────────────────────────────────── │
│ ├── 现代 asset-centric？ ─▶ Dagster │
│ ├── 传统 DAG 工作流？ ───▶ Airflow 3.0 │
│ └── 云原生/弹性？ ───────▶ Prefect │
│
│
└── AI 服务部署 ─────────────────────────────────────────── │
├── API 服务？ ──────────▶ FastAPI + LiteLLM │
├── 模型服务化？ ───────▶ BentoML │
└── Serverless GPU？ ───▶ Modal │
│
└────────────────────────────────────────────────────────────────────┘
GitHub Stars 增长趋势 (2024-2026)
120k ┤
│ ╭──── FastAPI
100k ┤ ╭───╯
│ ╭───╯
80k ┤ ╭───╯
│ ╭───╯
60k ┤ ╭───╯
│ ╭───╯
40k ┤ ╭╯ ╭── CrewAI
│ ╭──╯ ╭── Polars
20k ┤ ╭╯ ╭─╯ ╭─╯ ╭── DuckDB
│╭╯ ╭╯ ╭─╯ ╭─╯ ╭── Dagster
0k ┼╯──╯───╯─────╯─────╯────╯── LiteLLM
2024.1 2024.7 2025.1 2025.7 2026.1
┌────────────────────────────────────────────────────────────┐
│
│
Layer 4: 应用层 │
┌─────────────────────────────────────────────────────┐
│ │
│ FastAPI + PydanticV2 + LiteLLM │
│ └──────────────────────────┬──────────────────────────┘
│ │
│
Layer 3: 代理编排层 │
┌──────────────────────────┴──────────────────────────┐
│ │
│ LangGraph / CrewAI / smolagents │
│ └──────────────────────────┬──────────────────────────┘
│ │
│
Layer 2: 数据处理层 │
┌──────────────────────────┴──────────────────────────┐
│ │
│ Polars + DuckDB + LanceDB │
│ └──────────────────────────┬──────────────────────────┘
│
一句话总结：2026 年的 Python 不再只是"脚本语言"，它已经成为 AI 和数据工程的核心枢纽。掌握上述工具栈，将让你在这个快速演进的生态中保持竞争力。
本文所有代码基于 Python 3.13 + 最新版库编写，截至 2026 年 3 月。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online