1. 项目概述:AGI时代的新基建,从“智能体”到“智能体工厂”
最近几年,AI领域最火的概念莫过于“智能体”了。从AutoGPT的横空出世,到各种基于大语言模型的自动化工具层出不穷,大家似乎都看到了一个未来:让AI不仅能回答问题,更能主动规划、执行任务,成为我们数字世界里的“全能助手”。但当你真正上手去构建一个智能体时,会发现事情远没有想象中那么简单。如何让多个智能体协同工作?如何管理它们的状态和记忆?如何设计一个稳定、可扩展的执行流程?这些问题,往往会让开发者从“造梦者”变成“救火队员”。
这就是我今天想和大家深入聊聊的agno。它不是一个单一的智能体应用,而是一个旨在构建“智能体工厂”的开源框架。你可以把它理解为一套为生产级AI智能体应用量身定制的“操作系统”或“脚手架”。它的核心目标,是让开发者能够像搭积木一样,快速、可靠地组装出复杂的多智能体系统,而无需从零开始处理那些繁琐的底层通信、状态管理和错误处理逻辑。
简单来说,agno想解决的是智能体开发的“工程化”难题。它提供了从智能体定义、工作流编排、工具集成到持久化存储的一整套解决方案。无论你是想构建一个自动化的客服系统、一个复杂的代码审查助手,还是一个能够自主进行市场调研的分析平台,agno都试图为你提供一个坚实、灵活的基础。接下来,我将结合我自己的实践,拆解agno的核心设计、关键特性以及在实际落地中会遇到的那些“坑”。
2. 核心架构与设计哲学:为什么是“工厂”而非“单兵”
要理解agno的价值,首先要跳出“单个智能体”的视角。传统的智能体开发,往往是围绕一个LLM(大语言模型)调用,辅以一些工具函数,形成一个闭环。这种模式在简单任务上可行,但一旦任务变得复杂、需要多步骤推理或多人(多智能体)协作时,代码就会迅速变得臃肿且难以维护。
2.1 模块化与可组合性:智能体即组件
agno最核心的设计思想是模块化。它将一个智能体应用拆解为几个清晰的核心概念:
- Agent(智能体):这是执行任务的基本单元。每个Agent都有明确的职责,比如“代码分析员”、“文档撰写者”、“决策仲裁者”。在agno中,Agent不仅仅是一个Prompt模板加上LLM调用,它更是一个包含了记忆、工具、内部状态和通信能力的完整实体。
- Workflow(工作流):这是定义多个Agent如何协作的蓝图。工作流描述了任务的执行顺序、条件分支、循环以及Agent之间的数据传递。你可以把它想象成一个可视化的流程图,agno负责将这个流程图翻译成可靠的执行代码。
- Tool(工具):这是Agent与外部世界交互的“手”和“眼”。搜索网络、读写数据库、调用API、执行命令行,这些都可以封装成Tool。agno提供了一套标准化的方式来定义、注册和使用Tool,确保了它们在不同Agent间的可复用性。
- Memory(记忆):这是Agent的“经验库”。短期记忆(对话历史)、长期记忆(向量数据库存储的知识)、甚至是对自身执行过程的元记忆(比如“我上次在这个步骤失败了”),都可以通过agno的Memory模块来管理。这对于实现连贯的、有上下文的交互至关重要。
这种设计带来的最大好处是可组合性。你可以像搭乐高一样,将不同的Agent、预定义的工作流和各类Tool进行组合,快速构建出全新的应用。例如,一个“市场报告生成”应用,可以由“数据收集Agent”、“数据分析Agent”和“报告撰写Agent”通过一个线性工作流串联而成。如果需要增加一个“报告润色Agent”,只需在工作流中插入一个新节点即可,无需重写整个系统。
2.2 状态管理与持久化:让智能体“记住”自己是谁
智能体不是无状态的函数调用。一个复杂的任务可能被中断,需要恢复;智能体之间需要共享上下文;用户可能希望查询历史对话。agno内置了强大的状态管理机制。
每个工作流的执行实例、每个Agent的内部状态,都可以被持久化到数据库中(默认支持SQLite,可扩展至PostgreSQL等)。这意味着:
- 可恢复性:系统崩溃或任务被意外终止后,可以从断点继续执行。
- 可观测性:你可以随时查看某个任务执行到了哪一步,每个Agent输出了什么,便于调试和审计。
- 历史追溯:完整保留用户与智能体系统的交互历史,为后续的分析和模型优化提供数据。
实操心得:在早期测试中,我们忽略了状态持久化,导致一些长时间运行的任务一旦出错就要全部重来,用户体验极差。接入agno的状态管理后,我们实现了任务的“断点续传”,可靠性大幅提升。这里的关键是,要为工作流和Agent设计好有意义的、可序列化的状态字段,避免在状态中存储过大或不可序列化的对象(如数据库连接)。
2.3 异步与并发执行:榨干硬件性能
真实的智能体应用往往是I/O密集型的(等待LLM API返回、调用网络工具)。agno基于异步编程模型(如Python的asyncio)构建,原生支持Agent的并发执行。
在工作流中,如果两个Agent之间没有数据依赖关系,agno可以自动让它们并行运行,而不是傻傻地排队。例如,在调研一个公司时,“财务数据收集Agent”和“新闻舆情收集Agent”完全可以同时工作,最后将结果汇总给“分析报告生成Agent”。这种并发能力能显著缩短复杂工作流的整体执行时间。
3. 从零开始:搭建你的第一个agno智能体工作流
理论说了这么多,我们来点实际的。我将带你一步步搭建一个简单的“技术博客灵感生成器”智能体系统。这个系统会接受一个宽泛的主题(如“云原生”),然后通过两个协作的Agent,最终生成几个具体的博客标题和大纲。
3.1 环境准备与基础配置
首先,确保你的Python环境在3.9以上。然后安装agno:
pip install agnoagno默认使用OpenAI的GPT模型,所以你需要设置好你的API密钥。一种推荐的方式是通过环境变量:
export OPENAI_API_KEY='your-api-key-here'在代码中,我们可以这样初始化一个基础的“平台”对象,它是所有组件的容器:
from agno import Agent, Platform # 初始化平台 platform = Platform( name="BlogIdeaFactory", # 可以在这里配置默认的LLM、嵌入模型、数据库连接等 # 例如:llm="gpt-4-turbo-preview", database="sqlite:///blog_agent.db" )3.2 定义你的第一个Agent:创意发散者
我们来创建第一个Agent,它的职责是进行头脑风暴,根据宽泛主题生成一系列相关的子主题。
from agno import Agent class BrainstormAgent(Agent): def __init__(self): super().__init__( name="Brainstormer", role="创意发散专家", instructions=""" 你是一个科技领域的创意专家。用户会给你一个宽泛的技术主题(例如:云原生、机器学习、前端框架)。 你的任务是深入分析这个主题,并生成5个最具潜力的、具体的博客文章子方向。 每个子方向应该: 1. 足够具体,能直接作为博客的切入点。 2. 具有争议性或前沿性,能吸引读者。 3. 附带一句简要的说明,解释为什么这个方向值得写。 请以清晰的列表格式输出。 """, # 可以在此Agent级别覆盖平台的默认LLM配置 # model="gpt-4", ) # Agent的核心逻辑定义在 `run` 方法中 async def run(self, input_topic: str): # 构建给LLM的提示 prompt = f"请针对以下技术主题进行创意发散:{input_topic}" # 调用LLM response = await self.llm.generate(prompt) return response.content这里有几个关键点:
instructions是这个Agent的“角色设定”和“工作手册”,写得越清晰具体,Agent的表现就越稳定。run方法是Agent的入口。我们在这里处理输入,调用LLM,并返回结果。async/await语法表明这是异步的。- 我们还没有为这个Agent添加任何
Tool,它目前纯粹依靠LLM的知识进行创作。
3.3 定义第二个Agent:大纲架构师
第二个Agent负责将具体的子方向深化为完整的博客大纲。
class OutlineAgent(Agent): def __init__(self): super().__init__( name="Outliner", role="博客大纲架构师", instructions=""" 你是一位经验丰富的技术博客编辑。你将收到一个具体的博客文章方向。 你的任务是为这个方向创作一个详细、结构化的博客大纲。 大纲必须包含: 1. 一个吸引人的标题。 2. 引言部分的核心论点。 3. 至少3个主要章节,每个章节下包含2-4个子要点。 4. 总结与展望。 请确保大纲逻辑连贯,内容充实,能直接用于指导写作。 """, ) async def run(self, blog_idea: str): prompt = f”请为以下博客创意撰写详细大纲:{blog_idea}” response = await self.llm.generate(prompt) return response.content3.4 创建工作流:让Agent协同起来
现在,我们有了两个Agent,但它们是孤立的。我们需要一个Workflow来定义它们的协作关系:先由Brainstormer生成5个点子,然后由Outliner为每一个点子生成大纲。
from agno import Workflow, WorkflowStep class BlogIdeaWorkflow(Workflow): def __init__(self): super().__init__(name="generate_blog_outlines") self.brainstorm_agent = BrainstormAgent() self.outline_agent = OutlineAgent() async def run(self, initial_topic: str): # 步骤1:执行头脑风暴 self.logger.info(f“开始为主题‘{initial_topic}’进行头脑风暴...”) ideas_result = await self.brainstorm_agent.run(initial_topic) # 假设结果是一个由换行符分隔的列表,我们进行解析 ideas_list = [idea.strip() for idea in ideas_result.split('\n') if idea.strip()] final_output = [] # 步骤2:为每个想法创建大纲(并发执行) self.logger.info(f“将为 {len(ideas_list)} 个想法生成大纲...”) for idea in ideas_list: outline = await self.outline_agent.run(idea) final_output.append({ "idea": idea, "outline": outline }) return final_output这个工作流是顺序+循环的。在真实的agno工作流定义中,你可能会使用更声明式、可视化的方式来描述步骤间的依赖,但上面的代码展示了最核心的逻辑。agno的高级API允许你以更简洁的方式定义这种并行化:
# 伪代码,展示agno更高级的并发工作流定义理念 # workflow.parallel( # [self.outline_agent.run(idea) for idea in ideas_list] # )3.5 运行与测试
最后,让我们把这个系统跑起来:
import asyncio async def main(): platform = Platform(name="BlogDemo") workflow = BlogIdeaWorkflow() # 将工作流注册到平台(便于管理和持久化) platform.register_workflow(workflow) # 执行工作流 topic = “Serverless架构的优缺点” results = await workflow.run(topic) # 输出结果 for i, item in enumerate(results, 1): print(f“\n【创意 {i}】{item['idea']}”) print(f“【大纲】\n{item['outline']}”) print(“-” * 50) if __name__ == “__main__”: asyncio.run(main())运行这段代码,你应该能看到针对“Serverless架构的优缺点”这个主题,系统自动生成了5个具体的博客方向和对应的大纲。这只是一个起点,你可以在此基础上增加更多的Agent(比如“标题优化Agent”、“关键词SEO分析Agent”),或者引入真实的Tool(如从维基百科获取背景信息、从GitHub趋势榜获取灵感)。
4. 进阶实战:为智能体注入“感知”与“行动”能力
纯粹的LLM调用只是“思考”,真正的智能体必须能“感知”和“行动”。这就是Tool的用武之地。让我们升级之前的博客灵感生成器,让它能自动从网络上获取最新趋势来激发创意。
4.1 创建自定义Tool:网络搜索
假设我们想使用DuckDuckGo的搜索API(或者Serper API、Google Custom Search JSON API)来获取信息。首先,我们需要创建一个搜索Tool。
from agno import Tool import aiohttp import json class WebSearchTool(Tool): name = “web_search” description = “在互联网上搜索给定查询的最新信息。适用于获取技术趋势、新闻和事实。” def __init__(self, api_key: str = None): super().__init__() # 这里以Serper API为例,你需要注册并获取API Key self.api_key = api_key or os.getenv(“SERPER_API_KEY”) self.base_url = “https://google.serper.dev/search” async def run(self, query: str, num_results: int = 5): “””执行搜索并返回格式化结果。””” headers = { ‘X-API-KEY’: self.api_key, ‘Content-Type’: ‘application/json’ } payload = json.dumps({“q”: query, “num”: num_results}) async with aiohttp.ClientSession() as session: async with session.post(self.base_url, headers=headers, data=payload) as response: if response.status == 200: data = await response.json() # 简化处理,提取有机搜索结果 organic_results = data.get(‘organic’, []) summaries = [] for result in organic_results[:num_results]: title = result.get(‘title’, ‘’) snippet = result.get(‘snippet’, ‘’) link = result.get(‘link’, ‘’) summaries.append(f“标题:{title}\n摘要:{snippet}\n链接:{link}\n”) return “\n---\n”.join(summaries) else: return f“搜索失败,状态码:{response.status}”4.2 将Tool装配给Agent
现在,我们修改BrainstormAgent,让它拥有搜索能力。我们需要在Agent初始化时注册这个Tool,并在instructions中明确告知Agent可以使用它。
class EnhancedBrainstormAgent(Agent): def __init__(self, search_tool: WebSearchTool): super().__init__( name=“EnhancedBrainstormer”, role=“联网创意发散专家”, instructions=“”” 你是一个科技领域的创意专家。用户会给你一个宽泛的技术主题。 你的任务是生成5个最具潜力的、具体的博客文章子方向。 为了提高创意的时效性和相关性,**你可以并且应该**使用`web_search`工具来搜索该主题的最新趋势、争议和进展。 请先进行一轮搜索,根据搜索结果来激发和佐证你的创意。 每个创意输出格式:`创意描述 | 搜索依据摘要` “””, tools=[search_tool] # 关键:将工具实例传给Agent ) async def run(self, input_topic: str): # Agent在思考时,会根据instructions自动判断何时调用Tool。 # 我们需要构建一个包含“系统指令”和“用户输入”的完整对话。 messages = [ {“role”: “system”, “content”: self.instructions}, {“role”: “user”, “content”: f“请为‘{input_topic}’这个主题生成博客创意。”} ] # agno的Agent.run方法内部会处理与LLM的交互以及Tool的调用。 # 这里我们简化演示,实际中agno提供了更高级的会话执行引擎。 final_response = await self.execute_conversation(messages) return final_response注意事项:将Tool赋予Agent后,最大的挑战在于Prompt工程。你的
instructions必须清晰、明确地指导Agent何时以及如何使用工具。过于简略的指令可能导致Agent滥用工具(每个问题都搜索,拖慢速度且浪费资源)或完全忽略工具。通常需要经过几次迭代测试,才能找到最佳的指令描述。
4.3 设计更复杂的工作流:条件判断与循环
现在我们的系统更强大了,但工作流还是线性的。让我们设计一个更智能的流程:如果Brainstormer生成的某个创意在搜索后发现信息过少或质量不高,则自动过滤掉它。
这需要在工作流中引入条件判断。在agno中,你可以通过if/else逻辑或者专门的路由Agent来实现。
class QualityFilterAgent(Agent): def __init__(self): super().__init__( name=“QualityFilter”, role=“创意质量过滤器”, instructions=“”” 你评估一个博客创意的可行性。你会看到一个创意和其相关的搜索依据。 请从以下维度评估: 1. 信息充足度:搜索结果显示有足够多的讨论和材料吗?(满分5分) 2. 新颖性:这个角度是否新颖,还是老生常谈?(满分5分) 3. 受众吸引力:预计对开发者受众的吸引力如何?(满分5分) 如果总分低于10分,则认为该创意质量不佳,予以过滤。 请只输出你的评估总分和一个布尔值(True/False),格式为“总分|是否通过”。 例如:“12|True” 或 “8|False”。 “””, ) async def run(self, idea_with_evidence: str): prompt = f“请评估以下创意:\n{idea_with_evidence}” response = await self.llm.generate(prompt) return response.content.strip() class SmartBlogWorkflow(Workflow): def __init__(self): super().__init__(name=“smart_blog_workflow”) self.brainstormer = EnhancedBrainstormAgent(WebSearchTool()) self.filter = QualityFilterAgent() self.outliner = OutlineAgent() async def run(self, initial_topic: str): # 1. 生成带搜索依据的创意列表 raw_ideas_str = await self.brainstormer.run(initial_topic) # 解析字符串,假设每个创意一行 raw_ideas = [line for line in raw_ideas_str.split(‘\n’) if ‘|’ in line] qualified_ideas = [] # 2. 对每个创意进行质量过滤 for idea in raw_ideas: judgement = await self.filter.run(idea) try: score_str, pass_str = judgement.split(‘|’) if pass_str.strip().lower() == ‘true’: qualified_ideas.append(idea.split(‘|’)[0].strip()) # 只保留创意描述部分 except: self.logger.warning(f“无法解析过滤结果:{judgement},默认保留创意。”) qualified_ideas.append(idea.split(‘|’)[0].strip()) self.logger.info(f“原始创意{len(raw_ideas)}个,通过过滤{len(qualified_ideas)}个。”) # 3. 仅为通过的创意生成大纲 results = [] for qualified_idea in qualified_ideas: outline = await self.outliner.run(qualified_idea) results.append({“idea”: qualified_idea, “outline”: outline}) return results这个工作流展示了多智能体协作中的质量控制环节。在实际应用中,你还可以加入循环,让Brainstormer针对被过滤的创意方向进行二次发散,或者引入人工审核节点。
5. 部署、监控与性能调优:让智能体系统稳定运行
开发完成只是第一步,让系统在生产环境中稳定、高效地运行是更大的挑战。agno提供了一些内置特性和最佳实践来帮助我们。
5.1 持久化与状态恢复
在生产中,必须启用工作流的状态持久化。这通常在初始化Platform时配置。
from agno import Platform from agno.storage import SQLiteStorage platform = Platform( name=“ProductionBlogFactory”, storage=SQLiteStorage(“blog_production.db”), # 使用SQLite存储状态 # llm=“gpt-4”, # embedding_model=“text-embedding-3-small”, # 如果用到向量记忆 ) # 当你通过platform.run_workflow()执行工作流时,其状态会自动保存。 workflow_instance_id = await platform.run_workflow(“smart_blog_workflow”, input_data={“initial_topic”: “AI Agent”}) # 之后,你可以通过ID查询、恢复或重新执行这个实例 status = await platform.get_workflow_status(workflow_instance_id)5.2 日志、监控与可观测性
清晰的日志是调试和监控的命脉。agno集成了结构化的日志系统。
import logging # 设置agno的日志级别 logging.getLogger(“agno”).setLevel(logging.INFO) # 在你的Agent或Workflow中,可以使用self.logger class MyAgent(Agent): async def run(self, input): self.logger.info(f“开始处理输入:{input[:50]}...”) # ... 处理逻辑 self.logger.info(“处理完成。”)对于生产系统,你应该将日志收集到像ELK Stack、Loki或Datadog这样的集中式监控平台。关键指标包括:
- 工作流执行时长:识别性能瓶颈。
- Agent调用成功率/失败率:特别是LLM API和Tool调用的稳定性。
- Token消耗:监控成本。
- 队列长度:如果使用了异步任务队列。
5.3 性能调优与成本控制
智能体应用的成本和性能主要受LLM API调用影响。以下是一些实战技巧:
缓存LLM响应:对于内容相对稳定、重复性高的查询(例如,“解释什么是REST API”),可以使用agno的缓存中间件或外部缓存(如Redis)来存储LLM响应,避免重复调用。
from agno.middleware import CacheMiddleware agent = Agent( name=“CachedAgent”, instructions=“...”, middlewares=[CacheMiddleware(ttl=3600)] # 缓存1小时 )设置超时与重试:网络和API调用可能失败。为LLM和Tool调用配置合理的超时和重试策略。
platform = Platform( llm_config={ “timeout”: 30.0, # 秒 “max_retries”: 2, } )使用更经济的模型:并非所有步骤都需要GPT-4。对于创意发散、文本润色等任务,GPT-4可能效果更好;对于简单的信息提取、格式转换,使用GPT-3.5-Turbo或Claude Haiku可以大幅降低成本。agno允许你为不同的Agent指定不同的模型。
creative_agent = Agent(name=“Creator”, model=“gpt-4”, ...) formatter_agent = Agent(name=“Formatter”, model=“gpt-3.5-turbo”, ...)限制并发数:虽然异步并发能提高速度,但过高的并发可能触发LLM API的速率限制(Rate Limit)。需要在平台层面设置全局并发控制。
platform = Platform(max_concurrent_llm_calls=5) # 限制同时进行的LLM调用数
5.4 安全性考量
智能体能够执行代码(通过Tool)和访问网络,因此安全至关重要。
- Tool权限隔离:不要给所有Agent所有Tool的访问权限。根据“最小权限原则”,只为Agent分配其完成任务所必需的Tool。例如,一个“文本总结Agent”不需要拥有“执行shell命令”的Tool。
- 输入验证与清理:对所有从外部传入工作流或Agent的输入进行严格的验证和清理,防止提示词注入(Prompt Injection)攻击。
- 敏感信息处理:确保API密钥、数据库密码等敏感信息不会通过LLM的上下文泄露。使用环境变量或密钥管理服务,并避免在Tool的返回结果或日志中打印完整敏感信息。
6. 常见问题与排查实录
在开发和运维agno智能体系统的过程中,我踩过不少坑。这里总结一些典型问题和解决方法,希望能帮你节省时间。
6.1 Agent“不听指挥”,不按预期调用Tool
问题现象:你为Agent配置了搜索Tool,并写了详细的指令让它先搜索再回答,但它总是忽略Tool,直接基于自身知识回答。
排查思路:
- 检查instructions:指令是否足够清晰、强硬?使用“你必须先使用X工具搜索”、“在回答前,请务必调用Y工具获取最新信息”等措辞。在指令中明确描述Tool的用途和调用场景。
- 检查Tool的描述:
Tool类中的description字段非常重要。LLM通过这个描述来理解Tool的功能。确保描述准确、简洁,并包含关键词。例如,“搜索互联网获取关于某主题的最新信息”比“进行网络查询”更好。 - 测试Tool的调用:单独测试Tool是否能被Agent的LLM正确识别和调用。你可以编写一个简单的测试脚本,模拟Agent的决策过程。
- 模型能力:过于简单或能力较弱的模型可能无法很好地理解和执行复杂的工具调用指令。尝试换用更强大的模型(如从GPT-3.5升级到GPT-4)进行测试。
解决方案:通常这是Prompt工程问题。重构你的instructions,采用更结构化的格式,例如:
你是一个助手。请按以下步骤工作: 1. 用户提出一个问题。 2. 你必须首先使用‘web_search’工具搜索相关信息。搜索查询词应为问题的核心关键词。 3. 基于搜索结果,组织你的答案。 如果无法使用搜索工具,请直接说明。6.2 工作流执行卡住或超时
问题现象:工作流启动后,长时间没有进展,最终超时。
排查思路:
- 查看日志:首先检查agno和应用日志,看具体卡在哪一个Agent或Tool的执行步骤。
- 检查异步代码:确保你的
Agent.run()或Tool.run()方法是正确定义的async函数,并且在其中正确地使用了await。一个常见的错误是在异步函数中调用了同步的阻塞函数(如requests.get而非aiohttp)。 - 检查外部依赖:卡住通常发生在网络I/O环节。检查LLM API(如OpenAI)的可用性,以及你自定义Tool调用的第三方服务(如搜索API、数据库)是否响应缓慢或不可用。
- 检查循环或条件逻辑:在工作流逻辑中,是否存在死循环的条件?或者某个Agent的产出格式不符合下一个Agent的输入预期,导致流程无法推进?
解决方案:
- 为所有网络调用设置明确的超时。
# 在aiohttp调用中 async with session.post(url, timeout=aiohttp.ClientTimeout(total=30)) as response: ... - 在工作流的关键步骤添加更详细的日志和超时控制。
- 使用
asyncio.wait_for包装可能长时间运行的操作。
6.3 内存消耗过大或增长过快
问题现象:长时间运行后,应用内存占用持续增长。
排查思路:
- 检查记忆(Memory)存储:如果你使用了向量数据库存储长期记忆,并且持续向其中添加内容而不做清理,内存和存储会不断增长。检查记忆的保留策略。
- 检查大对象缓存:是否在Agent或Workflow的实例变量中缓存了大型数据(如图片、长文本)而没有及时释放?
- 工作流实例泄漏:是否持续创建新的工作流实例,而旧的实例因为被引用而无法被垃圾回收?确保平台对已完成的工作流实例有适当的清理机制。
解决方案:
- 为向量记忆设置基于时间或数量的滚动窗口。
- 避免在Agent状态中保存大型数据,优先使用外部存储(如数据库、对象存储)的引用ID。
- 定期重启工作进程(如果使用容器化部署,这是常规操作)。
6.4 如何处理LLM输出的非结构化内容
问题现象:你要求LLM输出一个JSON列表,但它有时会返回纯文本,甚至附带解释,导致后续解析失败。
排查思路:
- 强化输出格式指令:在
instructions中使用非常明确的格式描述,例如:“请严格输出一个JSON数组,格式如下: [{“title”: “...”, “summary”: “...”}]。不要有任何其他解释文字。” - 使用结构化输出功能:如果使用的LLM支持(如OpenAI的GPT-4 Turbo with JSON mode,或Anthropic Claude的响应格式),在调用时强制指定JSON格式。
response = await self.llm.generate(prompt, response_format={“type”: “json_object”}) - 增加后处理层:在Agent内部,对LLM的原始输出进行健壮性解析。使用
json.loads()配合ast.literal_eval()或正则表达式来提取所需的结构化数据,并做好异常处理。
解决方案:结合使用强格式指令、LLM的结构化输出功能以及容错的后处理代码。不要完全信任LLM的输出格式。
构建基于agno的智能体系统,是一个将创意、工程和运维紧密结合的过程。它不是一个“一键生成”的魔法,而是一个需要精心设计、迭代和调优的工程框架。从定义一个简单的Agent开始,逐步构建复杂的工作流,集成各种工具,并最终部署到生产环境,每一步都充满了挑战和乐趣。希望这篇长文能为你探索AGI应用开发提供一块坚实的垫脚石。记住,最好的学习方式就是动手去构建,然后从遇到的问题中不断学习和改进。