Agently框架：构建AI智能体的工程化实践与核心架构解析-平芜编程栈

1. 项目概述：从“智能体”到“智能体时代”的工程化跃迁

最近几年，AI领域最火的概念莫过于“智能体”了。无论是OpenAI的GPTs，还是各路创业公司推出的AI助手，都在强调一个核心：让AI不仅能回答问题，更能主动规划、调用工具、完成任务。听起来很酷，对吧？但当你真正上手，想把一个智能体想法落地时，往往会发现一堆工程上的“坑”：如何管理复杂的对话状态？如何优雅地调用外部API？如何设计一个可扩展的智能体架构？代码写着写着就成了一团乱麻。

这就是我最初接触AgentEra和Agently这个项目时的背景。它不是一个简单的聊天机器人框架，而是一个旨在定义“智能体时代”开发范式的开源项目。简单来说，Agently是一个用于快速构建、管理和部署AI智能体的开发框架与平台。它试图解决的核心痛点，正是我们这些一线开发者在构建复杂AI应用时遇到的：标准化、工程化和可复用性。

想象一下，你不再需要为每个智能体项目从头搭建一套状态管理、工具调用和流程控制的轮子。Agently提供了一套声明式的开发模式，让你可以像搭积木一样，用清晰的代码定义智能体的能力、工作流和交互逻辑。无论是构建一个能帮你分析数据的分析助手，还是一个能自动处理工单的客服机器人，Agently都试图提供一个“开箱即用”的底座。对于开发者而言，这意味着更高的开发效率和更低的维护成本；对于整个AI应用生态，这或许是在迈向真正的“智能体时代”过程中，不可或缺的一块基石。

2. 核心架构与设计哲学拆解

2.1 从“功能堆砌”到“角色定义”的范式转变

传统的AI应用开发，尤其是基于大语言模型的，很容易陷入“功能堆砌”的模式。我们写一个庞大的prompt，里面塞满了各种指令和示例，然后通过复杂的if-else逻辑来解析模型的输出，再调用相应的函数。这种方式在简单场景下尚可，一旦智能体需要处理多轮对话、记忆上下文、并行执行多个工具时，代码的复杂度会呈指数级上升，可读性和可维护性急剧下降。

Agently的设计哲学核心是“角色定义”和“声明式编程”。它鼓励开发者首先思考：这个智能体是谁？它应该扮演什么角色？然后，用框架提供的语法，去声明这个角色的属性、技能、知识库和行为约束，而不是去编写冗长的过程式代码。

举个例子，传统方式可能是：“写一个函数，先调用天气API，再解析结果，最后用特定格式回复用户”。而在Agently的范式里，你可能会这样定义：“我有一个‘天气助手’角色，它掌握‘查询天气’这个技能（该技能已封装好），当用户询问天气时，自动触发该技能，并按照预设的模板组织回复。” 框架负责中间的调度、状态管理和执行。这种转变，将开发者的注意力从“如何实现流程”转移到了“如何设计智能体本身”，更符合智能体作为“虚拟角色”的本质。

2.2 核心组件：Agent, Skill, Workflow 与 Runtime

要理解Agently，必须吃透它的几个核心抽象。这些抽象构成了整个框架的骨架。

1. Agent（智能体）这是最顶层的实体，代表一个具有特定身份、目标和能力的AI角色。每个Agent都拥有：

Profile（档案）：定义其基础身份，如名称、角色描述、通用指令。这相当于给大模型一个稳定的“人设”，避免它在对话中漂移。
Memory（记忆）：包括短期会话记忆和可持久化的长期记忆。Agently框架提供了记忆管理的接口，可以方便地实现上下文保持、历史总结等功能。
Skills（技能）：智能体所具备的可执行能力。一个技能可以是一个简单的函数，也可以是一个复杂的工作流。

2. Skill（技能）技能是智能体能力的原子化封装。一个“查询数据库”技能、“发送邮件”技能或“生成图表”技能，都被实现为一个独立的、可复用的模块。Skill的核心特点是：

