AgentFlow — 从 Agent 代码到生产 API，仅需几分钟

Source: Dev.to

为什么选择 AgentFlow？

大多数代理框架止步于原型阶段。你只能得到一个可爱的演示，然后需要花数周时间去添加身份验证、速率限制、持久化以及前端。AgentFlow 为演示之后的工作而构建。

一个框架。 从第一次 pip install 到生产环境的 Docker 部署。

🔗 链接

完整堆栈

agentflow            →  Core Python orchestration engine
agentflow-cli        →  FastAPI server + CLI tooling
agentflow-client     →  TypeScript/React SDK (@10xscale/agentflow-client)
agentflow-playground →  Hosted UI for testing agents

可以单独使用任意层，也可以将它们组合起来，构建完整的 AI 产品堆栈——从 LLM 调用到浏览器 UI——无需拼接四个不同的库。

60 秒快速上手

pip install 10xscale-agentflow-cli

agentflow init   # scaffold a new project
agentflow api    # start the dev server
agentflow play   # open the playground UI

就这样。你的代理在不到一分钟的时间内即可运行、流式输出并进行探索。

您将获得的内容

基于图的代理编排

AgentFlow 使用 StateGraph —— 有向节点、条件边以及对执行流的完整控制。没有黑箱。没有无法调试的神秘路由。

from agentflow.graph import Agent, StateGraph, ToolNode
from agentflow.state import AgentState, Message
from agentflow.utils.constants import END

def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"The weather in {location} is sunny, 72°F"

graph = StateGraph()
graph.add_node(
    "MAIN",
    Agent(
        model="gemini/gemini-2.5-flash",
        system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
        tool_node_name="TOOL",
    ),
)
graph.add_node("TOOL", ToolNode([get_weather]))

def route(state: AgentState) -> str:
    if state.context and state.context[-1].tools_calls:
        return "TOOL"
    return END

graph.add_conditional_edges("MAIN", route, {"TOOL": "TOOL", END: END})
graph.add_edge("TOOL", "MAIN")
graph.set_entry_point("MAIN")

app = graph.compile()
result = app.invoke(
    {"messages": [Message.text_message("Weather in NYC?")]},
    config={"thread_id": "1"},
)

有状态。工具调用。不到 30 行代码。

与 LLM 无关

只需传入模型字符串；AgentFlow 会自动路由。

无需学习特定提供商的抽象。更换模型时不必触及代理逻辑。

并行工具执行 —— 自动化

当 LLM 同时调用多个工具时，AgentFlow 会并发运行它们。无需额外配置。

其他框架:  1.0s + 1.5s + 0.8s = 3.3s
AgentFlow:         max(1.0s, 1.5s, 0.8s) = 1.5s  ⚡ 2.2x 更快

生产级记忆 —— 三层结构

工作记忆    →  当前执行状态 (AgentState)
会话记忆    →  Redis（热）+ PostgreSQL（持久）检查点
知识记忆    →  Qdrant 向量库 + Mem0 语义召回

Redis 提供快速的热会话状态。PostgreSQL 提供持久且水平可扩展的存储。两者协同运行——无需在二者之间取舍。

流式输出

stream_gen = app.astream(
    inp,
    config=config,
    response_granularity=ResponseGranularity.LOW,
)
async for chunk in stream_gen:
    print(chunk.model_dump())

三种粒度级别：逐 token（类似 ChatGPT）、逐消息、逐节点图踪。前端自行决定展示方式。

身份验证与安全 —— 内置而非后加

大多数框架把身份验证留给使用者自行实现。AgentFlow 已经帮你做好。

{ "auth": "jwt" }
{ "auth": null }
{ "auth": { "method": "custom", "path": "auth.my_backend:MyAuth" } }

在 agentflow.json 中只需一行配置。无需改动图代码即可在开发和生产环境之间切换身份验证方式。

已包含的安全特性：

