AI工作流框架 是 AI Skill Hub 本期精选Agent工作流之一。综合评分 8.0 分,整体质量较高。我们强烈推荐将其纳入你的 AI 工具库,帮助提升工作效率。
AI工作流框架 是一套完整的 AI Agent 自动化工作流方案。通过可视化的节点编排,将复杂的多步骤任务拆解为清晰的自动化流程,实现全程无人值守的智能处理。支持与数百种外部服务和 API 无缝集成,适合构建数据处理管线、业务自动化和 AI 辅助决策系统。
AI工作流框架 是一套完整的 AI Agent 自动化工作流方案。通过可视化的节点编排,将复杂的多步骤任务拆解为清晰的自动化流程,实现全程无人值守的智能处理。支持与数百种外部服务和 API 无缝集成,适合构建数据处理管线、业务自动化和 AI 辅助决策系统。
# 方式一:pip 安装(推荐)
pip install selectools
# 方式二:虚拟环境安装(推荐生产环境)
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install selectools
# 方式三:从源码安装(获取最新功能)
git clone https://github.com/johnnichev/selectools
cd selectools
pip install -e .
# 验证安装
python -c "import selectools; print('安装成功')"
# 命令行使用
selectools --help
# 基本用法
selectools input_file -o output_file
# Python 代码中调用
import selectools
# 示例
result = selectools.process("input")
print(result)
# selectools 配置文件示例(config.yml) app: name: "selectools" debug: false log_level: "INFO" # 运行时指定配置文件 selectools --config config.yml # 或通过环境变量配置 export SELECTOOLS_API_KEY="your-key" export SELECTOOLS_OUTPUT_DIR="./output"
┌─┐┌─┐┬ ┌─┐┌─┐┌┬┐┌─┐┌─┐┬ ┌─┐
└─┐├┤ │ ├┤ │ │ │ ││ ││ └─┐
└─┘└─┘┴─┘└─┘└─┘ ┴ └─┘└─┘┴─┘└─┘
An open-source project from NichevLabs.
Multi-agent orchestration in plain Python. Build agent graphs, compose pipelines with |, deploy with one command. No DSL, no compile step, no paid debugger. Works with OpenAI, Anthropic, Gemini, and Ollama.
AgentAPI — production REST endpoints (chat, SSE streaming, sessions) from any agentresponse_format with auto-retryresult.trace with typed timeline of every agent stepresult.reasoning explains why the agent chose a toolagent.batch() / agent.abatch() for concurrent classificationAgentScheduler runs an agent on a cron or interval schedule with per-job max-runs, failure isolation, and an async loopthink/analyze tools make reasoning explicit, inspectable, and bounded by min/max stepsPromptInjectionGuardrail — heuristic jailbreak/injection detection with high-precision patternsToolResult base class + Artifact side-channel via emit_artifact()selectools.pending for chat-channel destructive-tool confirmationcache_system/cache_tools with hit-rate visibility on UsageStatsremember + recall tools, pluggable stores (File, SQLite), importance scoring, TTLmax_total_tokens, max_cost_usd hard limits; CancellationToken for cooperative stoppingestimate_run_tokens() for pre-execution budget checksmodel_selector callback for per-iteration model selectionSemanticCache — embedding-based cache hits for paraphrased queries (cosine similarity, LRU + TTL)compress_context, compress_threshold, compress_keep_recentConversationMemory.branch() and SessionStore.branch() for A/B exploration and checkpointingAgentGraph with routing, parallel execution, HITL, checkpointing; SupervisorAgent with 4 strategies (plan_and_execute, round_robin, dynamic, magentic)Pipeline + @step + | operator + parallel() + branch() — chain agents, tools, and transformsrun_id correlation, LoggingObserver, SimpleStepObserver, OTel export```python
Two user-facing features plus a post-ship bug-hunt sweep that pinned 8 code-generator fixes in the visual builder.
SupabaseSessionStore — 4th SessionStore backend alongside JSON, SQLite, and Redis. Postgres-backed via Supabase PostgREST, with idempotent upserts, namespace isolation, and the same validation guards as RedisSessionStore. Optional dep: pip install selectools[supabase]. Demo: examples/96_supabase_session_store.py.Retriever (RAG) onto the canvas and pick any of 7 vector stores (memory, SQLite, Chroma, Pinecone, FAISS, Qdrant, pgvector), toggle Hybrid (BM25 + vector + RRF) and cross-encoder Rerank. Drag Session Store as a resource node and wire it into an agent via the new Session Store dropdown. Two new presets: Hybrid RAG and Multi-Tenant RAG. Python + YAML code generators emit real, runnable code.from supabase import create_client
from selectools import SupabaseSessionStore, Agent, AgentConfig
store = SupabaseSessionStore(client=create_client(URL, KEY))
agent = Agent(
tools=[...],
config=AgentConfig(session_store=store, session_id="u-1", max_iterations=5),
)
See CHANGELOG.md for the full entry including the 8 builder code-gen fixes.
The first AI agent framework to ship a visual graph builder in a single pip install. No React. No build step. No CDN.
Try the builder in your browser → — no install required.
```bash pip install selectools selectools serve --builder
```
"approved") for conditional routing```python
graph = AgentGraph() graph.add_node("planner", planner_agent) graph.add_node("writer", writer_agent) graph.add_node("reviewer", reviewer_agent) graph.add_edge("planner", "writer") graph.add_edge("writer", "reviewer") graph.add_edge("reviewer", AgentGraph.END) graph.set_entry("planner") result = graph.run("Write a blog post about AI safety")
pip install selectools # Core + basic RAG
pip install selectools[rag] # + Chroma, Pinecone, FAISS, Qdrant, Voyage, Cohere, PyPDF, BeautifulSoup
pip install selectools[observe] # + OpenTelemetry, Langfuse observers
pip install selectools[postgres] # + psycopg2 (enables pgvector)
pip install selectools[cache] # + Redis cache
pip install selectools[mcp] # + MCP client/server
pip install "selectools[rag,observe,cache,mcp]" # Everything
Add your provider's API key to a .env file in your project root:
``` OPENAI_API_KEY=sk-...
config = AgentConfig(reasoning_strategy="react") # Thought → Action → Observation config = AgentConfig(reasoning_strategy="cot") # Chain-of-Thought step-by-step config = AgentConfig(reasoning_strategy="plan_then_act") # Plan first, then execute
New to Selectools? Follow the 5-minute Quickstart tutorial — no API key needed.
Examples are numbered by difficulty. Start from 01 and work your way up.
| # | Example | Features | API Key? |
|---|---|---|---|
| 01 | 01_hello_world.py | First agent, @tool, ask() | No |
| 02 | 02_search_weather.py | ToolRegistry, multiple tools | No |
| 03 | 03_toolbox.py | 56 pre-built tools (file, data, text, datetime, web, code, search, and more) | No |
| 04 | 04_conversation_memory.py | Multi-turn memory | Yes |
| 05 | 05_cost_tracking.py | Token counting, cost warnings | Yes |
| 06 | 06_async_agent.py | arun(), concurrent agents, FastAPI | Yes |
| 07 | 07_streaming_tools.py | Generator-based streaming | Yes |
| 08 | 08_streaming_parallel.py | astream(), parallel execution, StreamChunk | Yes |
| 09 | 09_caching.py | InMemoryCache, RedisCache, cache stats | Yes |
| 10 | 10_routing_mode.py | Routing mode, intent classification | Yes |
| 11 | 11_tool_analytics.py | Call counts, success rates, timing | Yes |
| 12 | 12_observability_hooks.py | Lifecycle observers, tool validation | Yes |
| 13 | 13_dynamic_tools.py | ToolLoader, plugins, hot-reload | Yes |
| 14 | 14_rag_basic.py | RAG pipeline, document loading, vector search | Yes + [rag] |
| 15 | 15_semantic_search.py | Pure semantic search, metadata filtering | Yes + [rag] |
| 16 | 16_rag_advanced.py | PDFs, SQLite persistence, custom chunking | Yes + [rag] |
| 17 | 17_rag_multi_provider.py | Embedding/store/chunk-size comparisons | Yes + [rag] |
| 18 | 18_hybrid_search.py | BM25 + vector fusion, RRF, reranking | Yes + [rag] |
| 19 | 19_advanced_chunking.py | Semantic and contextual chunking | Yes + [rag] |
| 20 | 20_customer_support_bot.py | Multi-tool customer support workflow | Yes |
| 21 | 21_data_analysis_agent.py | Data exploration and analysis | Yes |
| 22 | 22_ollama_local.py | Fully local LLM via Ollama | No (Ollama) |
| 23 | 23_structured_output.py | Pydantic response_format, auto-retry, JSON extraction | No |
| 24 | 24_traces_and_reasoning.py | AgentTrace timeline, reasoning visibility, JSON export | No |
| 25 | 25_provider_fallback.py | FallbackProvider, circuit breaker, failover chain | No |
| 26 | 26_batch_processing.py | batch(), abatch(), structured batch, error isolation | No |
| 27 | 27_tool_policy.py | ToolPolicy, deny_when, HITL approval, memory trimming | No |
| 28 | 28_agent_observer.py | AgentObserver, LoggingObserver, multiple observers, OTel export | No |
| 29 | 29_guardrails.py | Input/output guardrails, PII redaction, topic blocking | No |
| 30 | 30_audit_logging.py | JSONL audit logging, privacy controls, daily rotation | No |
| 31 | 31_tool_output_screening.py | Prompt injection detection in tool outputs | No |
| 32 | 32_coherence_checking.py | LLM-based intent verification for injection defense | Yes |
| 33 | 33_persistent_sessions.py | JsonFileSessionStore, cross-restart persistence | No |
| 34 | 34_summarize_on_trim.py | Summarize trimmed messages for context preservation | No |
| 35 | 35_entity_memory.py | Named entity extraction and tracking | No |
| 36 | 36_knowledge_graph.py | Triple extraction, in-memory and SQLite storage | No |
| 37 | 37_knowledge_memory.py | Cross-session facts, daily logs, remember tool | No |
| 38 | 38_terminal_tools.py | @tool(terminal=True), stop_condition callback | No |
| 39 | 39_eval_framework.py | EvalSuite, TestCase, evaluators, HTML reports | No |
| 40 | 40_eval_advanced.py | Pairwise A/B, regression detection, snapshots | No |
| 41 | 41_mcp_client.py | MCPClient, mcp_tools(), tool interop | No |
| 42 | 42_mcp_server.py | MCPServer, expose tools as MCP endpoints | No |
| 43 | 43_token_budget.py | max_total_tokens, max_cost_usd budget limits | No |
| 44 | 44_cancellation.py | CancellationToken, cooperative stopping | No |
| 45 | 45_approval_gate.py | @tool(requires_approval=True), confirm_action | No |
| 46 | 46_simple_observer.py | SimpleStepObserver, single-callback integration | No |
| 47 | 47_token_estimation.py | estimate_run_tokens(), pre-flight cost checks | No |
| 48 | 48_model_switching.py | model_selector callback, per-iteration model | No |
| 49 | 49_knowledge_stores.py | SQLite, Redis, Supabase knowledge stores | No |
| 50 | 50_reasoning_strategies.py | ReAct, Chain-of-Thought, Plan-then-Act | No |
51_tool_result_caching.py | `@tool(cacheable=True, cache_
config = AgentConfig(
model="gpt-4o-mini",
temperature=0.0,
max_tokens=2000,
max_iterations=6,
max_retries=3,
retry_backoff_seconds=2.0,
request_timeout=60.0,
tool_timeout_seconds=30.0,
cost_warning_threshold=0.50,
parallel_tool_execution=True,
routing_only=False,
stream=False,
cache=None, # InMemoryCache or RedisCache
tool_policy=None, # ToolPolicy with allow/review/deny rules
confirm_action=None, # Human-in-the-loop approval callback
approval_timeout=60.0, # Seconds before auto-deny
enable_analytics=True,
verbose=False,
observers=[LoggingObserver()], # Lifecycle observer (replaces deprecated hooks)
system_prompt="You are a helpful assistant...",
)
agent.run("hello") # Just works inside async contexts ```
typing.Literal crashed @tool(), asyncio.run() re-entry in 8 sync wrappers, HITL silently lost in parallel groups + subgraphs, ConversationMemory had no thread lock<think> tag stripping, RAG batch limits, MCP concurrent race, str→int/float/bool argument coercion, Union[str, int] support, multi-interrupt generators, GraphState fail-fast validation, session namespace isolation, summary growth capAgentTrace lock, async observer exception logging, batch clone isolation, OTel/Langfuse observer locks, vector store search dedup, Optional[T] without default handlingtests/agent/test_regression.py, each with empirical fault-injection verification (test fails without fix, passes after)ConversationMemory, AgentTrace, OTelObserver, LangfuseObserver, MCPClient, FallbackProvider, batch clone isolationSee CHANGELOG.md for the full per-bug breakdown with cross-references to every original Agno/PraisonAI issue.
Every public class and function exported from selectools now carries a stability marker:
from selectools import Agent, AgentGraph, PlanAndExecuteAgent
print(Agent.__stability__) # "stable"
print(AgentGraph.__stability__) # "beta"
print(PlanAndExecuteAgent.__stability__) # "beta"
@stable — 60+ core symbols (Agent, AgentConfig, providers, memory, tools, evals, guardrails, sessions, knowledge, cache, cancellation)
@beta — 30+ newer symbols (AgentGraph, SupervisorAgent, Pipeline, @step, parallel, branch, all four patterns, compose)
```
from selectools import Agent, AgentConfig, tool
from selectools.providers.stubs import LocalProvider
@tool(description="Look up the price of a product")
def get_price(product: str) -> str:
prices = {"laptop": "$999", "phone": "$699", "headphones": "$149"}
return prices.get(product.lower(), f"No price found for {product}")
agent = Agent(
tools=[get_price],
provider=LocalProvider(),
config=AgentConfig(max_iterations=3),
)
result = agent.ask("How much is a laptop?")
print(result.content)
@stable class MyProductionAgent: ...
@beta class MyExperimentalFeature: ...
@deprecated(since="0.19", replacement="MyProductionAgent") class MyOldAgent: ...
```python from selectools import Pipeline, step, parallel, branch
@step def summarize(text: str) -> str: return agent.run(f"Summarize: {text}").content
@step def translate(text: str, lang: str = "es") -> str: return agent.run(f"Translate to {lang}: {text}").content
tools = ToolLoader.from_directory("./plugins", recursive=True) agent.add_tools(tools)
updated = ToolLoader.reload_file("./plugins/search.py") agent.replace_tool(updated[0])
agent.ask("What is Python?") agent.reset() agent.ask("What is Python?")
print(cache.stats) # CacheStats(hits=1, misses=1, hit_rate=50.00%) ```
For distributed setups: from selectools.cache_redis import RedisCache
高质量的AI工作流框架,内置安全防护和审计功能
AI Skill Hub 为第三方内容聚合平台,本页面信息基于公开数据整理,不对工具功能和质量作任何法律背书。
建议在沙箱或测试环境中充分验证后,再部署至生产环境,并做好必要的安全评估。
✅ Apache 2.0 — 宽松开源协议,可商用,需保留版权声明和 NOTICE 文件,含专利授权条款。
经综合评估,AI工作流框架 在Agent工作流赛道中表现稳健,质量优秀。如果你已有明确的使用需求,可以直接上手体验;如果还在评估阶段,建议对比同类工具后再做决策。
| 原始名称 | selectools |
| 原始描述 | 开源AI工作流:Production-ready Python framework for AI agents with built-in guardrails, audit 。⭐10 · Python |
| Topics | ai-agentsai-safetyenterprise-ai |
| GitHub | https://github.com/johnnichev/selectools |
| License | Apache-2.0 |
| 语言 | Python |
收录时间:2026-06-15 · 更新时间:2026-06-15 · License:Apache-2.0 · AI Skill Hub 不对第三方内容的准确性作法律背书。
选择 Agent 类型,复制安装指令后粘贴到对应客户端