声明式定义：通过装饰器或配置文件，声明技能的输入参数、输出格式、描述以及所需调用的工具（函数）。
自动工具调用：框架能自动将Skill的描述转化为大模型可理解的工具定义（遵循OpenAI的Function Calling或类似规范），并在对话中自动触发。
可组合性：简单的技能可以组合成复杂的技能。

3. Workflow（工作流）当单个技能无法完成任务时，就需要Workflow。它定义了多个技能或步骤的执行顺序和逻辑关系。Agently的工作流引擎允许你设计顺序、分支、循环等复杂逻辑，例如：“先执行技能A，如果结果满足条件X，则并行执行技能B和C，否则执行技能D，最后汇总所有结果执行技能E”。这为构建自动化智能体提供了强大的编排能力。

4. Runtime（运行时）这是框架的引擎，负责连接所有组件。它管理着会话状态、调度技能和工作流的执行、处理与大语言模型的通信（支持多种模型提供商，如OpenAI、Anthropic、国内主流平台等），并提供了统一的生命周期钩子（如on_message,on_skill_executed），方便开发者进行监控、日志记录和自定义扩展。

注意：理解这四个核心组件的关系至关重要。你可以把Agent看作一个“演员”，Skill是它的“演技招式”，Workflow是“剧本分镜”，而Runtime就是整个“剧场和导演系统”。开发者的工作就是选角、设计招式和编排剧本，框架负责让演出顺利进行。

3. 快速上手：构建你的第一个智能体

理论说了这么多，不如动手来感受一下。我们以构建一个“会议纪要生成助手”为例，快速走一遍Agently的开发流程。这个助手能接收一段会议录音文本，自动提取关键信息（议题、结论、待办事项），并生成结构清晰的纪要。

3.1 环境准备与安装

首先，确保你的Python环境在3.8以上。通过pip安装Agently框架非常简单：

pip install agently

这里有一个关键点：Agently框架本身是模型无关的，但它需要配合一个具体的LLM（大语言模型）来工作。因此，你通常还需要安装对应模型的SDK。例如，如果你使用OpenAI的模型：

pip install openai

然后，你需要设置你的API密钥。强烈建议使用环境变量来管理密钥，而不是硬编码在代码中，这是生产环境的基本安全要求。

# 在终端中设置（临时） export OPENAI_API_KEY='your-api-key-here' # 或者在代码中读取环境变量（推荐） import os from agently import AgentFactory factory = AgentFactory() factory.set_llm_name("openai") # 指定使用OpenAI factory.set_llm_auth("api_key", os.getenv("OPENAI_API_KEY")) factory.set_llm_model("gpt-4o") # 指定模型版本

3.2 定义智能体角色与技能

接下来，我们创建会议纪要助手。首先定义它的角色档案：

from agently import AgentFactory # 创建智能体工厂 factory = AgentFactory() factory.set_llm_name("openai") factory.set_llm_auth("api_key", os.getenv("OPENAI_API_KEY")) # 创建具体的智能体实例 meeting_agent = factory.create_agent() # 设置智能体角色 meeting_agent.set_role("你是一个专业的会议秘书，擅长从杂乱的口语化文本中提炼核心信息，并生成格式规范、语言精炼的会议纪要。") meeting_agent.set_goal("准确识别会议中的讨论议题、形成的结论或决定、以及产生的待办事项（Action Items）。")

现在，我们来为它赋予一个核心技能。在Agently中，定义技能有多种方式，最直观的是使用@skill装饰器：

