Python agent-hub 包完全指南:功能、安装、语法与案例
1. 引言
随着大语言模型(LLM)与自动化编排技术的快速发展,Python 生态中涌现出大量用于构建智能体(Agent)的工具包。agent-hub 正是其中一款专注于「多智能体协作」与「任务编排」的轻量级框架。它屏蔽了底层模型调用的复杂性,提供统一的任务分发、记忆管理和工具注册机制,帮助开发者快速搭建从单智能体问答到多智能体协同的完整应用。
本文将从功能特性、安装方式、核心语法与参数、16 个实际应用案例以及常见错误与注意事项五个维度,系统性地介绍 agent-hub 的使用方法。
2. agent-hub 核心功能
agent-hub 的核心设计目标是「让智能体开发像写普通函数一样简单」。它主要提供以下能力:
- 多智能体编排:支持创建多个独立智能体,并通过 Hub 统一调度,实现任务拆分、并行执行与结果汇总。
- 统一模型接入:内置 OpenAI、Anthropic、通义千问等主流模型适配器,通过统一接口切换底层模型。
- 工具注册与调用:支持将普通 Python 函数注册为智能体可调用的工具,自动完成参数解析与结果回传。
- 记忆管理:提供短期会话记忆与长期向量记忆两种模式,支持对话上下文的持久化。
- 任务队列与重试:内置任务队列,支持失败重试、超时控制与并发限制。
- 流式输出:支持流式返回模型生成结果,适合构建打字机效果的对话界面。
- 可观测性:提供结构化日志与调用链追踪,方便调试多智能体协作过程。
3. 安装与环境准备
agent-hub 要求 Python 3.9 及以上版本。推荐使用虚拟环境进行安装,避免污染全局环境。
# 创建并激活虚拟环境(可选但推荐) python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate 通过 pip 安装 pip install agent-hub 如需使用向量记忆功能,安装额外依赖 pip install agent-hub[memory] 如需使用全部内置模型适配器 pip install agent-hub[all]安装完成后,可以通过以下命令验证是否安装成功:
python -c "import agent_hub; print(agent_hub.__version__)"4. 核心语法与参数详解
agent-hub 的使用围绕三个核心类展开:Agent、Hub和Tool。下面逐一介绍其关键参数。
4.1 Agent 类
Agent是智能体的基本单元,负责接收用户输入并调用模型生成回复。
from agent_hub import Agent agent = Agent( name="assistant", # 智能体名称,用于日志与追踪 system_prompt="你是一个乐于助人的助手。", # 系统提示词 model="gpt-4o-mini", # 模型名称 temperature=0.7, # 采样温度,控制随机性 max_tokens=2048, # 单次生成的最大 token 数 timeout=60, # 请求超时时间(秒) memory=True, # 是否启用会话记忆 tools=[], # 可调用工具列表 api_key=None, # API 密钥,默认读取环境变量 base_url=None, # 自定义 API 地址 )关键参数说明:
name:智能体唯一标识,在多智能体协作中用于路由与日志区分。system_prompt:设定智能体的角色与行为边界,对输出质量影响最大。model:支持gpt-4o、gpt-4o-mini、claude-3-5-sonnet、qwen-plus等。temperature:取值范围 0 到 2,值越低输出越确定,适合代码生成;值越高越有创造性。memory:为True时自动维护对话历史,为False时每次请求独立。
4.2 Hub 类
Hub是任务编排的核心,负责管理多个智能体并分发任务。
from agent_hub import Hub hub = Hub( agents=[agent1, agent2], # 注册的智能体列表 strategy="auto", # 调度策略:auto / round_robin / manual max_concurrency=5, # 最大并发数 retry_times=3, # 失败重试次数 retry_interval=2, # 重试间隔(秒) log_level="INFO", # 日志级别 )关键参数说明:
strategy:auto根据任务内容自动选择最合适的智能体;round_robin轮询分发;manual手动指定目标智能体。max_concurrency:控制同时执行的智能体数量,避免触发模型 API 限流。retry_times:当模型调用失败或返回异常时自动重试的次数。
4.3 Tool 装饰器
通过@tool装饰器可以将普通函数注册为智能体可调用的工具。
from agent_hub import tool @tool(name="calculator", description="执行四则运算") def calculator(expression: str) -> str: """计算数学表达式的结果。""" return str(eval(expression))参数说明:
name:工具名称,模型通过该名称调用工具。description:工具功能描述,模型据此判断何时调用该工具。- 函数签名中的类型注解会被自动解析为工具参数 schema,因此务必为参数添加类型注解。
4.4 调用与流式输出
# 普通调用 response = agent.chat("你好,请介绍一下你自己") 流式调用 for chunk in agent.stream("写一首关于秋天的诗"): print(chunk, end="", flush=True) Hub 分发任务 result = hub.run("请帮我总结这份文档并生成摘要")5. 16 个实际应用案例
下面通过 16 个由浅入深的案例,展示 agent-hub 在不同场景下的实际用法。
案例 1:基础问答智能体
创建一个最简单的问答智能体,用于处理日常咨询类问题。
from agent_hub import Agent agent = Agent( name="qa_bot", system_prompt="你是一个知识渊博的问答助手,回答要简洁准确。", model="gpt-4o-mini", ) response = agent.chat("Python 中如何实现列表去重?") print(response)案例 2:带记忆的多轮对话
启用记忆功能,让智能体记住上下文,实现连贯的多轮对话。
from agent_hub import Agent agent = Agent( name="chat_bot", system_prompt="你是一个贴心的聊天伙伴。", model="gpt-4o-mini", memory=True, ) agent.chat("我叫小明,今年 25 岁。") agent.chat("请记住我的名字和年龄。") response = agent.chat("我叫什么名字?") print(response) # 输出:你叫小明。案例 3:注册并使用自定义工具
将本地函数注册为工具,让智能体具备调用外部能力。
from agent_hub import Agent, tool @tool(name="get_weather", description="查询指定城市的天气") def get_weather(city: str) -> str: """模拟天气查询接口。""" weather_map = {"北京": "晴,25°C", "上海": "多云,28°C"} return weather_map.get(city, "暂无数据") agent = Agent( name="weather_bot", system_prompt="你是一个天气助手,使用工具查询天气。", model="gpt-4o-mini", tools=[get_weather], ) response = agent.chat("北京今天天气怎么样?") print(response)案例 4:多智能体协作——翻译与校对
创建翻译智能体和校对智能体,通过 Hub 串联执行。
from agent_hub import Agent, Hub translator = Agent( name="translator", system_prompt="将用户输入翻译为英文,只输出译文。", model="gpt-4o-mini", ) proofreader = Agent( name="proofreader", system_prompt="检查英文语法和用词,输出修正后的版本。", model="gpt-4o-mini", ) hub = Hub(agents=[translator, proofreader], strategy="manual") translated = hub.run("今天天气很好,我们去公园散步吧。", target="translator") final = hub.run(translated, target="proofreader") print(final)案例 5:并行任务分发
将多个独立任务并行分发给不同智能体,提升处理效率。
from agent_hub import Agent, Hub summarizer = Agent( name="summarizer", system_prompt="用一句话总结用户输入。", model="gpt-4o-mini", ) classifier = Agent( name="classifier", system_prompt="将用户输入分类为:技术、生活、娱乐。", model="gpt-4o-mini", ) hub = Hub(agents=[summarizer, classifier], max_concurrency=2) tasks = [ "Python 是一种解释型语言。", "周末去爬山很放松。", "最近上映的电影很好看。", ] results = hub.run_many(tasks) for task, result in zip(tasks, results): print(f"输入:{task} -> 输出:{result}")案例 6:代码生成与执行
让智能体生成代码,并通过工具执行验证结果。
from agent_hub import Agent, tool @tool(name="run_python", description="执行 Python 代码并返回结果") def run_python(code: str) -> str: """在沙箱中执行 Python 代码。""" import subprocess result = subprocess.run( ["python", "-c", code], capture_output=True, text=True, timeout=10, ) return result.stdout or result.stderr agent = Agent( name="coder", system_prompt="你是一个 Python 编程助手,生成代码后用工具执行验证。", model="gpt-4o", tools=[run_python], ) response = agent.chat("写一个计算斐波那契数列前 10 项的程序并运行。") print(response)案例 7:结构化数据提取
从非结构化文本中提取结构化信息,并输出为 JSON 格式。
from agent_hub import Agent agent = Agent( name="extractor", system_prompt="从用户输入中提取姓名、年龄、城市,以 JSON 格式输出。", model="gpt-4o-mini", ) text = "我叫李华,今年 30 岁,住在杭州。" response = agent.chat(text) print(response) 输出:{"name": "李华", "age": 30, "city": "杭州"}案例 8:文档问答(RAG 简化版)
结合向量记忆功能,实现基于文档内容的问答。
from agent_hub import Agent agent = Agent( name="doc_qa", system_prompt="基于提供的文档内容回答问题,不要编造信息。", model="gpt-4o-mini", memory=True, ) 注入文档内容 agent.chat("以下是公司制度文档:员工每年享有 15 天年假,入职满一年后可申请。") response = agent.chat("员工每年有多少天年假?") print(response)案例 9:情感分析
构建一个情感分析智能体,判断文本的情感倾向。
from agent_hub import Agent agent = Agent( name="sentiment", system_prompt="判断用户输入的情感倾向,输出:积极、消极或中性。", model="gpt-4o-mini", temperature=0.2, ) for text in ["这个产品太好用了!", "服务态度很差,很失望。", "今天天气不错。"]: print(f"{text} -> {agent.chat(text)}")案例 10:多步骤任务编排
通过 Hub 实现「生成大纲 - 扩写内容 - 润色成稿」的多步骤流水线。
from agent_hub import Agent, Hub outliner = Agent( name="outliner", system_prompt="为指定主题生成文章大纲,输出编号列表。", model="gpt-4o-mini", ) writer = Agent( name="writer", system_prompt="根据大纲扩写为完整文章段落。", model="gpt-4o-mini", ) polisher = Agent( name="polisher", system_prompt="润色文章,使语言更流畅专业。", model="gpt-4o-mini", ) hub = Hub(agents=[outliner, writer, polisher], strategy="manual") outline = hub.run("人工智能的发展趋势", target="outliner") draft = hub.run(outline, target="writer") final = hub.run(draft, target="polisher") print(final)案例 11:客服自动回复
构建客服智能体,根据用户问题自动匹配答案或转人工。
from agent_hub import Agent, tool @tool(name="search_faq", description="搜索常见问题库") def search_faq(keyword: str) -> str: """模拟 FAQ 检索。""" faq = { "退货": "支持 7 天无理由退货。", "物流": "默认发货后 3-5 天送达。", } return faq.get(keyword, "未找到相关答案") agent = Agent( name="customer_service", system_prompt="你是电商客服,优先使用工具检索 FAQ,无法解决时请用户转人工。", model="gpt-4o-mini", tools=[search_faq], ) print(agent.chat("我想退货怎么办?"))案例 12:SQL 查询助手
让智能体根据自然语言生成 SQL 查询语句。
from agent_hub import Agent agent = Agent( name="sql_helper", system_prompt="将用户的中文描述转换为 SQL 查询语句,只输出 SQL。", model="gpt-4o", temperature=0.1, ) response = agent.chat("查询 users 表中年龄大于 18 的所有用户的姓名和邮箱") print(response) 输出:SELECT name, email FROM users WHERE age > 18;案例 13:内容审核
构建内容审核智能体,检测文本中的违规内容。
from agent_hub import Agent agent = Agent( name="moderator", system_prompt="审核用户输入是否包含暴力、色情、辱骂等违规内容,输出:通过或违规。", model="gpt-4o-mini", temperature=0, ) text = "这个方案太愚蠢了,简直是一堆垃圾!" print(agent.chat(text))案例 14:定时任务与自动化
结合任务队列,实现定时触发的自动化处理。
from agent_hub import Agent, Hub import time agent = Agent( name="reporter", system_prompt="生成每日工作简报。", model="gpt-4o-mini", ) hub = Hub(agents=[agent]) 模拟每日定时任务 for day in range(3): report = hub.run(f"生成第 {day + 1} 天的工作简报") print(f"Day {day + 1}: {report}") time.sleep(1)案例 15:多语言翻译服务
创建支持多语言互译的智能体服务。
from agent_hub import Agent agent = Agent( name="translator", system_prompt="将用户输入翻译为指定目标语言,目标语言由用户指定。", model="gpt-4o-mini", ) print(agent.chat("请把这句话翻译成日语:你好,很高兴认识你。")) print(agent.chat("请把这句话翻译成法语:今天是个好日子。"))案例 16:数据清洗与格式化
利用智能体对脏数据进行清洗和标准化处理。
from agent_hub import Agent agent = Agent( name="cleaner", system_prompt="清洗用户输入的文本:去除多余空格、统一标点、修正明显错别字。", model="gpt-4o-mini", ) dirty_data = "这是 一段 含 有 多余空格 的文本,,并且标点混乱" print(agent.chat(dirty_data))6. 常见错误与使用注意事项
在实际使用 agent-hub 的过程中,开发者常会遇到以下几类问题,下面逐一说明原因与解决方案。
6.1 API Key 未配置
错误现象:调用时抛出AuthenticationError或提示api_key is required。
原因:未设置环境变量,也未在创建 Agent 时显式传入api_key。
解决方案:
# 方式一:设置环境变量 export OPENAI_API_KEY="sk-xxxx" 方式二:代码中显式传入 agent = Agent(..., api_key="sk-xxxx")6.2 工具函数参数类型缺失
错误现象:模型调用工具时提示参数解析失败。
原因:工具函数未添加类型注解,导致无法生成参数 schema。
解决方案:为所有工具参数添加类型注解。
# 错误写法 @tool(name="add", description="加法") def add(a, b): return a + b 正确写法 @tool(name="add", description="加法") def add(a: int, b: int) -> int: return a + b6.3 并发过高触发限流
错误现象:大量请求返回429 Rate Limit错误。
原因:max_concurrency设置过高,超出模型 API 的速率限制。
解决方案:适当降低并发数,并开启重试机制。
hub = Hub( agents=[agent], max_concurrency=3, retry_times=5, retry_interval=3, )6.4 记忆功能导致上下文过长
错误现象:多轮对话后请求报错context length exceeded。
原因:长期开启memory=True,历史消息不断累积超出模型上下文窗口。
解决方案:定期清理记忆,或使用滑动窗口策略。
# 清理历史记忆 agent.clear_memory() 或限制记忆轮数(部分版本支持) agent = Agent(..., memory=True, max_memory_rounds=10)6.5 工具调用陷入死循环
错误现象:智能体反复调用同一工具,无法生成最终回复。
原因:工具返回结果不满足模型预期,模型不断重试。
解决方案:设置最大工具调用次数。
agent = Agent( ..., max_tool_calls=5, # 限制单次对话最多调用工具次数 )6.6 流式输出与工具调用冲突
错误现象:使用stream方法时,工具调用结果无法正常返回。
原因:部分版本在流式模式下不支持工具调用。
解决方案:工具调用场景使用普通chat方法,流式输出仅用于纯文本生成。
6.7 模型名称拼写错误
错误现象:请求返回ModelNotFoundError。
原因:模型名称拼写错误或该模型不在当前 API 服务范围内。
解决方案:核对模型名称,确认与 API 服务商提供的模型列表一致。
6.8 自定义 base_url 配置错误
错误现象:请求返回ConnectionError或404。
原因:base_url指向的地址不正确,或缺少必要的路径前缀。
解决方案:确认 base_url 格式,通常需要以/v1结尾。
agent = Agent( ..., base_url="https://api.example.com/v1", )6.9 系统提示词过于模糊
错误现象:智能体输出内容偏离预期,回答质量不稳定。
原因:system_prompt未明确角色、任务边界和输出格式。
解决方案:编写清晰、具体的系统提示词,明确输出格式要求。
agent = Agent( name="assistant", system_prompt="你是一个 Python 技术专家。回答时先给出结论,再给出代码示例,代码必须使用 markdown 代码块包裹。", model="gpt-4o-mini", )6.10 温度参数设置不当
错误现象:代码生成任务输出不稳定,或创意写作任务输出过于死板。
原因:temperature参数与任务类型不匹配。
解决方案:代码、SQL、数据提取等确定性任务使用低温度(0 到 0.3);创意写作、头脑风暴使用高温度(0.7 到 1.0)。
6.11 多智能体任务路由错误
错误现象:hub.run将任务分发给了错误的智能体。
原因:strategy="auto"时,Hub 根据系统提示词自动匹配,匹配不准确。
解决方案:使用strategy="manual"并显式指定target参数。
result = hub.run("翻译这段话", target="translator")6.12 依赖版本冲突
错误现象:安装 agent-hub 后,其他库导入报错。
原因:agent-hub 的依赖(如 pydantic、openai)与项目现有版本冲突。
解决方案:使用虚拟环境隔离依赖,或升级相关库到兼容版本。
pip install --upgrade pydantic openai6.13 忽略异常处理
错误现象:网络波动或 API 异常导致程序崩溃。
原因:未对模型调用做异常捕获。
解决方案:使用 try-except 包裹调用逻辑。
try: response = agent.chat("你好") except Exception as e: print(f"调用失败:{e}") # 执行降级逻辑6.14 日志级别设置过高
错误现象:调试时看不到任何日志输出。
原因:log_level设置为ERROR,屏蔽了 INFO 级别的调试信息。
解决方案:调试阶段设置为DEBUG或INFO。
hub = Hub(agents=[agent], log_level="DEBUG")6.15 未设置超时导致请求挂起
错误现象:程序长时间无响应。
原因:未设置timeout,模型 API 长时间未返回。
解决方案:为 Agent 设置合理的超时时间。
agent = Agent(..., timeout=30)6.16 生产环境安全注意事项
在生产环境部署时,需要注意以下几点:
- 密钥管理:不要将 API Key 硬编码在代码中,应使用环境变量或密钥管理服务。
- 输入过滤:对用户输入进行长度限制和内容过滤,防止 Prompt 注入攻击。
- 工具权限控制:注册工具时遵循最小权限原则,避免暴露危险操作。
- 日志脱敏:日志中不要记录完整的 API Key 或用户敏感信息。
- 限流与熔断:在应用层增加限流和熔断机制,保护下游模型服务。
7. 总结
agent-hub 通过简洁的 API 设计和灵活的多智能体编排能力,显著降低了 LLM 应用开发的门槛。本文从功能、安装、语法参数、16 个实战案例以及常见错误五个方面进行了系统梳理。在实际项目中,建议从单智能体场景入手,逐步过渡到多智能体协作;同时重视系统提示词设计、参数调优和异常处理,才能构建稳定可靠的智能体应用。
《动手学PyTorch建模与应用:从深度学习到大模型》是一本从零基础上手深度学习和大模型的PyTorch实战指南。全书共11章,前6章涵盖深度学习基础,包括张量运算、神经网络原理、数据预处理及卷积神经网络等;后5章进阶探讨图像、文本、音频建模技术,并结合Transformer架构解析大语言模型的开发实践。书中通过房价预测、图像分类等案例讲解模型构建方法,每章附有动手练习题,帮助读者巩固实战能力。内容兼顾数学原理与工程实现,适配PyTorch框架最新技术发展趋势。
