基于agent-foundry框架构建智能体：从核心原理到天气助手实战

1. 项目概述从零构建你的智能体开发框架最近在GitHub上看到一个挺有意思的项目叫hebertzhu/agent-foundry。乍一看名字你可能会觉得这又是一个跟风大语言模型热潮的“又一个Agent框架”。但当我真正深入去研究它的代码结构、设计理念和实际应用场景后发现它远不止于此。这个项目更像是一个为开发者准备的“智能体工坊”或者说是一个高度模块化、可插拔的智能体系统构建基础。它不试图提供一个开箱即用、功能大而全的超级智能体而是提供了一套清晰的规范和一系列可组合的“乐高积木”让你能根据自己的需求快速搭建出从简单到复杂的各类智能体应用。简单来说agent-foundry解决的核心痛点是当你想基于大语言模型LLM开发一个具备特定能力的智能体Agent时往往需要从零开始处理任务规划、工具调用、记忆管理、状态流转等一系列复杂问题。这个过程重复且繁琐不同项目间的代码也难以复用。agent-foundry的出现就是为了标准化这个开发流程它定义了智能体的核心组件如大脑、记忆、工具、执行器并提供了这些组件的标准接口和默认实现。开发者可以继承这些接口专注于实现自己业务逻辑而无需关心底层的调度和协同机制。无论你是想做一个能自动分析数据并生成报告的分析智能体一个能理解用户需求并调用API完成订餐、查询的助手智能体还是一个更复杂的、具备多步骤推理和长期目标规划的项目管理智能体agent-foundry都为你提供了一个坚实且灵活的起点。它尤其适合有一定Python开发基础希望将LLM能力深度集成到自身产品或工作流中的开发者、产品经理和技术决策者。接下来我就结合自己的实践经验带你彻底拆解这个框架看看如何用它来高效地构建属于你自己的智能体。2. 核心架构与设计哲学拆解要玩转agent-foundry首先得理解它的“世界观”。这个框架的设计深受现代软件工程中“关注点分离”和“依赖注入”思想的影响。它没有把智能体看作一个黑箱而是将其解构成几个相互协作、职责分明的核心部件。2.1 核心组件智能体的“五脏六腑”框架的核心是几个抽象基类ABC它们定义了智能体系统的骨架Agent智能体这是最高层次的抽象代表一个完整的智能体实例。它通常不直接包含复杂逻辑而是作为一个容器组装了Brain、Memory、Tools等组件并提供一个统一的run或chat接口与外界交互。Brain大脑智能体的核心决策器官。它的主要职责是理解输入用户指令、当前状态、记忆内容进行思考推理、规划并输出决策下一步要执行的动作或直接回复。在大多数实现中Brain会封装与大语言模型如GPT-4、Claude、本地部署的模型的交互逻辑。agent-foundry鼓励你将提示工程Prompt Engineering、思维链Chain-of-Thought等策略实现在Brain中。Memory记忆负责存储和检索智能体与用户交互的历史、知识片段以及智能体自身的状态。记忆可以分为多种类型对话记忆保存简单的轮次对话历史。向量记忆将文本转换成向量存入向量数据库如Chroma, Pinecone实现基于语义的知识检索。这是让智能体“拥有知识”的关键。摘要记忆在长对话中自动将历史对话总结成要点避免上下文窗口爆炸。agent-foundry定义了统一的记忆接口允许你灵活搭配不同的记忆后端。Tool工具智能体延伸的手脚。一个工具本质上是一个可以被智能体调用的函数它让智能体能够与外部世界互动比如执行计算、查询数据库、调用第三方API、操作文件系统等。框架要求每个工具都有清晰的名称、描述和参数定义这正好对应了LLM进行函数调用Function Calling所需的结构化信息。Executor执行器负责执行Brain决策出来的Action通常是调用某个Tool。它处理工具调用的具体逻辑管理执行状态并将执行结果返回给Brain进行下一轮思考。复杂的执行器还能处理并行工具调用、错误重试等。设计哲学解读这种组件化设计最大的好处是可测试性和可替换性。你可以单独为Brain写单元测试模拟LLM的返回可以轻松将对话记忆从内存切换到Redis而无需改动业务代码可以为一个智能体动态增删工具就像给电脑插拔USB设备一样方便。这彻底改变了早期Agent项目中各种逻辑胶着在一起、牵一发而动全身的混乱局面。2.2 控制流智能体如何“思考”与“行动”理解了静态组件我们再看动态的工作流程。一个典型的agent-foundry智能体运行遵循一个清晰的循环通常被称为“ReAct模式”Reasoning and Acting或“规划-执行”循环观察智能体接收用户输入并结合当前Memory中的上下文形成当前的“观察”。思考Brain基于“观察”进行思考。它可能会分析用户意图规划步骤或者决定直接给出答案。如果需要使用工具Brain会生成一个结构化的Action对象指明要调用哪个Tool以及传入什么参数。行动Executor拿到Action找到对应的Tool并执行获取执行结果。这个结果可能是一个计算数值、一段查询到的信息、一个API调用的返回状态等。反思与记忆Brain或专门的Observer组件对行动结果进行“反思”判断目标是否达成、结果是否合理。同时将本轮重要的交互信息用户输入、智能体思考、工具调用及结果存储到Memory中。循环或终止如果目标未达成例如还需要更多信息则回到第1步将行动结果作为新的“观察”输入如果目标已达成则Brain生成最终回复循环结束。这个循环可能发生多次直到任务完成。agent-foundry框架的核心价值之一就是为你标准化并管理了这个循环你只需要关心每个组件内部的业务逻辑实现。3. 从零开始手把手搭建你的第一个智能体理论讲得再多不如动手做一遍。让我们用一个实际案例来贯穿始终构建一个“天气查询与出行建议智能体”。这个智能体能理解用户关于天气的询问调用天气API获取数据并根据天气情况给出简单的出行建议如是否需要带伞、是否适合户外运动。3.1 环境准备与项目初始化首先确保你的Python环境在3.8以上。创建一个新的项目目录并初始化虚拟环境是良好的习惯。mkdir my-weather-agent cd my-weather-agent python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate接下来安装agent-foundry。由于它可能还在快速迭代建议直接从GitHub仓库安装最新版本并安装我们可能用到的额外依赖。pip install agent-foundry githttps://github.com/hebertzhu/agent-foundry.git pip install openai requests python-dotenv这里我们假设使用OpenAI的模型作为Brain使用requests来调用天气APIpython-dotenv用来管理环境变量如API密钥。创建一个.env文件来存放敏感信息OPENAI_API_KEYsk-your-openai-key-here WEATHER_API_KEYyour-weather-api-key-here然后创建一个main.py作为我们的入口文件。3.2 定义核心工具让智能体能“查天气”在agent-foundry中工具是智能体能力的扩展。我们首先来创建一个天气查询工具。# tools/weather_tool.py import requests import json from typing import Dict, Any from agent_foundation.tool import Tool, ToolParameter class WeatherQueryTool(Tool): 一个用于查询城市天气情况的工具。 def __init__(self, api_key: str): super().__init__( nameget_current_weather, description根据城市名称查询当前的天气状况包括温度、天气现象和湿度。, parameters[ ToolParameter( namecity_name, typestring, description需要查询天气的城市名称例如北京、New York。, requiredTrue ) ] ) self.api_key api_key # 这里以假想的天气API为例实际使用时请替换为真实API如OpenWeatherMap self.base_url https://api.weatherapi.com/v1/current.json def execute(self, **kwargs) - Dict[str, Any]: city_name kwargs.get(city_name) if not city_name: return {error: 城市名称不能为空} try: # 构造请求参数 params { key: self.api_key, q: city_name, lang: zh } response requests.get(self.base_url, paramsparams, timeout10) response.raise_for_status() data response.json() # 解析返回数据格式化为易读的文本 location data[location][name] temp_c data[current][temp_c] condition data[current][condition][text] humidity data[current][humidity] result_text f{location}当前天气{condition}气温{temp_c}摄氏度湿度{humidity}%。 return { success: True, data: data, summary: result_text } except requests.exceptions.RequestException as e: return {error: f网络请求失败: {str(e)}} except (KeyError, json.JSONDecodeError) as e: return {error: f解析天气数据失败: {str(e)}}关键点解析继承Tool类这是创建工具的标准方式。定义元信息在__init__中必须定义name,description和parameters。这些信息会被Brain用来生成可供LLM理解的工具调用规范。描述写得越清晰LLM越能准确判断何时调用它。实现execute方法这里是工具的实际业务逻辑。我们调用外部API处理响应并返回一个结构化的字典。返回的数据应该尽可能干净、结构化便于Brain理解和利用。错误处理工具执行可能失败网络、API限制、参数错误必须进行健壮的错误处理并返回明确的错误信息让智能体能知晓失败原因并可能采取补救措施。3.3 构建智能体大脑集成LLM与工具调用接下来我们构建智能体的“大脑”。我们将使用OpenAI的ChatCompletion API并利用其强大的函数调用能力。# brains/openai_brain.py import os from typing import List, Dict, Any, Optional from openai import OpenAI from agent_foundation.brain import Brain from agent_foundation.tool import Tool from agent_foundation.action import Action from dotenv import load_dotenv load_dotenv() class OpenAIBrain(Brain): 一个基于OpenAI API实现的智能体大脑。 def __init__(self, model: str gpt-3.5-turbo, temperature: float 0.1): super().__init__() self.client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) self.model model self.temperature temperature self.conversation_history [] # 简单的对话历史记忆 def set_tools(self, tools: List[Tool]): 大脑需要知道智能体有哪些工具可用。 self.tools tools # 将工具列表转换为OpenAI函数调用格式 self.available_functions [tool.to_openai_function() for tool in tools] def think(self, user_input: str, **kwargs) - Action: 核心思考方法处理用户输入决定下一步动作。 # 1. 更新对话历史 self.conversation_history.append({role: user, content: user_input}) # 2. 准备系统提示词塑造智能体角色和能力 system_prompt 你是一个友好的天气助手。你的主要能力是查询实时天气。 当用户询问某个地方的天气时你应该使用get_current_weather工具来获取准确信息。 得到天气数据后用友好、易懂的语言向用户汇报并可以附加一些简单的出行建议例如下雨记得带伞天气炎热注意防晒。 如果你无法回答用户的问题例如问题与天气无关或工具调用失败请礼貌地告知用户你的能力范围。 # 3. 构造发送给OpenAI的消息序列 messages [ {role: system, content: system_prompt}, *self.conversation_history # 包含历史对话 ] # 4. 调用OpenAI API允许函数调用 response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.available_functions, tool_choiceauto, # 让模型自行决定是否调用工具 temperatureself.temperature, ) response_message response.choices[0].message # 5. 处理响应判断是直接回复还是需要调用工具 if response_message.tool_calls: # 模型决定调用工具 tool_call response_message.tool_calls[0] tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) # 创建一个Action对象指示执行器去调用特定工具 return Action( typetool_call, tool_nametool_name, argumentstool_args, reasoningresponse_message.content # 模型可能附带的思考过程 ) else: # 模型直接给出最终回答 final_answer response_message.content self.conversation_history.append({role: assistant, content: final_answer}) return Action( typefinal_answer, contentfinal_answer ) def reflect(self, observation: Dict[str, Any]) - None: 对工具执行结果进行反思并更新内部状态。 # 将工具执行结果也加入到对话历史中供下一轮思考参考 if observation.get(type) tool_result: result_summary observation.get(summary, str(observation.get(data, ))) self.conversation_history.append({ role: tool, content: f[工具调用结果] {result_summary} })实操心得提示词工程是关键system_prompt定义了智能体的“人格”和能力边界。写一个好的提示词比调参更有效。这里我们明确告诉模型“你是天气助手主要用XX工具拿到数据后要解释并给建议”。温度参数temperature设置为较低的0.1是为了让模型的输出更稳定、更可预测这对于执行具体任务的智能体很重要。如果你希望智能体更有“创意”可以适当调高。历史管理这里用了简单的列表来管理对话历史。在实际复杂场景中你应该将其抽离到专门的Memory组件中支持长上下文、摘要等功能。错误处理上述代码省略了完整的错误处理如API调用失败、额度不足。在生产环境中必须添加重试、降级如切换模型等逻辑。3.4 组装与运行让智能体活起来现在我们把所有部件组装起来并创建一个简单的执行器来驱动整个循环。# main.py import os import json from dotenv import load_dotenv from brains.openai_brain import OpenAIBrain from tools.weather_tool import WeatherQueryTool from agent_foundation.action import Action load_dotenv() class SimpleExecutor: 一个简单的执行器负责执行大脑产生的Action。 def __init__(self, tools): # 建立工具名称到工具实例的映射 self.tool_registry {tool.name: tool for tool in tools} def execute(self, action: Action) - Dict: if action.type tool_call: tool self.tool_registry.get(action.tool_name) if not tool: return {error: f未知工具: {action.tool_name}} print(f[执行器] 正在调用工具 {action.tool_name}参数: {action.arguments}) # 执行工具并返回结果 result tool.execute(**action.arguments) return {type: tool_result, data: result} elif action.type final_answer: print(f[智能体回复]: {action.content}) return {type: final_answer, content: action.content} else: return {error: f未知的Action类型: {action.type}} def main(): # 1. 初始化工具 weather_tool WeatherQueryTool(api_keyos.getenv(WEATHER_API_KEY)) tools [weather_tool] # 2. 初始化大脑并为其配备工具 brain OpenAIBrain(modelgpt-3.5-turbo) brain.set_tools(tools) # 3. 初始化执行器 executor SimpleExecutor(tools) print(天气助手智能体已启动输入退出或quit结束对话。) while True: try: user_input input(\n你: ).strip() if user_input.lower() in [退出, quit, exit]: print(再见) break if not user_input: continue # 4. 思考-执行循环 max_turns 5 # 防止无限循环 for turn in range(max_turns): print(f\n[第{turn1}轮思考]...) action brain.think(user_input) if action.type final_answer: # 直接给出最终答案循环结束 executor.execute(action) break else: # 需要调用工具 result executor.execute(action) if result.get(error): print(f工具执行出错: {result[error]}) # 可以将错误信息作为观察让大脑重新思考 observation {type: error, message: result[error]} else: # 将工具执行结果作为新的观察 observation result # 大脑进行反思并准备下一轮思考在think方法中历史已更新 # 这里简化处理将观察传递给大脑的reflect方法 brain.reflect(observation) # 下一轮思考的“用户输入”是上一步的结果这里我们设定为让大脑自主继续 # 实际上更复杂的逻辑应由大脑根据历史决定是否继续或停止。 # 本例中我们假设一次工具调用后大脑在下一轮think中会基于完整历史生成最终答案。 # 因此我们清空user_input让大脑基于历史对话自行决策。 user_input except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f发生未知错误: {e}) break if __name__ __main__: main()运行这个程序你就可以在命令行和你的天气助手对话了。试试输入“上海今天天气怎么样”或者“北京和广州的天气对比如何”。观察控制台打印的日志你会清晰地看到“思考”、“调用工具”、“返回结果”、“生成回复”的完整过程。4. 进阶实践提升智能体的能力与可靠性一个基础的智能体跑起来后你会发现很多可以优化和增强的地方。agent-foundry的模块化设计让这些进阶功能变得易于实现。4.1 为智能体注入“长期记忆”之前的记忆只在单次对话中有效程序重启就消失了。我们可以实现一个基于向量数据库的长期记忆让智能体记住跨会话的关键信息。这里以使用Chroma这个轻量级向量数据库为例# memory/vector_memory.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import uuid from typing import List, Dict, Any class VectorMemory: 基于向量数据库的长期记忆模块。 def __init__(self, persist_directory: str ./chroma_db): self.client chromadb.PersistentClient(pathpersist_directory, settingsSettings(anonymized_telemetryFalse)) # 使用一个轻量级的句子嵌入模型 self.embedder SentenceTransformer(all-MiniLM-L6-v2) self.collection self.client.get_or_create_collection(nameagent_memory) def store(self, text: str, metadata: Dict[str, Any] None): 存储一段文本记忆。 embedding self.embedder.encode(text).tolist() doc_id str(uuid.uuid4()) self.collection.add( documents[text], embeddings[embedding], metadatas[metadata] if metadata else [{}], ids[doc_id] ) return doc_id def search(self, query: str, n_results: int 3) - List[Dict]: 根据查询文本检索最相关的记忆。 query_embedding self.embedder.encode(query).tolist() results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # 格式化返回结果 memories [] if results[documents]: for doc, meta in zip(results[documents][0], results[metadatas][0]): memories.append({content: doc, metadata: meta}) return memories def clear(self): 清空记忆谨慎使用。 self.client.delete_collection(nameagent_memory) self.collection self.client.create_collection(nameagent_memory)然后在你的OpenAIBrain.think方法中在生成回复前可以先从VectorMemory中检索相关记忆并将其作为上下文注入系统提示词或用户消息中。例如当用户说“还记得我上次问的哪个城市最宜居吗”智能体可以通过检索记忆来回答。4.2 实现复杂的任务规划与分解对于“帮我规划一个周末的北京旅行要考虑天气和景点”这类复杂任务单次工具调用无法解决。我们需要增强Brain的规划能力。一种常见模式是让Brain先输出一个规划Plan然后由Executor逐步执行这个规划中的每一步。你可以定义一个Plan数据结构包含多个步骤Step。Brain在首次思考时生成一个PlanAction。Executor接收到Plan后将其存入状态并开始执行第一个Step。每个Step可能对应一个工具调用执行完一个Step后将结果反馈给BrainBrain进行反思并决定是继续下一个Step还是修改Plan。这需要你扩展Action的类型并实现一个更复杂的、有状态的Executor。agent-foundry的框架本身不限制这种复杂控制流的实现它提供了足够的基础设施让你来构建。4.3 工具调用验证与安全允许LLM直接调用工具存在潜在风险如无限循环、调用危险操作。在生产环境中必须加入验证层。参数验证在工具execute方法内部对输入参数进行严格的类型和范围检查。权限控制可以为工具打上标签如read_only,high_cost,dangerous并在Executor中根据智能体角色或用户权限来决定是否允许调用。调用限制在Executor或框架层面限制单个会话的工具调用次数、频率或总耗时防止恶意或错误的指令导致资源耗尽。人工确认对于关键操作如发送邮件、支付可以让Executor暂停将行动请求提交给用户确认后再执行。5. 常见问题、调试技巧与性能优化在实际开发中你肯定会遇到各种问题。下面是一些常见坑点和解决思路。5.1 智能体不调用工具或调用错误工具问题你明明定义了工具但智能体总是直接回答“我不知道”或者试图用自然语言描述而不是调用工具。排查检查工具描述name和description是否清晰、无歧义描述是否准确反映了工具的功能LLM主要靠这个来决定是否调用。试着让描述更贴近用户可能的问题例如“查询实时天气”比“获取气象数据”更好。检查系统提示词你的system_prompt是否明确指示了智能体要使用工具像“当你需要查询天气时请务必使用get_current_weather工具”这样的强指令往往更有效。检查函数调用格式确保tool.to_openai_function()返回的格式符合OpenAI的函数调用规范。可以用print(json.dumps(tool.to_openai_function(), indent2))查看。测试LLM理解单独用一段包含工具定义的对话历史去调用OpenAI API看它是否会生成正确的tool_calls。这可以排除框架代码的问题。5.2 工具执行结果未被有效利用问题工具成功执行并返回了数据但智能体最终的回复里没有包含这些数据或者处理错了。排查检查结果格式工具返回的字典是否结构清晰Brain的reflect方法是否正确地将结果摘要加入了对话历史确保结果以LLM容易理解的文本格式呈现。检查历史上下文工具执行结果作为一条role: tool的消息插入历史后在下一次think时这条消息是否被包含在发送给LLM的messages列表中确保历史管理逻辑正确。提示词引导在系统提示词中可以加入“当你收到工具返回的结果后请基于结果组织你的回答”之类的引导。5.3 处理长上下文与成本控制挑战对话历史越来越长导致API调用token数暴涨速度变慢成本激增。解决方案记忆摘要实现一个SummarizerMemory定期将过长的对话历史总结成几个要点然后用摘要替换掉原始的长文本历史。滑动窗口只保留最近N轮对话作为上下文丢弃更早的。选择性记忆不是所有对话都需要记住。可以让Brain或一个单独的Filter组件来判断哪些信息是重要的例如包含用户偏好、任务关键信息只将这些存入长期记忆向量数据库。使用更经济的模型在非关键思考步骤可以使用gpt-3.5-turbo代替gpt-4。或者将复杂的规划任务拆解部分子任务用小型模型处理。5.4 性能优化与异步处理场景当智能体需要并行调用多个不依赖的工具时同步顺序执行会拖慢速度。优化将Executor改造成异步的。使用asyncio库当Brain产生一个包含多个并行Action的Plan时Executor可以使用asyncio.gather来并发执行这些工具调用大幅减少等待时间。注意这要求你使用的工具库或HTTP客户端支持异步操作如aiohttp。import asyncio class AsyncTool(Tool): async def execute_async(self, **kwargs): # 异步执行逻辑 pass class AsyncExecutor: async def execute_parallel(self, actions: List[Action]): tasks [] for action in actions: if action.type tool_call: tool self.tool_registry[action.tool_name] if isinstance(tool, AsyncTool): tasks.append(tool.execute_async(**action.arguments)) else: # 同步工具包装成异步任务 tasks.append(asyncio.to_thread(tool.execute, **action.arguments)) results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果 return results5.5 监控与日志对于一个运行中的智能体系统完善的监控和日志至关重要。结构化日志使用structlog或logging模块为每一轮思考、每一个工具调用、每一次错误记录结构化的日志包含会话ID、时间戳、动作类型、输入输出摘要等。这便于后期分析和调试。关键指标监控平均响应时间、工具调用成功率、各LLM模型的token消耗、错误率等。追踪与调试为每个用户会话或任务生成一个唯一的trace_id贯穿整个处理链路。这样当出现问题时你可以轻松地拉取该次会话的所有相关日志重现整个决策过程。通过agent-foundry构建智能体是一个从理解框架、组装组件、到不断迭代优化智能体行为的过程。它提供的不是一条捷径而是一条清晰、可维护的路径。当你熟悉了这套范式后你会发现构建一个功能强大的智能体应用不再是一个从零开始的庞大工程而是变成了有趣的“组件拼装”和“大脑调教”工作。你可以专注于领域业务逻辑和用户体验而将智能体系统的复杂性交给框架来处理。