from agently import skill @skill( name="generate_meeting_minutes", description="根据提供的会议文本，生成结构化的会议纪要。", inputs={ "meeting_text": {"type": "string", "description": "原始的会议对话或转录文本"}, "template": {"type": "string", "description": "期望的纪要模板，例如：'议题：\\n结论：\\n待办：'", "required": False} }, outputs={"type": "string", "description": "生成的格式化会议纪要"} ) def generate_minutes(meeting_text: str, template: str = None) -> str: # 这个函数体本身可以很简单，因为复杂的逻辑交给LLM和prompt工程 # 我们在这里主要构建给LLM的指令 if template is None: template = """请基于以下会议文本，生成会议纪要。 会议文本： {meeting_text} 请按以下格式组织内容： ## 会议纪要 ### 1. 核心议题 - [列出讨论的主要议题，每条简短明确] ### 2. 结论与决定 - [列出达成的共识、做出的决策，注明负责人（如可识别）] ### 3. 待办事项 (Action Items) - [列出具体的行动项，每条格式为：* [任务描述] (@负责人) - [截止时间，如可识别] ] 要求：内容准确、条理清晰、语言正式。""" # 将技能“注册”到智能体，并告诉智能体如何执行它。 # 实际上，技能的触发和执行是由Runtime通过LLM来协调的。 # 这里我们演示的是“指令型”技能，直接让LLM根据指令生成内容。 prompt = template.format(meeting_text=meeting_text) # 通过智能体的会话能力来执行这个“指令” response = meeting_agent.chat(prompt) return response

上面的代码展示了技能的定义，但更常见的模式是，技能内部会封装对真实工具（如数据库、API）的调用。对于我们这个例子，生成纪要完全由LLM完成，所以技能函数的核心是构造精准的prompt。

3.3 运行与测试

技能定义好后，我们就可以在对话中使用了。Agently框架会自动将技能描述注入到给LLM的系统指令中，并在对话中识别用户意图，触发相应的技能。

# 模拟用户输入 user_input = """ （以下是模拟会议文本） 小王：我们接下来讨论一下Q3的产品上线计划。目前进度怎么样？ 小李：后端开发基本完成了，但前端还有两个页面没搞定，主要是图表组件比较耗时。 老王：测试环境部署了吗？ 小李：部署了，但昨晚的自动化测试跑出了5个中等级别的Bug。 小王：好。那我们的目标是月底前上线。小李，前端最晚什么时候能好？ 小李：争取下周三。 老王：那测试这边，等前端联调完，我们还需要3个工作日做全量回归。 小王：行。那就这么定：小李下周三完成前端；老王，下周五下班前完成全量回归并出具报告。我这边同步准备发布公告和运营材料。有任何风险及时同步。 """ # 将技能“附加”到智能体（在实际框架中，可能有更优雅的注册方式） # 这里为了演示，我们直接调用之前定义的函数，模拟技能执行。 minutes = generate_minutes(user_input) print(minutes)

执行这段代码，你应该能得到一份结构化的会议纪要，大致如下：

## 会议纪要 ### 1. 核心议题 - Q3产品上线计划与当前进度评审 ### 2. 结论与决定 - 产品上线最终目标时间为本月底。 - 前端开发（小李负责）需于下周三完成。 - 全量回归测试（老王负责）于前端联调完成后启动，需3个工作日，目标下周五下班前完成并出具报告。 - 发布公告与运营材料由小王准备。 ### 3. 待办事项 (Action Items) - * 完成前端剩余两个页面，特别是图表组件 (@小李) - 下周三 - * 进行全量回归测试并出具测试报告 (@老王) - 下周五下班前 - * 准备产品发布公告和运营材料 (@小王) - 上线前

通过这个简单的例子，你可以感受到Agently如何将“会议纪要生成”这个任务，封装成一个明确的技能，并通过智能体角色来上下文化地执行。这比直接向ChatGPT发送一段零散的文本并祈祷它给出好结果，要可靠和可控得多。

4. 深入核心：技能编排与工作流引擎

单一技能能解决特定问题，但真实世界的任务往往是串联或并联的。Agently的Workflow（工作流）引擎，就是为了解决复杂任务编排而生的。它允许你以可视化或代码的方式，定义智能体的执行蓝图。

4.1 工作流设计模式

