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采用了一个清晰的四层架构模型这不仅是技术上的分层更是概念上的分离使得每个部分都可以被独立理解、替换和测试。应用/业务逻辑层这是你编写业务代码的地方使用TriggerFlow进行流程编排或者直接调用Agent。智能体层每个Agent实例是核心执行单元它封装了提示词管理、会话记忆、动作执行和自身配置。模型层负责将Agent构造的请求转换为对LLM API如OpenAI、DeepSeek、Ollama的调用并处理返回的流式或非流式响应。工作流编排层即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运行时概念上更清晰扩展性也更强。这三层是ActionRuntime决定“如何规划”一次动作调用例如ReAct模式还是直接函数调用。ActionFlow决定“如何循环”执行动作例如顺序执行、带暂停恢复、并发控制。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: fFailed to fetch RSS feed: {str(e)}}] # --- 方式2: 连接MCP服务器假设我们有一个管理内部知识库的MCP服务--- # worker_agent.use_mcp(internal-knowledge, transportstdio, 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的RSShttps://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沙箱存在安全风险务必在受控环境如容器内运行并对输入进行严格的过滤和限制。对于生产环境更推荐使用MCPModel 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_exceptionsTrue) # 过滤掉失败的请求 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, keylambda 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}, concurrency3) # 并发数为3 .to(fetch_single_feed) .collect(fetched_articles) # 将每个结果收集到一个列表中 .end() .to(filter_and_deduplicate, use_collectedfetched_articles) # 使用收集的结果 # ... 后续步骤 )更强大的是持久化。想象一下你的新闻分析流程在生成了简报草稿后需要等待主编人工审核。这个流程可能被挂起数小时甚至数天。# 开始执行但不等待最终结果比如在生成草稿后暂停 execution flow.start_execution(initial_data, wait_for_resultFalse) # 将当前执行状态包括所有中间数据保存到文件或数据库 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(timeout30)关键解析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: 2config/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_envTrue) # 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(typeinstant): 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()关键解析typeinstant的事件生成器会在每个字段有内容更新时立即触发而不是等整个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], jsoninternal_payload, headers{Authorization: fBearer {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 问题排查速查表问题现象可能原因排查步骤输出字段缺失即使使用了ensure_keys1. 模型能力不足始终无法生成该字段。2. 字段描述不清模型不理解。3. 重试次数用尽。1. 检查response.result.full_result_data中的原始响应和错误信息。2. 简化字段描述或将其拆分为更简单的子字段。3. 增加.start(ensure_keys..., max_retry5)中的重试次数。TriggerFlow 执行卡住或无反应1. 某个chunk函数是同步的阻塞了异步循环。2. 存在未处理的异常。3. 等待某个事件 (when)但事件从未被触发。1. 确保所有chunk函数都是async def或使用asyncio.to_thread包装同步IO操作。2. 在每个chunk内部添加try...except并打印日志。3. 检查事件发射逻辑确保条件匹配。使用DevTools观察执行状态。动作Action调用失败1. 函数签名或文档字符串不符合要求。2. MCP服务器未启动或连接失败。3. 沙箱执行超时或权限不足。1. 检查action_func装饰的函数确保参数有类型注解且有清晰的docstring。2. 检查MCP服务器的进程状态和日志。3. 检查沙箱配置如超时时间、资源限制。查看action_logs获取详细错误。内存消耗随时间增长1. Session历史未清理。2. TriggerFlow执行状态未及时释放。3. 大型对象如原始网页内容在流程中被传递和保留。1. 实现Session的摘要或滚动窗口策略。2. 对于一次性流程在执行完成后主动清理execution对象。3. 在流程中尽早将原始数据转换为精简的中间表示并丢弃原始数据。请求速度慢1. 网络延迟高。2. 模型响应慢。3. 复杂的输出模式导致模型生成慢。4. 未充分利用并发。1. 考虑使用离你更近的模型端点或部署本地模型如Ollama。2. 尝试更快的模型或调整max_tokens。3. 简化输出模式或将复杂任务拆分为多个连续请求。4. 对独立任务使用flow.batch()或flow.for_each(concurrencyN)。5.2 性能优化建议连接池与超时设置在全局配置中为HTTP客户端如aiohttp配置连接池和合理的超时避免因网络波动导致线程阻塞。Agently.set_settings(OpenAICompatible.request_options, { timeout: 30.0, # 如果你的请求库支持可以传递更多底层参数 })缓存策略对于频繁且结果不变的Agent调用例如将固定文本翻译成另一种语言可以考虑在Agent层之上添加一个缓存层如functools.lru_cache或Redis缓存直接返回缓存结果。批量处理当有大量独立文本需要处理如情感分析、分类时不要用for循环串行调用Agent。尽可能利用TriggerFlow的for_each进行并发处理或者探索模型是否支持批量API但需注意Agently当前输出模式对批量API的支持度。监控与告警务必接入监控。关键指标包括Token消耗通过模型的响应元数据或自行估算。请求延迟(P50, P95, P99)监控每个Agent调用或TriggerFlow chunk的耗时。错误率特别是ensure_keys重试失败率和动作调用失败率。队列长度如果使用了异步任务队列。可以将action_logs和tool_logs作为结构化日志输出方便被日志收集系统抓取和分析。经过几个月的实战我的体会是Agently更像是一个“AI应用的操作系统”它提供了构建稳定、可维护的AI产品所需的基础设施和开发范式。它可能不像一些框架那样在原型阶段有最多的现成“零件”但它提供的“工程工具箱”能确保你的作品在从实验室走向市场的道路上不会因为底盘的松散而中途散架。如果你正在为一个即将上线的AI功能寻找可靠的技术栈Agently值得你投入时间深入探索。它的学习曲线可能略陡但换来的是对生产环境复杂性的掌控力从长期来看这笔投资是划算的。