8个让我的AI代理真正工作的Python库

我希望早点发现的技术栈

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

需要记忆？"我只是实现一个向量存储包装器。" 想要重试？"给我一个小时和一些try-except块。"

三周和400行代码后，我有了一个记忆系统，其中有一个bug，重复条目会损坏整个索引。实际的代理逻辑？也许50行。

所以今天，我将分解8个让我的自定义代码过时的库。这些不仅仅是流行的。它们实用、设计简洁，并且优雅地解决了真实代理的痛点。

1. LiteLLM — 因为供应商锁定是傻瓜才会做的事

你的代理使用GPT-4工作。现在你想测试Claude。恭喜，你正在重写你的API层。

from litellm import completion

response = 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 vs Claude Sonnet vs Llama 3 70B。差异是每月847美元。没有LiteLLM，测试这将需要一周的重构。

✔️ 使用它： 多模型测试、供应商回退、成本优化

💁‍♂️ 小贴士： 设置自动回退。当OpenAI宕机时（它会的），你的代理会切换到Anthropic，而无需你醒来。

2. Instructor — 因为LLM会在JSON上撒谎

你要求JSON。你得到带有markdown反引号的JSON。或额外的字段。或看起来像JSON但不是的字符串。

import instructor
from pydantic import BaseModel
from openai import OpenAI

client = instructor.from_openai(OpenAI())

class UserInfo(BaseModel):
    name: str
    age: int

user = 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()函数，有十二个边缘情况。它仍然每周在生产数据上失败一次。现在？四个月零解析错误。

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

💡 见解： 与Pydantic AI完美配对。相同的理念，不同的层次。

3. Tenacity — 因为API会在凌晨3点背叛你

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返回503错误20分钟，而我的"健壮"错误处理只是记录并继续。

✔️ 使用它： 每个外部API调用。每一个。

⚠️ 警告： 不要在4xx错误上重试。你只会更快地被速率限制。

4. Logfire — 因为"处理完成"不是调试

你的代理做了一些奇怪的事情。日志说"处理完成"。你不知道输入和输出之间发生了什么。

import logfire

logfire.configure()

@logfire.instrument("process_user_request")
def handle_request(user_input: str):
    logfire.info("Received input", input=user_input)
    result = agent.run(user_input)
    logfire.info("Agent complete", result=result, tokens_used=result.tokens)
    return result

结构化、可搜索的可观察性。确切地看到你的代理做了什么，LLM返回了什么，以及哪里出了问题。

上个月，我花了3个小时调试为什么代理不断推荐错误的产品。原来检索步骤返回的是过时的数据。有了适当的可观察性，我会在5分钟内看到它。Logfire解决了这个问题。

✔️ 使用它： 生产调试、令牌跟踪、延迟监控

💁‍♂️ 小贴士： 与Pydantic集成。你的结构化数据会自动以完整的类型信息记录。

5. Diskcache — 因为你不需要为所有东西都使用Redis

你的代理不断用相同的输入调用相同的API。你在毫无理由地烧钱并增加延迟。

from diskcache import Cache

cache = 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响应缓存、昂贵计算的记忆化

💡 见解： 多个代理进程可以共享同一个缓存而不会损坏。我通过艰难的方式测试了这一点。

6. Tiktoken — 因为猜测令牌计数是生产事件的开始

你的代理填充上下文直到达到限制，然后崩溃并显示神秘的API错误。或者更糟，默默地截断重要信息。

import tiktoken

enc = 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%的错误率。不要像我一样学习这一点。

7. Rich — 因为打印调试不应该伤害你的眼睛

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

from rich.console import Console
from rich.table import Table
from rich import print_json

console = 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 替换内置的print。一个导入，即时升级。上个月我向客户演示了一个代理，Rich格式化的输出比代理本身得到了更多的赞美。

8. Watchfiles — 因为你的迭代循环正在扼杀你的生产力

你更改提示词，重启代理，等待它初始化，测试，意识到你打错了字，再次重启...

from watchfiles import run_process

def main():
    from my_agent import Agent
    agent = Agent()
    agent.run_interactive()

if name == "main":
    run_process("./", target=main)

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

在使用这个之前，我跟踪了一天的工作流程。我重启了代理40+次。每次重启需要8秒。那是超过5分钟的等待。每天。持续了几周。

✔️ 使用它： 本地开发、提示词迭代、快速原型设计

💁‍♂️ 小贴士： 智能判断哪些更改重要。修改提示词文件，即时重新加载。保存日志文件，不重新加载。

元教训

每个库都做好一件事。这不是偶然的。

当我开始构建代理时，我伸手去拿承诺处理一切的框架。它们复杂、固执己见且脆弱。当某件事出错时，我在调试框架而不是我的代理。

现在我组合小型库。LiteLLM用于模型调用。Instructor用于结构化输出。Tenacity用于重试。每个部分都是可替换的。每个部分都是可理解的。

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