假设我们的“会议纪要助手”需要升级。现在，它不仅要生成纪要，还要自动将识别出的“待办事项”同步到项目管理工具（如Jira），并给相关责任人发送一封提醒邮件。这个任务涉及三个步骤，且有明确的顺序依赖：

步骤A：分析文本，生成结构化纪要（包含待办事项）。
步骤B：解析纪要中的待办事项，在Jira中创建对应的Task或Bug。
步骤C：根据创建的Jira Issue，向责任人发送邮件通知。

这是一个典型的顺序工作流。在Agently中，你可以这样定义（这里使用伪代码风格展示概念）：

from agently import WorkflowBuilder builder = WorkflowBuilder() # 定义工作流节点（每个节点对应一个Skill） extract_node = builder.add_skill_node( skill_name="extract_meeting_minutes", inputs={"meeting_text": "{{input.text}}"} # 引用工作流初始输入 ) create_jira_node = builder.add_skill_node( skill_name="create_jira_issues", inputs={"action_items": "{{nodes.extract_node.outputs.action_items}}"} # 引用上一个节点的输出 ) send_email_node = builder.add_skill_node( skill_name="send_notification_email", inputs={ "issues_created": "{{nodes.create_jira_node.outputs.issue_links}}", "assignees": "{{nodes.create_jira_node.outputs.assignees}}" } ) # 定义节点间的连接关系（顺序执行） builder.connect(extract_node, create_jira_node) builder.connect(create_jira_node, send_email_node) # 编译工作流 complex_meeting_workflow = builder.build()

这样，当你将一段会议文本丢给这个工作流时，它会自动按顺序执行“提取 -> 创建Jira -> 发送邮件”的全流程。工作流引擎会处理节点间的数据传递、错误处理和状态持久化。

4.2 错误处理与条件分支

更复杂的工作流可能需要条件判断。例如，如果会议纪要中没有识别出待办事项，则跳过创建Jira和发送邮件的步骤。Agently的工作流支持条件节点。

# ... 接上文定义 extract_node 之后 ... # 添加一个条件判断节点 check_items_node = builder.add_condition_node( condition_expression="{{nodes.extract_node.outputs.has_action_items}}", # 如果条件为真 (has_action_items == True)，执行 then_branch then_branch=[create_jira_node, send_email_node], # 如果条件为假，执行 else_branch，这里可以是一个空列表或发送“无待办”通知的技能 else_branch=[send_no_action_email_node] ) builder.connect(extract_node, check_items_node) # then_branch 和 else_branch 内部的连接在定义时已隐含

通过工作流引擎，你可以将复杂的业务逻辑可视化、模块化，使智能体的行为不再是“黑盒”，而是一个可审计、可调试的自动化流程。这对于企业级应用的可靠性和可维护性至关重要。

5. 生产环境部署与性能考量

当你的智能体在本地运行良好，准备部署到生产环境服务真实用户时，会面临一系列新的挑战。Agently作为一个框架，提供了基础构建块，但生产化部署需要你额外考虑很多方面。

5.1 状态管理与持久化

在Web服务中，每个用户会话通常是独立且无状态的。但智能体对话本质上是有状态的，它需要记住之前的对话历史。Agently框架中的Agent实例通常包含了会话状态（Memory）。在生产环境中，你不能简单地将Agent实例存在内存里，因为服务重启或扩缩容会导致状态丢失。

解决方案是实现一个外部的状态存储后端。你需要：

自定义Memory存储：继承Agently的Memory基类，实现基于数据库（如Redis、PostgreSQL）或分布式缓存（如Memcached）的save和load方法。
会话标识：为每个用户或每个对话线程创建一个唯一的session_id。
序列化与反序列化：将Agent的Memory对象序列化（如JSON）后存入数据库，下次请求时根据session_id加载并还原Agent状态。

