Agently框架实战：从AI原型到生产部署的工程化解决方案

1. 项目概述：Agently，一个为生产环境而生的AI应用开发框架

如果你和我一样，在过去的两年里尝试过用LangChain、CrewAI或者AutoGen来构建一个真正能上线的AI应用，大概率会经历一个相似的循环：在Jupyter Notebook里快速搭建原型时感觉无比顺畅，充满了“AI即将改变世界”的兴奋感；但一旦要把这个原型部署到生产环境，面对真实的用户和流量，各种问题就接踵而至了。模型输出格式飘忽不定，导致下游业务逻辑频繁崩溃；多轮对话的状态管理像一团乱麻，难以调试和持久化；复杂的多智能体工作流一旦出错，就像掉进了黑盒，根本不知道问题出在哪一步。我们花费了大量时间不是在构建业务价值，而是在和框架的“不确定性”作斗争。

这就是我最初接触并决定深入研究Agently的原因。它不是一个为了追求最新潮概念而诞生的框架，它的设计哲学非常明确：为工程化、可交付的AI应用提供稳定、可观测、可测试的基础设施。简单来说，Agently瞄准的是从“原型验证”到“生产部署”之间那道令人头疼的鸿沟。它通过“契约优先”的输出控制、可序列化与恢复的工作流、分层级的配置管理以及全新的、可插拔的Action运行时（v4.1版本的核心），让开发者能够像构建传统软件一样，以工程化的思维来构建AI应用。

无论你是想开发一个需要精准解析用户意图的客服机器人，一个自动化处理文档并生成报告的数据流水线，还是一个需要协调多个专家模型完成复杂任务的分析系统，Agently提供了一套完整且严谨的工具链。它尤其适合那些对系统稳定性、可维护性和可观测性有要求的团队。接下来，我将结合自己从零开始搭建一个智能新闻摘要系统的实战经验，为你深度拆解Agently的核心能力与最佳实践。

2. 核心设计哲学：为什么是“工程优先”？

在深入代码之前，理解Agently的设计哲学至关重要。这决定了你能否真正用好它，而不是仅仅把它当作另一个“链式调用”的库。与市面上许多框架不同，Agently的出发点不是“如何让AI做更多事”，而是“如何让AI做的事更可靠、更可控”。

2.1 从“探索”到“交付”的范式转变

LangChain、CrewAI等框架极大地降低了AI应用的原型开发门槛，它们擅长快速组合各种组件，进行探索性实验。然而，当项目进入交付阶段，我们需要的是截然不同的特性：

• 确定性 vs. 灵活性：原型阶段我们追求灵活性，允许模型“自由发挥”；生产阶段我们需要确定性，确保每次调用返回的数据结构一致，否则下游的业务逻辑（如数据库写入、API响应）就会崩溃。
• 可观测性 vs. 黑盒：原型阶段我们可能只关心最终结果；生产阶段我们必须能追踪每个步骤的输入、输出、耗时和错误，以便于监控、调试和审计。
• 可测试性 vs. 一次性运行：原型代码往往难以进行单元测试；生产代码要求每个组件、每个工作流分支都能被独立测试，以保障代码质量。
• 状态管理 vs. 单次会话：简单的聊天或许不需要持久化状态，但一个跨越多次用户交互、甚至需要暂停等待人工审核的复杂业务流程，必须有能力保存和恢复执行现场。

Agently的每一层设计都围绕着这些生产环境的需求展开。它默认你关心输出格式的稳定性，所以引入了强制性的输出模式（Schema）校验。它假设你的工作流可能很复杂，所以提供了TriggerFlow这种可以暂停、恢复、序列化的工作流引擎。它明白你的配置会随着项目增长而变得复杂，所以内置了分层级的配置文件管理系统。

2.2 分层架构：清晰的关注点分离

Agently采用了一个清晰的四层架构模型，这不仅是技术上的分层，更是概念上的分离，使得每个部分都可以被独立理解、替换和测试。

