项目结构
单脚本适合试 API。项目开始变大后,最先出问题的通常不是代码行数,而是边界:模型配置散落在脚本里,prompt 和业务逻辑混在一起,Actions 注册到处复制,服务端 wrapper 和 flow 定义互相 import。
一个 Agently 项目可以先按下面的骨架拆开。不是所有目录一开始都要有,但 settings.yaml、prompts/、app/agents.py 这条主线建议尽早建立。
先定义服务契约
服务化不是先选 FastAPI、Flask、gRPC 还是消息队列。它的本质是接口化约束:你向调用方承诺“给我这些输入,我返回这样的输出;出现错误时按这样的语义处理”。协议只是承载这个承诺的方式。
把一段 Agently 能力交给前端、Go/Java 服务、IM 网关或后台系统前,先回答这些问题:
| 契约问题 | 需要定清楚什么 |
|---|---|
| 输入是什么 | 字段名、类型、必填/可选、自然语言文本和业务 payload 的边界 |
| 输出是什么 | 结构化字段、最终文本、artifact ref、错误形态和业务状态码 |
| 过程怎么看 | 是否需要 instant stream、TriggerFlow runtime stream、SSE/WebSocket 或 IM 回推 |
| 怎么调用 | HTTP、SSE、WebSocket、CLI、队列、MCP 或内部函数 |
| 谁负责配置 | 模型 endpoint、key、prompt、环境变量和运行时策略放在哪一层 |
| 谁负责错误 | 请求体错误、模型超时、Action 失败、审批 pending、流程中断分别怎么返回 |
契约一旦发布,调用方不应该关心内部用了 Agently、TriggerFlow、DeepSeek、Ollama 还是别的实现。你的服务边界应该保护内部演进自由。
职责分层
项目结构的核心不是“目录多一点”,而是每一层只知道自己需要知道的事:
| 层 | 负责 | 不负责 |
|---|---|---|
| 接入层 | HTTP/SSE/WebSocket、请求校验、鉴权、错误包装、版本前缀 | 写 prompt、直接调模型、决定业务路径 |
| 业务层 | AgentExecution、Actions 注册、TriggerFlow、业务服务对象 | 感知 HTTP 细节、直接读环境变量 |
| 配置层 | settings、prompt files、provider、运行时策略、环境差异 | 混入业务判断或接口返回格式 |
这样做的收益是可替换:HTTP 换队列只动接入层,模型 provider 换私有部署只动配置层,流程从单请求升级到 TriggerFlow 时仍能复用业务 service。
推荐布局
my-agently-app/
pyproject.toml
.env # 本地敏感配置,加入 gitignore
settings.yaml # 全局模型和运行时设置
prompts/
summarize.yaml
triage.yaml
app/
agents.py # agent 工厂
actions.py # action 注册
services.py # 业务 service wrapper
api.py # FastAPI 入口
flows/
ticket_triage.py # TriggerFlow 定义
tests/
test_ticket_triage.py先把模型设置、prompt 和 agent 创建逻辑拆开,后面接 Actions、TriggerFlow、FastAPI 时会更自然。
settings.yaml
OpenAICompatible:
base_url: ${ENV.OPENAI_BASE_URL}
api_key: ${ENV.OPENAI_API_KEY}
model: ${ENV.OPENAI_MODEL}
runtime:
show_model_logs: false启动时加载:
from agently import Agently
Agently.load_settings("yaml_file", "settings.yaml", auto_load_env=True).env 放本地 key,settings.yaml 放结构。这样换模型或 endpoint 时,不需要改业务代码。
Prompt 写进文件
# prompts/triage.yaml
.agent:
system: 你是一个工单分流助手。
info:
severities: ["P0", "P1", "P2", "P3"]
.execution:
instruct: 对工单文本分类,并给出一句理由。
output:
$format: json
severity:
$type: str
$desc: P0/P1/P2/P3 之一
$ensure: true
rationale:
$type: str
$desc: 一句理由
$ensure: true加载:
agent = Agently.create_agent().load_yaml_prompt("prompts/triage.yaml")Prompt 文件适合 review。产品、运营或领域专家不需要读 Python,也能看懂模型被要求做什么。
Agent 工厂
# app/agents.py
from agently import Agently
def make_triage_agent():
return (
Agently
.create_agent("ticket-triage")
.load_yaml_prompt("prompts/triage.yaml")
)调用点只拿 agent,不重复配置模型、不重复加载 prompt。
Actions 放在哪里
如果 agent 要调用业务能力,集中写在 app/actions.py:
# app/actions.py
def register_ticket_actions(agent, ticket_repo):
@agent.action_func
async def lookup_ticket(ticket_id: str):
return await ticket_repo.get(ticket_id)
agent.use_actions([lookup_ticket])
return agent新代码使用 action-first 入口:@agent.action_func、agent.use_actions(...)、内置 action packages 和 Agent Component helpers。旧的 tools API 可以维护旧项目,但不要作为新项目默认路径。
Flow 放在哪里
每个 TriggerFlow 放一个模块,flow 定义不要耦合 FastAPI、队列或 CLI:
# flows/ticket_triage.py
from agently import TriggerFlow
def build_ticket_triage_flow(agent):
flow = TriggerFlow(name="ticket-triage")
# flow.to(...)
return flow服务层 import flow,再创建 execution。这样 flow 可以被 HTTP、worker、测试同时复用。
Service wrapper
业务系统通常不直接暴露 Agently 原始 result 或 snapshot。可以在 app/services.py 里投影:
class TicketTriageService:
def __init__(self, agent):
self.agent = agent
async def triage(self, ticket_text: str):
result = (
self.agent
.input(ticket_text)
.output({
"severity": (str, "P0/P1/P2/P3", True),
"rationale": (str, "一句理由", True),
})
.get_result()
)
return await result.async_get_data()这层负责把 Agently 的执行结果变成业务对象。
测试放什么
先测确定性边界:
- settings 文件能加载。
- prompt 文件能渲染。
- output schema 包含必填字段。
- action mock 能被调用。
- service wrapper 返回业务对象形态。
模型语义正确性不要用关键词或快照硬测。需要语义质量时,用明确规则或 model judge。
常见误用
- 把 API key、prompt、业务函数、HTTP 路由都写在一个脚本里。
- 每个 endpoint 重新创建和配置 agent。
- Flow 定义里 import FastAPI app。
- Action 注册散落在多个调用点,导致同一个 agent 挂载能力不一致。
- 测试直接断言模型自然语言全文。
下一步
- 设置文件:设置
- Prompt 文件:Prompt 管理
- Action 注册:Actions 概览
- Flow 设计:TriggerFlow 概览
- 服务封装:FastAPI 服务封装