# 伪代码示例：自定义Redis Memory存储 import json import redis from agently.memory import BaseMemory class RedisMemory(BaseMemory): def __init__(self, session_id, redis_client): self.session_id = session_id self.redis = redis_client self.key = f"agent_memory:{session_id}" def save(self, memory_data: dict): """将记忆数据保存到Redis""" self.redis.setex(self.key, timeout=3600*24*7, value=json.dumps(memory_data)) # 设置7天过期 def load(self) -> dict: """从Redis加载记忆数据""" data = self.redis.get(self.key) return json.loads(data) if data else {}

在你的Web服务（如FastAPI）中，每个请求的处理流程变为：

@app.post("/chat") async def chat_endpoint(request: ChatRequest): session_id = request.session_id # 1. 从Redis加载或创建新的Memory memory_backend = RedisMemory(session_id, redis_client) memory_data = memory_backend.load() # 2. 创建或复用Agent，并注入记忆 agent = agent_factory.create_agent() agent.inject_memory(memory_data) # 3. 处理用户消息 response = await agent.chat(request.message) # 4. 保存更新后的记忆 updated_memory = agent.export_memory() memory_backend.save(updated_memory) return {"response": response}

5.2 异步处理与流式响应

LLM的生成速度可能较慢，尤其是在处理长文本或复杂思考时。让用户前端一直等待一个HTTP请求返回是不现实的。

异步处理：对于耗时较长的任务（如生成长篇报告），应该采用异步任务队列（如Celery + Redis/RabbitMQ）。API接口立即返回一个task_id，前端通过轮询或WebSocket来获取任务状态和结果。
流式响应：对于对话本身，支持Server-Sent Events (SSE)或WebSocket进行流式输出至关重要。这能让用户看到模型逐字生成的过程，体验远优于等待整个回复完成。你需要确保Agently框架底层与LLM API的交互支持流式输出（大多数现代SDK都支持），并将这些数据块实时推送给前端。

5.3 监控、日志与成本控制

在生产环境中，你必须知道你的智能体在做什么、表现如何、花了多少钱。

结构化日志：记录每一次用户交互、技能调用、工作流执行、LLM请求和响应。日志应包含session_id,skill_name,input_params,output,model_used,token_usage,latency等关键字段。这有助于调试问题、分析用户行为和分析模型性能。
性能监控：监控平均响应时间、错误率、令牌消耗速率等指标。设置警报，当延迟过高或错误激增时及时通知。
成本控制：LLM API调用是按Token计费的。必须实施配额和限流策略。例如，为每个用户设置每日/每月的Token消耗上限，或在技能调用前估算本次请求的Token数（通过计算提示词长度），如果超过阈值则拒绝或降级处理（如换用更便宜的模型）。Agently框架层面可以集成这些钩子，在调用LLM前后进行拦截和计算。

6. 避坑指南与最佳实践

在近一年的Agently项目实践和社区交流中，我积累了一些宝贵的经验教训，这里分享几个最常见的“坑”和应对策略。

6.1 技能设计的“单一职责”与“边界清晰”

坑点：设计一个“超级技能”，试图在一个函数里做太多事情，比如“分析数据并生成报告并发送邮件”。这会导致技能逻辑复杂、难以测试、且复用性差。

最佳实践：遵循单一职责原则。将“分析数据”、“生成报告”、“发送邮件”拆分成三个独立的技能。然后通过工作流将它们组合起来。这样做的好处是：

每个技能都可以独立开发、测试和复用。
当“发送邮件”的逻辑需要从SMTP改为调用企业微信API时，你只需修改这一个技能，不影响其他部分。
工作流可以灵活编排，例如，你可以轻松创建一个只“分析数据并生成报告”但不发送邮件的新流程。

6.2 Prompt工程是技能的核心，而非附属

坑点：认为技能就是写Python函数，Prompt随便写写就行。结果技能表现不稳定，时好时坏。

最佳实践：将技能视为“Prompt + 代码”的复合体。为每个技能精心设计它的Prompt模板，这包括：

