资料入口 立即体验

Agently · 工程化输出控制与事件驱动编排

Agently AI应用开发框架

让输出可验证，让流程可编排

Demo 很容易，生产很困难。Agently 把模型不确定性关进工程边界。

把模型输出变成可验证接口，把多步行为变成可编排系统。

输出格式 + 关键字段保障 + 即时解析，让结果稳定可验。

响应对象固定一次请求，文本/数据/事件流可重复读取。

TriggerFlow 用实时信号驱动流程，运行时可触发、可扩展。

服务一上线，最先遇到的往往是吞吐与超时：异步优先，让等待变并发，并把并发上限收进边界。

一行安装： pip install -U Agently
开始构建 查看 GitHub 加入微信社群
结构化输出 结构化输出可控
响应对象 统一结果读取
异步优先 高并发友好
上线要求怎么落到框架？

一张对照表 + 一条能力路径，把共识变成可用能力。

继续阅读 →
为什么是信号驱动编排

when → to → emit + 运行时流：流程不是画出来的，是触发出来的。

继续阅读 →
核心能力，先看价值再看代码

输出稳定、即时解析、事件编排：每条都给你最小示例。

继续阅读 →
按任务读文档

从“我想做什么”出发，把关键文档放进一条阅读路径。

继续阅读 →

为什么Agently能力适合构建智能体系统

业界谈的 Context Engineering、结构化输出、事件驱动，本质上是在提出同一个问题：怎样把模型的不确定性，翻译成系统的确定性？
Agently 把“可上线要求”落成一条可直接调用的能力链路。

一眼看懂：上线要求怎么变成框架能力
输出要可验证
让结构稳定进入业务系统，能测试、能回归、能交付。
信息要能成为信号
即时解析让局部结构提前可用，触发运行态逻辑。
流程要在运行时可触发
when → to → emit 让事件回流，复杂流程按信号演进。
行业共识 Agently 能力
上下文工程
把上下文当成系统资产治理，而不是临时拼 Prompt。
分层提示词、变量映射、配置化提示词、会话/记忆。
结构化输出
输出要能稳定进入业务系统，才能测试、回归与交付。
输出格式 + 关键字段保障 + 即时解析。
事件驱动编排
复杂系统靠信号与状态演进，而不是静态流程图。
TriggerFlow 信号链路 + 运行时流式输出。
统一结果读取
一次请求的产物要可复用，避免隐式重复请求。
响应对象一次请求，多次读取文本/数据/事件流。
并发与边界
模型请求本质是 IO 等待：吞吐要可控，资源要可限流，失败要可恢复。
异步优先 + 并发上限控制（请求/Flow）。
观测与评估
过程要可追踪，才能定位问题、评估质量与迭代。
元信息、事件流、日志结构化。
工具与扩展
能力要可插拔，才能接入真实系统与团队资产。
工具调用、MCP 接入、auto_func。
数据与检索
让外部知识可结构化注入，才能稳定复用与更新。
结构化注入外部知识、KB 案例。

事件驱动工作流，而不是拖拽流程

TriggerFlow 以信号触发为核心语法，信号来自事件、运行时数据与流程级数据。 处理块完成后自动发出完成信号，下游继续触发；运行时流让过程可被实时输出。

Agently 先把输出做“可解析”，再把关键信息做“可触发”。
基础层
Token 流
模型 token-by-token 输出
结构化输出
结果可被程序消费
即时解析
局部结构提前可用
实时信号
关键信息可触发
编排层
TriggerFlow
信号驱动编排系统
只有当“结构化输出的局部信息”能被即时解析为信号，事件驱动编排才真正成立。
查看 TriggerFlow 教程 运行时流式输出
Flow（蓝图）
先定义 chunk，再用 when / to 描述信号链路。
chunk
when
to
emit
等待信号 → 执行 chunk → 发出完成信号
Execution（执行实例）
每次请求生成自己的 execution，事件在运行时触发与回流。
Event Bus
chunk.done runtime_data.set custom.emit
提示：runtime_data 的更新也会触发事件，可被 when(...) 截获并触发下游任务。

适用场景

从真实需求出发，看看如何更快交付到生产。

PRD → 测试用例生成

需求变更频繁，用例字段对不上，回归成本居高不下。

把输出做成稳定结构，并用关键字段保障进行校验，让用例可自动回归。

输出控制 → 场景案例 →

问卷调查 / 多轮访谈

问题顺序、分支路由、并发采集，越做越像一套系统。

用事件与运行时数据驱动流程，在运行中按回答触发下一步。

TriggerFlow → when 用法 →

知识库问答 / 客服助手

引用来源、答案结构与元信息要一致，才能接入你的业务系统。

用分层 Prompt 与响应对象，把检索、回答、引用与元信息统一读写。

提示词管理 → 结果读取 →

实时交互 / 具身反馈

你希望边生成边行动：推理、动作与输出能在一次请求里更早到达前端。

即时解析把局部结构变成实时信号，再用运行时流把过程稳定地推到用户侧。