1. 应用/业务逻辑层：这是你编写业务代码的地方，使用TriggerFlow进行流程编排，或者直接调用Agent。
2. 智能体层：每个Agent实例是核心执行单元，它封装了提示词管理、会话记忆、动作执行和自身配置。
3. 模型层：负责将Agent构造的请求转换为对LLM API（如OpenAI、DeepSeek、Ollama）的调用，并处理返回的流式或非流式响应。
4. 工作流编排层：即TriggerFlow，它独立于Agent，负责定义和执行包含条件分支、并行处理、事件驱动等复杂逻辑的业务流程。

这种架构带来的最大好处是可维护性。当需要更换模型供应商时，你只需修改模型层的配置，业务逻辑和Agent层代码几乎不用动。当需要调整一个复杂工作流的某个分支时，你可以单独测试那个分支对应的TriggerFlow chunk（一个普通的Python函数）。

实操心得：在项目初期，就按照这个四层模型来组织你的代码目录结构（例如agents/,flows/,config/,models/），会为后续的迭代和维护省下大量时间。Agently提供的agently-devtools init命令生成的脚手架就遵循了这个理念。

3. 实战核心：深度解析Agently四大核心能力

理论说再多，不如一行代码。让我们通过构建一个“智能新闻收集与摘要系统”的实战案例，来感受Agently的核心能力。这个系统的目标是：定期从多个RSS源抓取新闻，由AI分析并分类，最后生成一份结构化的每日简报。

3.1 契约优先的输出控制：告别JSON解析的噩梦

这是Agently让我眼前一亮的第一个功能。在传统方式中，我们让LLM输出JSON，然后祈祷它格式正确，再用json.loads()解析，并写一堆if-else来处理字段缺失。在Agently中，这个过程被彻底规范化。

from agently import Agently import os # 1. 全局配置：连接到DeepSeek模型 Agently.set_settings("OpenAICompatible", { "base_url": "https://api.deepseek.com/v1", "model": "deepseek-chat", "auth": os.getenv("DEEPSEEK_API_KEY"), # 从环境变量读取密钥 }) # 2. 创建一个新闻分析Agent news_analyzer = Agently.create_agent() news_analyzer.role("你是一个专业的新闻编辑，擅长对新闻内容进行精准分类和摘要。") # 3. 定义分析任务并指定严格的输出格式 def analyze_news_article(article_text: str): """分析单篇新闻文章""" result = ( news_analyzer .input(f"请分析以下新闻内容：\n\n{article_text}") .instruct("请严格遵循指定的JSON格式输出，不要添加任何额外解释。") .output({ "title": (str, "新闻标题的提炼，不超过20字"), "summary": (str, "新闻内容摘要，100字左右"), "category": (str, "分类，如：科技、财经、体育、娱乐、国际、国内"), "sentiment": (str, "情感倾向：positive/neutral/negative"), "keywords": [(str, "3-5个关键词")], "priority": (int, "重要性评分，1-5分，5分为最高"), }) .start(ensure_keys=["title", "summary", "category", "keywords[*]"]) ) return result # 使用示例 article = "OpenAI今日发布了新一代多模态模型...（此处为模拟新闻内容）" analysis = analyze_news_article(article) print(f"标题: {analysis['title']}") print(f"分类: {analysis['category']}") print(f"关键词: {', '.join(analysis['keywords'])}") # 输出保证是一个包含所有指定键的字典，且keywords列表中的每个元素都存在。

关键解析：

• .output()方法接受一个字典，定义了返回数据的模式。值是一个元组(类型, 描述)。描述会作为系统提示的一部分送给模型，指导其生成内容。
• .start(ensure_keys=...)参数是安全网。它告诉Agently：必须确保返回的字典中包含这些键（[*]表示列表中的每个元素），如果模型返回的JSON里缺失了，Agently会自动重试请求（可配置重试次数），直到获取到符合要求的数据。这从根本上解决了生产环境中因为模型“胡言乱语”导致的服务中断。

避坑指南：ensure_keys是保障下游业务逻辑稳定的关键。对于核心业务字段，务必加上。但要注意，这会增加API调用次数和耗时。对于非核心的辅助字段（如sentiment），可以不加入ensure_keys，并在代码中提供默认值处理逻辑，例如sentiment = result.get('sentiment', 'neutral')。

3.2 全新的Action运行时（v4.1）：统一且可扩展的动作执行

在v4.1之前，Agently的“工具”调用功能已经很强大了。v4.1的重写将其升级为一个三层插件化的Action运行时，概念上更清晰，扩展性也更强。这三层是：