清晰的指令：告诉模型要做什么，步骤是什么。
严格的输出格式约束：使用JSON Schema、XML标签或明确的标记（如## 标题）来规范输出，这能极大简化后续的代码解析。
高质量的示例：在Prompt中提供1-2个少样本示例（Few-shot Examples），能显著提升模型在复杂任务上的表现。
将Prompt模板外部化：不要将Prompt字符串硬编码在Python代码里。将它们放在配置文件（如YAML）、数据库或单独的文本文件中。这方便非开发人员（如产品经理）参与优化Prompt，也便于进行A/B测试。

# skills/generate_minutes.yaml name: generate_meeting_minutes description: 生成结构化会议纪要。 prompt_template: | 你是一个专业的会议秘书。请基于以下会议文本，生成会议纪要。 会议文本： {meeting_text} 请严格按以下JSON格式输出： {{ "topics": ["议题1", "议题2", ...], "decisions": ["决定1", "决定2", ...], "action_items": [ {{"task": "任务描述", "assignee": "负责人", "due_date": "截止时间"}}, ... ] }}

然后在技能代码中加载这个模板并填充变量。这样，Prompt的迭代就完全独立于业务逻辑代码了。

6.3 管理智能体的“认知负荷”与上下文长度

坑点：无节制地将所有历史对话、知识库文档都塞进每次请求的上下文（Prompt）中，导致Token消耗巨大、响应变慢、成本飙升，甚至可能因为超过模型上下文窗口而丢失关键信息。

最佳实践：实施积极的上下文管理策略。

记忆总结：当对话轮数超过一定阈值（如10轮）时，触发一个“总结”技能。让模型将之前的对话历史总结成一段精炼的要点，然后用这个总结替换掉冗长的原始历史，作为新的“长期记忆”放入后续对话中。这能有效压缩上下文。
选择性回忆：不要每次都传递全部知识库。实现一个“检索”技能，当用户提问时，先用嵌入模型（Embedding）将问题向量化，然后从向量数据库中检索最相关的几个知识片段，只将这些片段作为上下文提供给LLM。这就是检索增强生成（RAG）的核心思想，Agently可以很好地与向量数据库（如Chroma, Weaviate）集成来实现它。
设定上下文窗口预算：为每个会话设定一个Token上限。在每次准备构造Prompt时，计算当前累积的Token数，如果接近上限，则优先丢弃最早、最不重要的消息，或触发记忆总结。

6.4 测试策略：从单元测试到“对话仿真”

测试AI智能体比测试传统软件更复杂，因为它的输出具有非确定性。

技能单元测试：对于封装了确定性逻辑的技能（如调用某个API、处理特定格式数据），编写标准的单元测试。
技能集成测试：测试技能与LLM的集成。使用固定的输入和固定的模型（如gpt-3.5-turbo），断言其输出结构符合预期（例如，是否能解析出JSON），并对关键内容进行模糊匹配（如是否包含某个关键词）。避免断言完全相同的字符串。
工作流测试：模拟工作流各个节点的输入输出，测试流程是否能正确执行，分支逻辑是否正确。
端到端“对话仿真”测试：这是最接近真实场景的测试。编写一系列模拟用户对话的脚本，与你的智能体进行多轮交互，检查最终的任务完成情况。可以计算任务完成率、关键信息抽取准确率等指标。定期运行这些仿真测试，可以监控智能体性能是否发生退化。

构建一个稳定、高效、易维护的智能体系统，Agently框架提供了优秀的起点和范式。但它不是银弹，成功的关键依然在于开发者对业务逻辑的深刻理解、严谨的软件工程实践，以及对大语言模型特性的熟练把握。从定义一个清晰的角色开始，设计好原子化的技能，用工作流编织复杂的业务逻辑，最后为生产环境做好状态、性能和监控的保障——沿着这条路，你构建的将不再是一个脆弱的“聊天demo”，而是一个真正能创造价值的AI智能体。