返回市场
快速代理

快速代理

作者:evalstate3464 星标更新:2025-11-24

项目介绍

<p align="center"> <a href="https://pypi.org/project/fast-agent-mcp/"><img src="https://img.shields.io/pypi/v/fast-agent-mcp?color=%2334D058&label=pypi" /></a> <a href="#"><img src="https://github.com/evalstate/fast-agent/actions/workflows/main-checks.yml/badge.svg" /></a> <a href="https://github.com/evalstate/fast-agent/issues"><img src="https://img.shields.io/github/issues-raw/evalstate/fast-agent" /></a> <a href="https://discord.gg/xg5cJ7ndN6"><img src="https://img.shields.io/discord/1358470293990936787" alt="discord" /></a> <img alt="Pepy Total Downloads" src="https://img.shields.io/pepy/dt/fast-agent-mcp?label=pypi%20%7C%20downloads"/> <a href="https://github.com/evalstate/fast-agent-mcp/blob/main/LICENSE"><img src="https://img.shields.io/pypi/l/fast-agent-mcp" /></a> </p>

概述

[!TIP] 请参阅: https://fast-agent.ai 获取最新文档。这里也有一个 LLMs.txt 文件 这里

fast-agent 允许您在几分钟内创建和与复杂的多模态代理和工作流进行交互。它是第一个具有完整的端到端测试的 MCP 特性支持框架,包括采样和激发。