1. ActionRuntime：决定“如何规划”一次动作调用（例如，ReAct模式还是直接函数调用）。
2. ActionFlow：决定“如何循环”执行动作（例如，顺序执行、带暂停恢复、并发控制）。
3. ActionExecutor：决定“如何执行”具体的动作（例如，调用本地Python函数、连接MCP服务器、在沙箱中运行代码）。

对于我们新闻系统，我们需要让AI能主动获取新闻。我们可以同时展示本地函数、MCP服务器和沙箱三种执行方式。

import requests import feedparser from agently import Agently # 假设我们已经有了一个配置好的Agent实例 `worker_agent` # --- 方式1: 注册本地函数作为Action --- @worker_agent.action_func def fetch_rss_feed(feed_url: str) -> list: """从指定的RSS源获取最新的新闻条目""" try: feed = feedparser.parse(feed_url) articles = [] for entry in feed.entries[:5]: # 取最新5条 articles.append({ "title": entry.title, "link": entry.link, "published": entry.get('published', ''), "summary": entry.get('summary', '')[:200] # 截断摘要 }) return articles except Exception as e: return [{"error": f"Failed to fetch RSS feed: {str(e)}"}] # --- 方式2: 连接MCP服务器（假设我们有一个管理内部知识库的MCP服务）--- # worker_agent.use_mcp("internal-knowledge", transport="stdio", command=["node", "knowledge-mcp-server.js"]) # --- 方式3: 使用Python沙箱执行动态代码（谨慎使用）--- worker_agent.use_sandbox("python") # 告诉Agent可以使用哪些动作 worker_agent.use_actions([fetch_rss_feed, "python"]) # 这里加入了python沙箱 # 现在，Agent在规划时，就知道它可以调用`fetch_rss_feed`函数和`python`沙箱了。 # 当我们提出复杂请求时，它可能会自主规划使用这些动作。 response = worker_agent.input( "先去获取Hacker News的RSS（https://hnrss.org/frontpage），然后挑出最热门的一条，用Python写一段代码分析一下它的标题情感。" ).get_response() # 查看完整的动作执行日志 action_logs = response.result.full_result_data.get("extra", {}).get("action_logs", []) for log in action_logs: print(f"[Action] {log.get('action_name')} | Input: {log.get('input')} | Output: {log.get('output')[:100]}...")

关键解析：

• @agent.action_func装饰器将普通Python函数转化为Agent可调用的“动作”。函数签名和文档字符串会被自动用于生成工具描述。
• agent.use_actions()显式声明该Agent实例可用的动作列表，实现了能力的按需加载和隔离。
• 动作日志是生产环境调试的利器。每个动作的调用参数、返回结果、耗时都被完整记录，你可以轻松地将这些日志发送到ELK、Datadog等观测平台。

注意事项：使用沙箱（尤其是Bash沙箱）存在安全风险，务必在受控环境（如容器内）运行，并对输入进行严格的过滤和限制。对于生产环境，更推荐使用MCP（Model Context Protocol）来安全地暴露内部能力。

3.3 TriggerFlow：超越简单链式调用的工作流引擎

这是Agently的“杀手级”特性。如果你以为它只是另一个LLMChain的替代品，那就错了。TriggerFlow是一个功能完备的工作流引擎，支持并发、事件驱动、人工干预和状态持久化。

让我们用TriggerFlow来编排整个新闻收集系统的核心流程。

