开发AI代理必备的8个Python库

让我直言不讳：我曾经就是那样的人。从头开始构建每个代理组件的人。

需要记忆吗？ “我将实现一个向量存储包装器。”想要重试吗？ “给我一个小时和一些尝试除外的块。”

三周、400 行后，我的内存系统出现了一个错误，重复的条目损坏了我的整个索引。实际的代理逻辑？也许 50 行。

所以今天，我将分解 8 个使我的自定义代码过时的库。这些不仅仅是流行。它们实用、设计简洁，并以优雅的方式解决真正的代理难题。

您的代理使用 GPT-4。现在你想考验克劳德。恭喜，您正在重写 API 层。

from litellm import completionresponse = completion(model="gpt-4", messages=[{"role": "user", "content": "Hello"}])response = completion(model="claude-3-opus-20240229", messages=[{"role": "user", "content": "Hello"}])response = completion(model="ollama/llama2", messages=[{"role": "user", "content": "Hello"}])

一个接口。 100多个型号。通过更改字符串来切换提供商。

我上个季度进行了成本比较：相同的工作负载，GPT-4 与 Claude Sonnet 与 Llama 3 70B。差额为 847 美元/月。如果没有 LiteLLM，测试将需要花费一周的重构时间。

✔️ 将其用于： 多模型测试、提供商回退、成本优化

♂️ 专业提示： 设置自动后备。当 OpenAI 宕机时（它确实会宕机），你的代理会在你不醒来的情况下切换到 Anthropic。

你要求 JSON。您将获得带有 Markdown 反引号的 JSON。或者额外的字段。或者看起来像 JSON 但实际上不是的字符串。

import instructorfrom pydantic import BaseModelfrom openai import OpenAIclient = instructor.from_openai(OpenAI())class UserInfo(BaseModel):    name: str    age: intuser = client.chat.completions.create(    model="gpt-4",    messages=[{"role": "user", "content": "Extract: John is 25 years old"}],    response_model=UserInfo)

保证有效输出。如果提取失败，Instructor 会向模型显示出了什么问题并自动重试。

在使用 Instructor 之前，我有一个 150 行的 parse_llm_json() 函数，有 12 个边缘情况。它仍然在每周一次的生产数据上失败。现在？四个月内零解析错误。

✔️ 用于： 任何结构化提取、工具调用解析、数据管道

洞察力： 与 Pydantic AI 完美搭配。相同的理念，不同的层次。

API 失败。网络超时。达到速率限制。你的代理崩溃了，你醒来时看到愤怒的 Slack 消息。

from tenacity import retry, stop_after_attempt, wait_exponential@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))def call_flaky_api():    return external_service.get_data()

指数退避、抖动、自定义重试条件。包装任何函数并忘记暂时性故障。

我曾经嵌套了四层深度的 try- except 块。仍然错过了边缘情况。一个周日早上，我收到了 47 个 Slack 警报，因为第三方 API 在 20 分钟内返回了 503，而我的“强大”错误处理刚刚记录并继续。

✔️ 将其用于： 每个外部 API 调用。每一个。

⚠️ 警告： 不要重试 4xx 错误。您只会更快地获得速率限制。

PART 04

Logfire： 提高AI代理的可观察性

您的代理不断地使用相同的输入调用相同的 API。您无缘无故地烧钱并增加延迟。

from diskcache import Cachecache = Cache('./agent_cache')@cache.memoize(expire=3600)def expensive_embedding(text: str):    return openai.embeddings.create(input=text, model="text-embedding-3-small")

基于文件的持久缓存。幸存重新启动。处理并发访问。无需维护 Redis 服务器。

我将 Diskcache 添加到每天进行 10K 以上调用的嵌入管道中。许多都是重复的。 API 成本第一周下降了 34%。应该早几个月就做好了。

✔️ 将其用于： 嵌入缓存、API 响应缓存、昂贵的计算记忆

见解： 多个代理进程可以共享相同的缓存而不会损坏。经过艰苦的测试。

您的代理会填充上下文，直到达到限制，然后因神秘的 API 错误而崩溃。或者更糟糕的是，默默地截断重要信息。

import tiktokenenc = tiktoken.encoding_for_model("gpt-4")def fits_in_context(messages: list, max_tokens: int = 8000) -> bool:    total = sum(len(enc.encode(m["content"])) for m in messages)    return total < max_tokens

按照 OpenAI 的方式精确计算代币。发送之前先了解一下。

我过去常常通过将字符数除以 4 来估计令牌。工作得很好，直到客户上传了包含大量代码和特殊字符的文档。我的估计是 6K 代币。实际计数为 11K。特工在谈话中崩溃了。客户没有留下深刻的印象。

✔️ 用于： 上下文窗口管理、成本估算、智能截断

⚠️ 警告： 基于字符的估计在真实数据上有 30% 的时间是错误的。不要像我一样学习这个。

您的代理打印 500 行 JSON blob。你滚动。你眯起眼睛。您放弃并添加更多打印语句。

from rich.console import Consolefrom rich.table import Tablefrom rich import print_jsonconsole = Console()print_json(data=agent_response)table = Table(title="Agent Actions")table.add_column("Step")table.add_column("Tool")table.add_column("Result")for step in agent.history:    table.add_row(str(step.num), step.tool, step.result[:50])console.print(table)

表格、语法突出显示、进度条、树视图。您可以实际读取调试输出。

✔️ 用于： 开发调试、演示输出、CLI 界面

真正的谈话： from rich import print 取代了内置打印。一次导入，即时升级。上个月，我向客户演示了一个代理，其丰富格式的输出比代理本身得到了更多的称赞。

您更改提示，重新启动代理，等待其初始化，测试，意识到您犯了拼写错误，然后再次重新启动......

from watchfiles import run_processdef main():    from my_agent import Agent    agent = Agent()    agent.run_interactive()if __name__ == "__main__":    run_process("./", target=main)

检测更改，触发重新加载。当您保存时，您的代理会自动重新启动。

在使用它之前，我跟踪了我的工作流程一天。我重新启动代理 40 多次。每次重新启动需要 8 秒。就这样等了5分钟多。每天。持续数周。

✔️ 用途： 本地开发、快速迭代、快速原型制作

♂️ 专业提示： 明智地了解哪些变化很重要。修改提示文件，即时重新加载。保存日志文件，无需重新加载。

每个库都做好一件事。这并非偶然。

我刚开始构建智能体时，总是选择那些号称能处理所有问题的框架。但它们复杂、固执己见，而且非常脆弱。一旦出现问题，我调试的却是框架本身，而不是我的智能体。我曾写过一篇关于大多数人工智能智能体在生产环境中失败原因的文章，而这正是其中一个原因。

现在我编写小型库。 LiteLLM 用于模型调用。结构化输出讲师。重试的毅力。每一块都是可更换的。每一件作品都可以理解。

最好的代理架构并不是最复杂的。您可以在五分钟内追踪任何错误。

版权声明

转自新缸中之脑，版权属于原作者，仅用于学术分享