即时解析 → 运行时流 →

核心能力与代码示例

先看能力与价值，再看最小可运行代码。

结构化输出 + 关键字段保障（ensure_keys）
from agently import Agently

Agently.set_settings(
  "OpenAICompatible",
  {
    "base_url": "http://127.0.0.1:11434/v1",
    "api_key": "YOUR_API_KEY",
    "model": "qwen2.5:7b",
  },
)

agent = Agently.create_agent()

result = (
  agent
    .input("给我输出3个单词和2个句子")
    .output({
      "单词": [("str",)],
      "句子": [("str",)],
    })
    .start(
      ensure_keys=["单词[*]", "句子[*]"],
      max_retries=1,
      raise_ensure_failure=False,
    )
)
print(result)
from agently import Agently

Agently.set_settings(
  "OpenAICompatible",
  {
    "base_url": "http://127.0.0.1:11434/v1",
    "api_key": "YOUR_API_KEY",
    "model": "qwen2.5:7b",
  },
)

agent = Agently.create_agent()

result = (
  agent
    .input("Give me 3 words and 2 sentences")
    .output({
      "words": [("str",)],
      "sentences": [("str",)],
    })
    .start(
      ensure_keys=["words[*]", "sentences[*]"],
      max_retries=1,
      raise_ensure_failure=False,
    )
)
print(result)
运行结果
{'单词': ['苹果', '电脑', '学习'], '句子': ['我喜欢吃苹果。', '他用电脑工作。']}
{'words': ['apple', 'computer', 'learning'], 'sentences': ['I like apples.', 'He works on a computer.']}
即时结构化流式解析（instant）
response = (
  agent
    .input("解释递归，并给出 2 个提示")
    .output({
      "definition": ("str", "Short definition"),
      "tips": [("str", "Short tip")],
    })
    .get_response()
)

for msg in response.get_generator(type="instant"):
  if msg.path == "definition" and msg.delta:
    print(msg.delta, end="", flush=True)
  if msg.wildcard_path == "tips[*]" and msg.delta:
    print(msg.delta, end="", flush=True)
print()
response = (
  agent
    .input("Explain recursion and give 2 tips")
    .output({
      "definition": ("str", "Short definition"),
      "tips": [("str", "Short tip")],
    })
    .get_response()
)

for msg in response.get_generator(type="instant"):
  if msg.path == "definition" and msg.delta:
    print(msg.delta, end="", flush=True)
  if msg.wildcard_path == "tips[*]" and msg.delta:
    print(msg.delta, end="", flush=True)
print()
流式输出片段
▍
▍
事件编排最小示例
from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
async def normalize(data: TriggerFlowEventData):
  return str(data.value).strip().lower()

@flow.chunk
async def greet(data: TriggerFlowEventData):
  return f"hi {data.value}"

flow.to(normalize).to(greet).end()
print(flow.start(" Agently "))
运行结果
hi agently
配置化提示词（YAML）
# prompt.yaml
.agent:
  system: 你是企业知识助手
.request:
  input: "{question}"
  output:
    answer:
      $type: str
      $desc: 一句话回答

# load
from agently import Agently
agent = Agently.create_agent()
agent.load_yaml_prompt("prompt.yaml", mappings={"question": "Agently 有哪些能力？"})
print(agent.start())
# prompt.yaml
.agent:
  system: You are an enterprise knowledge assistant.
.request:
  input: "{question}"
  output:
    answer:
      $type: str
      $desc: One-sentence answer

# load
from agently import Agently
agent = Agently.create_agent()
agent.load_yaml_prompt("prompt.yaml", mappings={"question": "What can Agently do?"})
print(agent.start())
运行结果
{'answer': '输出可控、流程可编排、工程化可落地。'}
{'answer': 'Stable outputs, orchestrated workflows, and production-ready delivery.'}

资料与文档入口

从你的下一个任务开始：安装、模型设置、输出控制、提示词管理与事件编排，一路走到可上线。

快速开始

我想跑通第一个请求：装好、设模型、拿到结构化输出与即时解析。

进入快速开始

模型设置

我想接入/切换模型：用 OpenAI 风格接口统一接入，密钥与参数集中管理。

模型设置

模型输出控制

我想让输出稳定：输出格式 + 关键字段保障 + 即时解析。

模型输出控制

模型返回结果

我想一次请求多次取用：文本、结构化数据、元信息与事件流统一读取。

模型返回结果

提示词管理

我想让提示词可维护：分层、变量映射、配置化与版本化。

提示词管理

事件编排

我想把流程跑起来：信号触发、分支并发、以及运行时流式输出。

事件编排综述

Agent 扩展能力

我想接入业务能力：工具调用、函数接口与会话管理。

Agent 扩展

场景实战案例

我想照着做：完整业务场景方案与可运行示例。

场景实战案例

加入 Agently v4 研发共创计划

获取路线图、试验能力与生产落地支持。

加入微信群 联系团队
Agently community QR code