跳到主要内容 2026 GitHub 热门 Python 项目:AI 代理与数据工具精选 | 极客日志
Python
2026 GitHub 热门 Python 项目:AI 代理与数据工具精选 > 2026 年的 Python 生态正在被 AI 代理(AI Agent)和数据工程工具重新定义。精选 GitHub 上最具影响力的开源项目,涵盖 AI 代理框架、数据管道工具、向量数据库客户端等关键领域,附带代码示例与架构解析。 * * 一、2026 Python 开源生态全景图 * * 二、AI 代理框架 2.1 LangGraph — 状态机驱动的代理编排 **GitHub**: la…
氛围 发布于 2026/4/6 更新于 2026/4/13 57K 浏览
2026 年的 Python 生态正在被 AI 代理(AI Agent)和数据工程工具重新定义。本文精选 GitHub 上最具影响力的开源项目,涵盖 AI 代理框架、数据管道工具、向量数据库客户端等关键领域,附带代码示例与架构解析。
一、2026 Python 开源生态全景图 ┌─────────────────────────────────────────────────────────────────────┐
│ 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 │
└──────────────────┴──────────────────┴───────────────────────────────┘
二、AI 代理框架
2.1 LangGraph — 状态机驱动的代理编排 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" )
def plan_research (state: ResearchState ) -> dict :
prompt = f"为以下主题制定研究计划:{state['research_topic' ]} "
response = llm.invoke(prompt)
return {"messages" : [response]}
def execute_search (state: ResearchState ) -> dict :
topic = state["research_topic" ]
results = search_web(topic)
return {"findings" : [results], "iteration" : state.get("iteration" , 0 ) + 1 }
def synthesize (state: ResearchState ) -> dict :
all_findings = "\n" .join(state["findings" ])
prompt = f"基于以下发现进行综合分析:\n{all_findings} "
response = llm.invoke(prompt)
return {"messages" : [response]}
def should_continue (state: ResearchState ) -> str :
if state.get("iteration" , 0 ) >= 3 :
return "synthesize"
return "execute_search"
graph = StateGraph(ResearchState)
graph.add_node("plan" , plan_research)
graph.add_node("execute_search" , execute_search)
graph.add_node("synthesize" , synthesize)
graph.set_entry_point("plan" )
graph.add_edge("plan" , "execute_search" )
graph.add_conditional_edges("execute_search" , should_continue)
graph.add_edge("synthesize" , END)
return graph.compile ()
agent = create_research_agent()
result = agent.invoke({"messages" : [], "research_topic" : "2026 年 AI Agent 在企业中的应用趋势" , "findings" : [], "iteration" : 0 })
print (result["messages" ][-1 ].content)
2.2 CrewAI — 多代理协作框架 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 开发实践" })
print (result)
2.3 smolagents — HuggingFace 的轻量代理框架 smolagents 主打极简主义 ,整个框架核心仅几千行代码,适合快速原型和嵌入式场景。
from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel
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)
三、数据工程工具
3.1 Polars — 高性能 DataFrame 库 Polars 基于 Rust 编写,采用惰性求值(Lazy Evaluation)和多线程并行 ,在大多数基准测试中比 pandas 快 5-30 倍。
┌───────────────── Polars vs Pandas 性能对比 ──────────────────┐
│ │
│ 操作:读取 5 GB 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)
与 AI 结合:自动数据分析代理 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)
3.2 DuckDB — 嵌入式分析数据库 DuckDB 被称为"分析领域的 SQLite",支持直接查询 Parquet、CSV、JSON 等文件,无需导入数据。
import duckdb
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
""" )
df = result.pl()
print (df)
result.write_parquet("output/monthly_sales.parquet" )
3.3 Dagster — 现代数据编排平台 Dagster 3.0 将数据管道定义为软件定义资产(Software-Defined Assets) ,天然支持增量计算和血缘追踪。
┌──────────────── Dagster 数据管道血缘图 ────────────────┐
│ │
│ ┌────────────┐ ┌──────────────┐ ┌────────────┐ │
│ │ raw_events │────▶│ cleaned_data │───▶│ user_table│ │
│ └────────────┘ └──────┬───────┘ └─────┬──────┘ │
│ │ │ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌────────────┐ │ │
│ │ feature_store│ │ order_table│ │ │
│ └──────┬───────┘ └─────┬──────┘ │ │
│ │ │ │ │
│ └────────┬───────────┘ │ │
│ ▼ │ │
│ ┌──────────────┐ │ │
│ │ ml_training │ │ │
│ └──────┬───────┘ │ │
│ ▼ │ │
│ ┌──────────────┐ │ │
│ │ model_registry│ │ │
│ └──────────────┘ │ │
└─────────────────────────────────────────────────────────┘
代码示例:AI 训练数据管道 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("dow" ),
)
.drop_nulls(subset=["user_id" , "event_time" ])
)
context.log.info(f"清洗后剩余 {len (cleaned)} 条记录" )
return cleaned
@asset(
description="ML 训练特征表" ,
compute_kind="duckdb" ,
group_name="ml"
)
def feature_store (context: AssetExecutionContext, cleaned_data: pl.DataFrame ) -> MaterializeResult:
"""生成 ML 训练特征"""
result = duckdb.sql("""
SELECT user_id, category, COUNT(*) AS event_count,
AVG(amount) AS avg_amount, STDDEV(amount) AS std_amount,
COUNT(DISTINCT DATE(event_time)) AS active_days,
MAX(event_time) - MIN(event_time) AS activity_span
FROM cleaned_data
GROUP BY user_id, category
HAVING event_count >= 5
""" .pl())
result.write_parquet("output/features.parquet" )
return MaterializeResult(
metadata={
"row_count" : len (result),
"preview" : MetadataValue.md(result.head(5 ).to_pandas().to_markdown())
}
)
defs = Definitions(assets=[raw_events, cleaned_data, feature_store])
四、基础设施与工具链
4.1 LiteLLM — 统一 LLM API 网关 一个 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" )
4.2 FastAPI — 高性能 API 框架 + AI 集成 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 流式响应"""
from litellm import completion
response = completion(
model=request.model,
messages=[{"role" : "user" , "content" : request.message}],
stream=True ,
)
for chunk in response:
content = chunk.choices[0 ].delta.content or ""
if content:
yield f"data: {content} \n\n"
五、项目选型速查表 ┌────────────────────────────────────────────────────────────────────┐
│ 选型决策树 │
├────────────────────────────────────────────────────────────────────┤
│ │
│ 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 │
│ │
└────────────────────────────────────────────────────────────────────┘
六、项目 Star 增长趋势 GitHub Stars 增长趋势 (2024-2026)
120k ┤
│ ╭──── FastAPI
100k ┤ ╭───╯
│ ╭───╯
80k ┤ ╭───╯
│ ╭───╯
60k ┤ ╭───╯
│ ╭───╯
40k ┤ ╭───╯
│ ╭╯ ╭── LangGraph ─────╯
20k ┤ ╭╯ ╭── CrewAI │ ╭╯ ╭──╯ ╭── Polars
│ ╭╯ ╭─╯ ╭─╯ ╭── DuckDB │╭╯ ╭╯ ╭─╯ ╭─╯ ╭── Dagster
0k ┼╯──╯───╯─────╯─────╯────╯── LiteLLM
2024.1 2024.7 2025.1 2025.7 2026.1
七、总结与展望
2026 年 Python 开发者的核心技能栈 ┌────────────────────────────────────────────────────────────┐
│ │
│ Layer 4: 应用层 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ FastAPI + PydanticV2 + LiteLLM │ │
│ └──────────────────────────┬──────────────────────────┘ │
│ │ │
│ Layer 3: 代理编排层 │
│ ┌──────────────────────────┴──────────────────────────┐ │
│ │ LangGraph / CrewAI / smolagents │ │
│ └──────────────────────────┬──────────────────────────┘ │
│ │ │
│ Layer 2: 数据处理层 │
│ ┌──────────────────────────┴──────────────────────────┐ │
│ │ Polars + DuckDB + LanceDB │ │
│ └──────────────────────────┬──────────────────────────┘ │
│ │ │
│ Layer 1: 基础设施层 │
│ ┌──────────────────────────┴──────────────────────────┐ │
│ │ Python 3.13 + uv (包管理) + Dagster (编排) │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└────────────────────────────────────────────────────────────┘
关键趋势
AI 代理成为标配 :从简单的 ChatBot 到多代理协作系统,AI 代理正在成为每个应用的内置能力
Rust 加速 Python :Polars、DuckDB 等用 Rust 重写核心引擎,Python 生态性能飞升
SQL 回归 :DuckDB 让 SQL 分析重新成为数据工程师的首选
统一 LLM 接口 :LiteLLM 等工具让模型切换成本趋近于零
Asset-centric 编排 :Dagster 的资产管理模式正在取代传统 DAG
一句话总结 :2026 年的 Python 不再只是"脚本语言",它已经成为 AI 和数据工程的核心枢纽。掌握上述工具栈,将让你在这个快速演进的生态中保持竞争力。
本文所有代码基于 Python 3.13 + 最新版库编写,截至 2026 年 3 月。
微信扫一扫,关注极客日志 微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
相关免费在线工具 curl 转代码 解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
Base64 字符串编码/解码 将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
Base64 文件转换器 将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
Markdown 转 HTML 将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML 转 Markdown 互为补充。 在线工具,Markdown 转 HTML在线工具,online
HTML 转 Markdown 将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML 转 Markdown在线工具,online
JSON 压缩 通过删除不必要的空白来缩小和压缩JSON。 在线工具,JSON 压缩在线工具,online