from agently import TriggerFlow import asyncio # 创建TriggerFlow实例 flow = TriggerFlow() # 假设我们已经定义好了一些“块”（Chunk），也就是工作流中的步骤函数 async def fetch_all_feeds(data): """并发获取所有RSS源的数据""" rss_urls = data.get("rss_urls", []) tasks = [fetch_single_feed(url) for url in rss_urls] # fetch_single_feed是另一个异步函数 results = await asyncio.gather(*tasks, return_exceptions=True) # 过滤掉失败的请求 successful_feeds = [r for r in results if not isinstance(r, Exception)] return {"all_articles": [article for feed in successful_feeds for article in feed]} async def filter_and_deduplicate(data): """过滤和去重文章""" articles = data["all_articles"] # 简单的基于标题和链接的去重逻辑 seen = set() unique_articles = [] for article in articles: key = (article["title"], article["link"]) if key not in seen: seen.add(key) unique_articles.append(article) return {"unique_articles": unique_articles} async def analyze_articles_batch(data): """批量分析文章（使用之前定义的news_analyzer Agent）""" articles = data["unique_articles"] analyzed = [] for article in articles: # 这里调用我们3.1章节定义的analyze_news_article函数 analysis = await asyncio.to_thread(analyze_news_article, article["summary"]) analysis["original_link"] = article["link"] analyzed.append(analysis) return {"analyzed_articles": analyzed} async def generate_daily_briefing(data): """生成每日简报""" analyzed = data["analyzed_articles"] # 按分类和优先级排序 sorted_articles = sorted(analyzed, key=lambda x: (-x['priority'], x['category'])) # 调用另一个专门负责写作的Agent来生成简报文本 briefing_text = await writing_agent.generate_briefing(sorted_articles) return {"final_briefing": briefing_text, "raw_data": sorted_articles} # 定义工作流 ( flow.to(fetch_all_feeds) # 步骤1：抓取 .to(filter_and_deduplicate) # 步骤2：清洗 .to(analyze_articles_batch) # 步骤3：分析 .to(generate_daily_briefing) # 步骤4：生成报告 .end() ) # 执行工作流 async def main(): initial_data = {"rss_urls": [ "https://hnrss.org/frontpage", "https://rss.nytimes.com/services/xml/rss/nyt/Technology.xml", # ... 更多源 ]} result = await flow.start(initial_data) print("每日简报生成完成！") print(result["final_briefing"][:500]) # 打印前500字符 # asyncio.run(main())

高级特性实战：并发与持久化

上面的流程是顺序执行的。但抓取多个RSS源完全可以并行。TriggerFlow的for_each和batch让这变得很简单。

# 使用 for_each 实现并发抓取 async def fetch_single_feed(url): """抓取单个RSS源""" # ... 实现抓取逻辑 return articles # 在flow定义中 ( flow.for_each("${rss_urls}", concurrency=3) # 并发数为3 .to(fetch_single_feed) .collect("fetched_articles") # 将每个结果收集到一个列表中 .end() .to(filter_and_deduplicate, use_collected="fetched_articles") # 使用收集的结果 # ... 后续步骤 )

更强大的是持久化。想象一下，你的新闻分析流程在生成了简报草稿后，需要等待主编人工审核。这个流程可能被挂起数小时甚至数天。

# 开始执行，但不等待最终结果（比如在生成草稿后暂停） execution = flow.start_execution(initial_data, wait_for_result=False) # 将当前执行状态（包括所有中间数据）保存到文件或数据库 execution.save("news_briefing_checkpoint.json") # 此时流程可以暂停，服务器甚至可以重启。 # --- 几小时后，主编审核通过 --- # 从持久化状态恢复执行 restored_execution = flow.create_execution() restored_execution.load("news_briefing_checkpoint.json") # 发送一个“审核通过”的事件，触发后续流程（例如，发送邮件） restored_execution.emit("EditorApproved", {"approved": True, "comments": "Good to go."}) # 继续执行，获取最终结果 final_result = restored_execution.get_result(timeout=30)

关键解析：

• TriggerFlow的每个.to()节点都是一个普通的异步Python函数，这意味着它可以被单独进行单元测试。
• for_each和batch提供了原生的并发控制，无需手动管理asyncio.gather。
• 执行状态持久化是构建长周期、需人工干预业务流程的基石。它使得AI工作流不再是“一跑到底”的黑盒，而是可以暂停、恢复、与外部系统交互的可靠流程。

3.4 分层配置与项目管理：从脚本到工程

当你的AI应用从单个脚本成长为一个拥有多个Agent、多种环境（开发/测试/生产）、复杂提示词模板的项目时，配置管理就成了噩梦。Agently的项目级配置系统就是为了解决这个问题。

推荐的项目结构：