<!-- ![multi_model_trim](https://github.com/user-attachments/assets/c8bf7474-2c41-4ef3-8924-06e29907d7c6) -->

简单的声明式语法让您专注于组合您的提示和 MCP 服务器以构建有效的代理

模型支持是全面的,原生支持 Anthropic、OpenAI 和 Google 提供商,以及通过 TensorZero 支持 Azure、Ollama、Deepseek 和数十个其他提供商。结构化输出、PDF 和视觉支持简单易用且经过充分测试。传递和回放 LLM 使应用程序的 Python 胶水代码快速开发和测试成为可能。

最近的功能包括:

  • 代理技能(SKILL.md)
  • MCP-UI 支持 |
  • OpenAI 应用程序 SDK(Skybridge)
  • 命令行模式
  • 高级 MCP 传输诊断
  • MCP 激发
<img width="800" alt="MCP 传输诊断" src="https://gips3.baidu.com/it/u=3564399581,4159974273&fm=3081&app=3081&f=PNG?w=2313&h=1477" />

fast-agent 是唯一允许您检查可流式传输的 HTTP 传输使用的工具——这是确保可靠、合规部署的关键特性。OAuth 支持使用 KeyRing 存储密钥。使用 fast-agent auth 命令进行管理。

[!IMPORTANT]

fast-agent 快速代理文档仓库在此处: https://github.com/evalstate/fast-agent-docs。请随时提交有关文档、经验报告或其他您认为他人可能会发现有用的 PR。所有帮助和反馈都热烈欢迎。

代理应用开发

定义代理应用的提示和配置存储在简单的文件中,几乎没有样板代码,便于简单管理和版本控制。

在工作流执行前后与单个代理和组件聊天,以调整和诊断您的应用。代理可以请求人类输入以获取任务完成所需的额外上下文。

简单的模型选择使得测试模型 <-> MCP 服务器交互变得无痛。您可以在这里了解更多关于这个项目背后的想法 这里

2025-03-23-fast-agent

开始使用:

首先安装 uv 包管理器 用于 Python。然后:

uv pip install fast-agent-mcp          # 安装 fast-agent!
fast-agent go                          # 启动交互会话
fast-agent go --url https://hf.co/mcp  # 使用远程 MCP
fast-agent go --model=generic.qwen2.5  # 使用 ollama qwen 2.5
fast-agent setup                       # 创建示例代理和配置文件
uv run agent.py                        # 运行您的第一个代理
uv run agent.py --model=o3-mini.low    # 指定一个模型
uv run agent.py --transport http --port 8001  # 作为 MCP 服务器公开(隐含服务器模式)
fast-agent quickstart workflow  # 创建“构建有效代理”示例

--server 为了向后兼容仍然可用,但已弃用;--transport 现在自动将代理切换到服务器模式。

其他快速入门示例包括研究者代理(带有评估器-优化器工作流)和数据分析代理(类似于 ChatGPT 的体验),展示了 MCP 根的支持。

[!TIP] Windows 用户 - 对于文件系统和 Docker MCP 服务器,需要对配置进行一些更改 - 所需的更改在配置文件中有详细说明。

基本代理

定义一个代理就像这样简单:

@fast.agent(
  instruction="给定一个对象,仅回应其大小的估计值。"
)

然后我们可以向代理发送消息:

async with fast.run() as agent:
  moon_size = await agent("the moon")
  print(moon_size)

或者开始与代理进行交互式聊天:

async with fast.run() as agent:
  await agent.interactive()

这里是完整的 sizer.py 代理应用,带有一些样板代码:

import asyncio
from fast_agent import FastAgent

# 创建应用
fast = FastAgent("代理示例")

@fast.agent(
  instruction="给定一个对象,仅回应其大小的估计值。"
)
async def main():
  async with fast.run() as agent:
    await agent.interactive()

if __name__ == "__main__":
    asyncio.run(main())

该代理可以通过运行 uv run sizer.py 来启动。

使用 --model 开关指定一个模型 - 例如 uv run sizer.py --model sonnet

组合代理和使用 MCP 服务器

要生成示例,请使用 fast-agent quickstart workflow。此示例可以通过 uv run workflow/chaining.py 运行。fast-agent 在当前目录查找配置文件,如果未找到,则递归检查父目录。

代理可以串联起来构建一个工作流,使用在 fastagent.config.yaml 文件中定义的 MCP 服务器:

@fast.agent(
    "url_fetcher",
    "给定一个 URL,提供一个完整而全面的摘要。",
    servers=["fetch"], # 在 fastagent.config.yaml 中定义的 MCP 服务器名称
)
@fast.agent(
    "social_media",
    """
    为任何给定的文本编写一条 280 字符的社交媒体帖子。
    仅回应帖子,永远不使用标签。
    """,
)
@fast.chain(
    name="post_writer",
    sequence=["url_fetcher", "social_media"],
)
async def main():
    async with fast.run() as agent:
        # 使用链工作流
        await agent.post_writer("http://llmindset.co.uk")

所有代理和工作流都响应 .send("message").prompt() 以开始聊天会话。

保存为 social.py,我们现在可以从命令行运行此工作流:

uv run workflow/chaining.py --agent post_writer --message "<url>"

添加 --quiet 开关以禁用进度和消息显示,并仅返回最终响应 - 对于简单的自动化非常有用。

MCP OAuth (v2.1)

对于 SSE 和 HTTP MCP 服务器,默认启用 OAuth,只需最少的配置。使用本地回调服务器捕获授权码,如果端口不可用,可以使用粘贴 URL 作为备用。

  • 每个服务器在 fastagent.config.yaml 中的最小设置:
mcp:
  servers:
    myserver:
      transport: http # 或 sse
      url: http://localhost:8001/mcp # 或 /sse 对于 SSE 服务器
      auth:
        oauth: true # 默认:true
        redirect_port: 3030 # 默认:3030
        redirect_path: /callback # 默认:/callback
        # scope: "user"       # 可选;如果省略,使用服务器默认值
  • OAuth 客户端使用 PKCE 和内存令牌存储(不写入磁盘)。
  • 令牌持久性:默认情况下,令牌安全地存储在您的操作系统密钥链中,通过 keyring。如果密钥链不可用(例如,在无头容器中),则使用会话中的内存存储。
  • 若要强制每个服务器仅使用内存存储,设置:
mcp:
  servers:
    myserver:
      transport: http
      url: http://localhost:8001/mcp
      auth:
        oauth: true
        persist: memory
  • 若要禁用特定服务器的 OAuth,将 auth.oauth: false 设置为该服务器。

工作流

chain 工作流提供了更声明式的顺序调用代理的方法:


@fast.chain(
  "post_writer",
   sequence=["url_fetcher","social_media"]
)

# 我们可以直接提示它:
async with fast.run() as agent:
  await agent.post_writer()

这启动了一个交互式会话,为给定的 URL 产生一个简短的社交媒体帖子。如果提示了 chain,它将返回到链中最后一个代理的聊天。您可以通过键入 @agent-name 切换要提示的代理。

链可以被纳入其他工作流,或包含其他工作流元素(包括其他链)。您可以设置一个 instruction 来精确描述其能力,以便其他工作流步骤使用。

人类输入

代理可以请求人类输入来协助任务或获取额外的上下文:

@fast.agent(
    instruction="一个辅助基本任务的人工智能代理。需要时请求人类输入。",
    human_input=True,
)

await agent("打印序列中的下一个数字")

在示例 human_input.py 中,代理将提示用户以获取完成任务所需的更多信息。

并行

并行工作流同时向多个代理发送相同的消息(扇出),然后使用 扇入 代理处理合并的内容。

@fast.agent("translate_fr", "将文本翻译成法语")
@fast.agent("translate_de", "将文本翻译成德语")
@fast.agent("translate_es", "将文本翻译成西班牙语")

@fast.parallel(
  name="translate",
  fan_out=["translate_fr","translate_de","translate_es"]
)

@fast.chain(
  "post_writer",
   sequence=["url_fetcher","social_media","translate"]
)

如果您没有指定 fan-in 代理,parallel 将直接返回组合代理的结果。

parallel 还可用于从不同的 LLM 组合想法。

当在其他工作流中使用 parallel 时,指定一个 instruction 来描述其操作。

评估器-优化器

评估器-优化器结合了两个代理:一个生成内容(生成器),另一个评估该内容并提供可操作的反馈(评估器)。消息首先发送给生成器,然后这对代理在一个循环中运行,直到评估器满意质量或达到最大改进次数为止。生成器的最终结果被返回。

如果生成器关闭了 use_history,在请求改进时会返回上一次迭代的结果 - 否则会使用对话上下文。

@fast.evaluator_optimizer(
  name="researcher",
  generator="web_searcher",
  evaluator="quality_assurance",
  min_rating="EXCELLENT",
  max_refinements=3
)

async with fast.run() as agent:
  await agent.researcher.send("制作一份如何制作完美意式咖啡的报告")

当在工作流中使用时,它返回最后 generator 消息作为结果。

查看 evaluator.py 工作流示例,或使用 fast-agent quickstart researcher 查看更完整的示例。

路由器

路由器使用 LLM 来评估消息,并将其路由到最合适的代理。路由提示根据代理指令和可用服务器自动生成。

@fast.router(
  name="route",
  agents=["agent1","agent2","agent3"]
)

查看 router.py 工作流示例。

编排器

给定一个复杂任务,编排器使用 LLM 生成一个计划,将任务分配给可用的代理。编排器生成规划和聚合提示,这得益于使用更强大的模型。计划可以在开始时一次性构建(plan_type="full")或迭代构建(plan_type="iterative")。

@fast.orchestrator(
  name="orchestrate",
  agents=["task1","task2","task3"]
)

查看 orchestrator.pyagent_build.py 工作流示例。

代理特性

调用代理

所有定义都可以省略名称和指令参数以简化:

@fast.agent("您是一个有帮助的代理")          # 创建一个具有默认名称的代理。
@fast.agent("greeter","愉快地回应!")    # 创建一个名为 "greeter" 的代理

moon_size = await agent("the moon")             # 使用消息调用默认(首先定义的代理)

result = await agent.greeter("早上好!")   # 使用点表示法按名称向代理发送消息
result = await agent.greeter.send("你好!")     # 显式调用 'send'

await agent.greeter()                           # 如果未指定消息,将打开一个聊天会话
await agent.greeter.prompt()                    # 可以使其更明确
await agent.greeter.prompt(default_prompt="OK") # 支持设置默认提示

agent["greeter"].send("晚上好!")          # 如果偏好,支持字典访问

定义代理

基本代理

@fast.agent(
  name="agent",                          # 代理的名称
  instruction="您是一个有帮助的代理", # 代理的基本指令
  servers=["filesystem"],                # 代理的 MCP 服务器列表
  model="o3-mini.high",                  # 为代理指定一个模型
  use_history=True,                      # 代理维护聊天历史
  request_params=RequestParams(temperature= 0.7), # LLM 的附加参数(或 RequestParams())
  human_input=True,                      # 代理可以请求人类输入
)

@fast.chain(
  name="chain",                          # 链的名称
  sequence=["agent1", "agent2", ...],    # 按执行顺序列出的代理
  instruction="instruction",             # 描述链的指令,供其他工作流使用
  cumulative=False,                      # 是否在链中累积消息
  continue_with_final=True,              # 在提示后打开链尾部代理的聊天会话
)

并行

@fast.parallel(
  name="parallel",                       # 并行工作流的名称
  fan_out=["agent1", "agent2"],          # 并行运行的代理列表
  fan_in="aggregator",                   # 结合结果的代理名称(可选)
  instruction="instruction",             # 描述并行工作流的指令,供其他工作流使用
  include_request=True,                  # 在扇入消息中包含原始请求
)

评估器-优化器

@fast.evaluator_optimizer(
  name="researcher",                     # 工作流的名称
  generator="web_searcher",              # 内容生成器代理的名称
  evaluator="quality_assurance",         # 评估器代理的名称
  min_rating="GOOD",                     # 最低可接受的质量(EXCELLENT, GOOD, FAIR, POOR)
  max_refinements=3,                     # 最大改进迭代次数
)

路由器

@fast.router(
  name="route",                          # 路由器的名称
  agents=["agent1", "agent2", "agent3"], # 路由器可以委托的代理名称列表
  model="o_3-mini.high",                 # 指定路由模型
  use_history=False,                     # 路由器维护对话历史
  human_input=False,                     # 路由器是否可以请求人类输入
)

编排器

@fast.orchestrator(
  name="orchestrator",                   # 编排器的名称
  instruction="instruction",             # 编排器的基本指令
  agents=["agent1", "agent2"],           # 编排器可以使用的代理名称列表
  model="o3-mini.high",                  # 指定编排器规划模型
  use_history=False,                     # 编排器不维护聊天历史(无影响)。
  human_input=False,                     # 编排器是否可以请求人类输入
  plan_type="full",                      # 规划方法:"full" 或 "iterative"
  plan_iterations=5,                     # 最大全计划尝试次数或迭代次数
)

多模态支持

使用内置的 prompt-server 或直接使用 MCP 类型向提示添加资源。提供了方便的类来简化这一过程,例如:

  summary: str =  await agent.with_resource(
      "请总结这份 PDF",
      "mcp_server",
      "resource://fast-agent/sample.pdf",
  )

MCP 工具结果转换

LLM API 对通过其聊天完成 API 返回的工具调用/函数结果的内容类型有限制:

  • OpenAI 支持文本
  • Anthropic 支持文本和图像

对于 MCP