• 可配置密钥的 JWT 身份验证
• 支持 OAuth2、API Key、会话等自定义身份验证后端
• 基于角色的访问控制（RBAC）
• 滑动窗口限流（内存或 Redis 后端）
• 可配置的请求大小限制（防 DoS，默认 10 MB）
• 自动从日志中脱敏 token 与密钥
• 启动时校验 —— 在你意外部署不安全的 CORS 或调试模式前给出警告

生命周期回调

在执行的每一层都可以挂钩 —— 在每次 LLM 调用、工具调用或 MCP 调用的前后，亦或在图本身的开始、结束、检查点、打断、恢复、错误等事件上。可用于日志记录、链路追踪、自定义指标或动态修改图结构。

快速入门回顾

pip install 10xscale-agentflow-cli
agentflow init
agentflow api
agentflow play

您现在拥有一个已准备好投入生产、具备身份验证保护、支持流式传输、多代理的 AI 堆栈，随时可以发布。

AgentFlow 概述

AgentFlow 让您使用基于图的 DSL、自动持久化、并行工具执行以及完整的 CLI 来构建生产级别的 LLM 代理。无需样板代码，无需自定义状态管理。

命令行界面

agentflow init                 # 脚手架项目 + 配置
agentflow api                  # 开发服务器，自动重新加载
agentflow play                 # 打开针对本地后端的 Playground
agentflow build --docker-compose  # 生成 Dockerfile + compose

自动生成的 FastAPI 端点

Your agent graph becomes a production API—no FastAPI boilerplate required.

Dependency Injection with InjectQ

from agentflow.utils import tool

@tool(tags=["weather"])
async def get_weather(
    location: str,
    user_id: str = Inject(UserService)
) -> str:
    """Get weather for a location."""
    return f"Weather for user {user_id} in {location}: sunny"

• 干净、可测试的工具。
• 每请求上下文，无全局状态。

Human‑in‑the‑Loop

• 在图的中途暂停执行。
• 注入人工决策。
• 在保持完整状态的情况下恢复——无需重新运行之前的步骤。

使用场景：审批工作流、审核关卡、交互式调试。

事件发布

React/TypeScript 客户端 SDK

@10xscale/agentflow-client 提供 React hooks（useAgent、useStream、useThreads），支持 ChatGPT 风格 UI 的 token 级别流式传输，以及客户端工具执行。前端可以直接与您的 AgentFlow API 通信，无需自定义集成代码。

功能比较

安装

# Core library
pip install 10xscale-agentflow

# Full CLI + API server
pip install 10xscale-agentflow-cli

可选额外组件

pip install 10xscale-agentflow[pg_checkpoint]   # PostgreSQL + Redis persistence
pip install 10xscale-agentflow[mcp]             # Model Context Protocol
pip install 10xscale-agentflow[google-genai]    # Google GenAI adapter
pip install 10xscale-agentflow[kafka]           # Kafka event publishing
pip install 10xscale-agentflow[redis]           # Redis publisher + rate limiting

当前版本

在 v0.7.x 中新增： 多模态支持（图像、音频、视频），扩展推理/链式思考，3 层记忆，回调和生命周期钩子，代理技能，Vertex AI 支持，结构化 Pydantic 输出。

路线图

• ✅ Graph engine with nodes, edges, and conditional routing
• ✅ Redis + PostgreSQL state checkpointing
• ✅ Tool integration — local Python, MCP, optional adapters
• ✅ Parallel tool execution
• ✅ Lifecycle callbacks and graph hooks
• ✅ Streaming + event publishing
• ✅ Human‑in‑the‑loop
• ✅ Multimodal agents
• 🚧 Remote node execution for distributed processing
• 🚧 OpenTelemetry tracing
• 🚧 More persistence backends (DynamoDB, etc.)
• 🚧 Visual graph editor

隐私与许可

• MIT 许可证 – 可免费用于商业用途。
• 不收集数据 – 对话保留在您的基础设施上。
• 无按调用计费 – 您只需支付 LLM API 和基础设施费用。
• 随处部署 – Docker、Kubernetes、AWS ECS、Cloud Run、Azure、Heroku。

链接

由 10xScale 和社区构建。MIT 许可证。