my_ai_news_system/ ├── .env # 环境变量，存放API密钥等敏感信息 ├── config/ │ ├── global.yaml # 全局默认配置（模型、超时等） │ ├── development.yaml # 开发环境覆盖配置 │ └── agents/ # 各Agent的专属配置 │ ├── analyzer.yaml │ └── writer.yaml ├── prompts/ # 提示词模板 │ ├── analyzer_role.yaml │ └── briefing_writer.yaml ├── agents/ # Agent定义 │ ├── news_analyzer.py │ └── briefing_writer.py ├── flows/ # TriggerFlow定义 │ └── daily_briefing_flow.py ├── actions/ # 自定义Action函数 │ └── rss_fetcher.py └── main.py # 应用入口

config/global.yaml示例：

OpenAICompatible: base_url: ${DEEPSEEK_API_BASE:-https://api.deepseek.com/v1} # 支持环境变量替换 model: deepseek-chat request_options: temperature: 0.7 max_tokens: 2000 timeout: 30 Agently: default_agent_settings: session: max_length: 4096 action: max_retries: 2

config/agents/analyzer.yaml示例：

# 继承全局配置，并覆盖特定设置 _extends: ../global.yaml OpenAICompatible: request_options: temperature: 0.2 # 分析任务需要更低的随机性 role: "你是一个专业的新闻编辑，擅长对新闻内容进行精准分类和摘要。" # 可以预加载一些提示词信息 info: style_guide: "摘要应客观、简洁，突出事实。"

在代码中使用：

from agently import Agently import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件 # 1. 加载全局配置 Agently.load_settings("yaml", "config/global.yaml", auto_load_env=True) # 2. 创建Agent并加载其专属配置 news_analyzer = Agently.create_agent() news_analyzer.load_settings("yaml", "config/agents/analyzer.yaml") # 3. 在代码中还可以动态覆盖（例如，根据请求参数调整） # news_analyzer.set_settings("OpenAICompatible.request_options.temperature", 0.1)

关键解析：

• 分层继承：Agent配置继承全局配置，请求级配置又可以覆盖Agent配置。这提供了极大的灵活性。
• 环境变量替换：使用${VAR_NAME}语法，可以将敏感信息（API Key、数据库地址）完全隔离在代码之外。
• 配置即代码：YAML/JSON/TOML格式的配置文件易于版本控制、评审和在不同环境间同步。

最佳实践：使用agently-devtools init my_project命令初始化你的项目，它会生成一个符合最佳实践的项目骨架，帮你省去搭建目录结构的麻烦。

4. 进阶技巧与避坑指南

在深度使用Agently构建了几个生产项目后，我积累了一些宝贵的经验和需要避开的“坑”。

4.1 Session会话管理的正确姿势

Session用于管理多轮对话的记忆。但直接无脑存储所有历史记录会导致token消耗剧增和上下文窗口溢出。

agent.activate_session("user-123") agent.set_settings("session.max_length", 8000) # 设置最大token数限制 # 关键：注册记忆处理策略 def smart_summarize_handler(session, messages): """当历史消息过长时，触发摘要处理""" # 例如，将最早的5轮对话总结成一条系统消息 old_messages = messages[:10] # 假设每条消息约消耗一定token summary = your_summarization_function(old_messages) # 用摘要替换旧消息 new_messages = [{"role": "system", "content": f"历史对话摘要：{summary}"}] + messages[10:] return new_messages # 注册摘要处理器 agent.activated_session.register_resize_handler("smart_summarize", smart_summarize_handler) # 注册分析处理器，决定何时触发摘要 def should_summarize(session, new_message): # 基于token数或轮数等逻辑判断 estimated_tokens = estimate_token_count(session.messages) return estimated_tokens > 6000 agent.activated_session.register_analysis_handler(should_summarize)

避坑指南：不要依赖模型的“长上下文”能力来无限制存储历史。主动的会话管理（摘要、关键信息提取）对于长期运行的对话机器人和成本控制至关重要。Agently的Session扩展提供了钩子（hook）让你实现这些策略。

4.2 结构化流式输出的妙用

当你构建需要实时反馈的UI应用（如打字机效果、渐进式内容展示）时，Agently的即时流式事件（Instant Events）功能非常强大。

response = ( agent .input("写一篇关于Agently框架的简短介绍，包含三个主要特点。") .output({ "title": (str, "文章标题"), "features": [(str, "特点描述")], # 列表也会流式返回 "conclusion": (str, "总结"), }) .get_response() ) # 处理流式事件 for event in response.get_generator(type="instant"): if event.path == "title": # event.delta 是标题字段新增的字符 ui.update_title_stream(event.delta) elif event.wildcard_path == "features[*]": # 列表的每个元素完成时，会触发一个 is_complete 事件 if event.is_complete: ui.append_feature(event.value) # event.value 是整个特点描述字符串 elif event.path == "conclusion" and event.is_complete: ui.mark_as_finished()

关键解析：type="instant"的事件生成器会在每个字段有内容更新时立即触发，而不是等整个JSON对象生成完毕。这允许你构建响应极其迅速的交互界面。

4.3 自定义扩展：按需改造框架

Agently的插件化架构意味着你几乎可以定制任何部分。例如，如果你需要对接一个非OpenAI兼容的私有模型API，你可以实现自己的ModelRequester。

from agently.types.plugins import ModelRequester, ModelRequest, ModelResponse class MyCustomModelRequester(ModelRequester): name = "my-custom-model" DEFAULT_SETTINGS = { "endpoint": "http://internal-llm:8080/v1/chat/completions", "api_key": None, } async def request(self, request: ModelRequest) -> ModelResponse: # 1. 将Agently的标准请求格式，转换成你内部模型的格式 internal_payload = self._convert_request(request) # 2. 发起HTTP请求 async with aiohttp.ClientSession() as session: async with session.post( self.settings["endpoint"], json=internal_payload, headers={"Authorization": f"Bearer {self.settings['api_key']}"} ) as resp: result = await resp.json() # 3. 将内部模型的响应，转换回Agently的标准响应格式 return self._convert_response(result) def _convert_request(self, request: ModelRequest) -> dict: # 实现转换逻辑... pass def _convert_response(self, raw_response: dict) -> ModelResponse: # 实现转换逻辑... pass # 注册你的自定义请求器 from agently import Agently Agently.plugin_manager.register("ModelRequester", MyCustomModelRequester) # 使用它 Agently.set_settings("MyCustomModel", { "endpoint": "...", "api_key": "...", }) agent = Agently.create_agent() agent.set_settings("model.request_plugin", "my-custom-model")

5. 常见问题排查与性能优化

在实际部署中，你可能会遇到以下问题。这里是我的排查清单和优化建议。

5.1 问题排查速查表

5.2 性能优化建议

1. 连接池与超时设置：在全局配置中为HTTP客户端（如aiohttp）配置连接池和合理的超时，避免因网络波动导致线程阻塞。

   Agently.set_settings("OpenAICompatible.request_options", { "timeout": 30.0, # 如果你的请求库支持，可以传递更多底层参数 })

2. 缓存策略：对于频繁且结果不变的Agent调用（例如，将固定文本翻译成另一种语言），可以考虑在Agent层之上添加一个缓存层（如functools.lru_cache或Redis缓存），直接返回缓存结果。

3. 批量处理：当有大量独立文本需要处理（如情感分析、分类）时，不要用for循环串行调用Agent。尽可能利用TriggerFlow的for_each进行并发处理，或者探索模型是否支持批量API（但需注意Agently当前输出模式对批量API的支持度）。

4. 监控与告警：务必接入监控。关键指标包括：

   • Token消耗：通过模型的响应元数据或自行估算。
   • 请求延迟(P50, P95, P99)：监控每个Agent调用或TriggerFlow chunk的耗时。
   • 错误率：特别是ensure_keys重试失败率和动作调用失败率。
   • 队列长度：如果使用了异步任务队列。

   可以将action_logs和tool_logs作为结构化日志输出，方便被日志收集系统抓取和分析。

经过几个月的实战，我的体会是，Agently更像是一个“AI应用的操作系统”，它提供了构建稳定、可维护的AI产品所需的基础设施和开发范式。它可能不像一些框架那样在原型阶段有最多的现成“零件”，但它提供的“工程工具箱”能确保你的作品在从实验室走向市场的道路上，不会因为底盘的松散而中途散架。如果你正在为一个即将上线的AI功能寻找可靠的技术栈，Agently值得你投入时间深入探索。它的学习曲线可能略陡，但换来的是对生产环境复杂性的掌控力，从长期来看，这笔投资是划算的。