aisuite Agents 快速上手:用统一接口在多个 LLM 上构建多轮工具调用 Agent
【免费下载链接】aisuiteSimple, unified interface to multiple Generative AI providers项目地址: https://gitcode.com/GitHub_Trending/ai/aisuite
本篇指南以 aisuite 的 Agents 能力为核心,讲解如何让模型调用真实 Python 函数、自动生成工具 Schema、执行多轮工具循环,并组合 Toolkits、MCP 服务器、工具策略(Tool Policies)、状态存储(State Stores)与 Artifacts,搭建一套可用于生产环境的 Agent 编排管线。读完你将掌握max_turns自动工具循环、手动工具处理、Agent+Runner声明式开发,以及 MCP 工具的接入与安全管理。
本文假设你已按 Chat Completions 快速入门 完成 aisuite 安装并配置好至少一个提供商的 API Key。模型名采用<provider>:<model-name>格式,如openai:gpt-4o、anthropic:claude-sonnet-4-6。
一、自动工具调用:把普通 Python 函数直接交给模型
aisuite 最核心的 Agent 能力是:你只需传入普通 Python 函数,aisuite 会根据函数签名和 docstring 自动生成工具 Schema,负责执行模型发起的调用,并把执行结果反馈给模型,如此循环直到模型给出最终回答(或达到max_turns上限)。
1.1 最小可运行示例
import aisuite as ai def will_it_rain(location: str, time_of_day: str): """Check if it will rain in a location at a given time today. Args: location (str): Name of the city time_of_day (str): Time of the day in HH:MM format. """ return "YES" client = ai.Client() response = client.chat.completions.create( model="openai:gpt-4o", messages=[{ "role": "user", "content": "I live in San Francisco. Can you check for weather " "and plan an outdoor picnic for me at 2pm?" }], tools=[will_it_rain], max_turns=2 ) print(response.choices[0].message.content)运行流程大致如下:
- aisuite 从
will_it_rain的函数签名解析出参数location: str、time_of_day: str,从 docstring 提取参数说明,生成 OpenAI 格式的function工具规范; - 模型看到工具后,返回一次
tool_calls请求(而不是直接回答); - aisuite 在内部执行该函数,把返回值包装成
tool角色的消息回传给模型; - 模型基于工具结果继续推理,输出最终答案;若轮次达到
max_turns则停止。
1.2 关键约定:Schema 从签名与 docstring 自动生成
函数返回值的类型不限,但参数建议用类型注解(str、int、float、bool等),并在 docstring 中为每个参数写出Args:说明——这些信息会直接映射到生成工具 Schema 的properties与required字段中,直接影响模型调用工具的准确率。
1.3 用intermediate_messages续接对话
一次带工具调用的响应中,response.choices[0].intermediate_messages保存了完整的工具交互历史(模型发起的每次工具调用请求 + 每次工具执行结果),类型定义见 aisuite/framework/choice.py。想继续同一段对话时,把它追加进messages传给下一次create即可,模型就能感知之前的工具调用上下文:
all_messages = messages + list(response.choices[0].intermediate_messages) response2 = client.chat.completions.create( model="openai:gpt-4o", messages=all_messages, tools=[will_it_rain], max_turns=2, )二、手动工具处理:完全掌控工具循环
如果不传max_turns,aisuite不会替你执行任何工具,而是把模型的工具调用请求原样返回,由你自行执行、校验或过滤。此时tools需要传 OpenAI 格式的 JSON 工具规范:
tools = [{ "type": "function", "function": { "name": "will_it_rain", "description": "Check if it will rain in a location at a given time today", "parameters": { "type": "object", "properties": { "location": {"type": "string", "description": "Name of the city"}, "time_of_day": {"type": "string", "description": "Time of the day in HH:MM format."} }, "required": ["location", "time_of_day"] } } }] response = client.chat.completions.create( model="openai:gpt-4o", messages=messages, tools=tools )拿到响应后,从response.choices[0].message.tool_calls读取模型的调用请求,自行决定执行策略:自定义错误处理(如工具抛异常时返回特定提示)、选择性执行(如只执行白名单内的工具)、或接入既有工具管线。
两种风格的对比与选择:
| 方式 | 适用场景 | 特点 |
|---|---|---|
max_turns自动循环 | 快速原型、工具可信、追求开发效率 | aisuite 自动生成 Schema、自动执行并回填结果 |
| 手动处理 | 需要权限校验、审计、自定义失败逻辑 | 完全掌控循环,工具 Schema 需要手写 |
两种方式的完整可运行示例见 examples/tool_calling_abstraction.ipynb。
三、Agents API:声明式 Agent + Runner 编排
对于更长时、更结构化的任务,推荐使用声明式Agent与Runner的组合:一次声明 Agent,反复运行,并挂载工具策略、状态存储、Artifacts 与 Tracing 等生产级能力。
3.1 最小示例
import aisuite as ai from aisuite import Agent, Runner agent = Agent( name="repo-helper", model="anthropic:claude-sonnet-4-6", instructions="You are a careful repo assistant. Use your tools to answer from the code.", tools=[*ai.toolkits.files(root="."), *ai.toolkits.git(root=".")], ) result = Runner.run_sync(agent, "What changed in the last commit? Summarize in 3 bullets.") print(result.final_output)Agent是一个声明式数据类(定义见 aisuite/agents/types.py),核心字段:
| 字段 | 类型 | 说明 |
|---|---|---|
name | str | Agent 名称,用于日志、Tracing 与上下文标记 |
model | str | 模型标识,格式<provider>:<model-name> |
instructions | str | None | 系统指令,Runner 会自动将其作为首条system消息注入 |
tools | list[Callable] | 工具函数列表(含 Toolkit 生成的工具) |
model_settings | dict | 透传给模型的额外参数(如temperature) |
tags/metadata | list[str]/dict | 用于运行分组与观测的元信息 |
3.2 Runner:同步与异步两种入口
aisuite/agents/runner.py 中的Runner提供静态方法:
Runner.run(agent, input, ...):异步运行(对应client.chat.completions.acreate);Runner.run_sync(agent, input, ...):同步包装。当从已有事件循环内(如 Jupyter Notebook)调用时,会自动借助nest_asyncio兜底;未安装相关依赖时会给出明确提示。
input可以是普通字符串、消息列表,也可以是RunState(用于续跑已持久化的会话)。常见参数:
| 参数 | 默认值 | 说明 |
|---|---|---|
max_turns | 5 | 工具循环最大轮数(与Client的max_turns语义一致) |
tool_policy | None | 工具执行策略,见下文第四节 |
state_store/thread_id | None | 持久化状态存储,二者必须同时提供 |
artifact_store | None | Artifact 存储,用于保存/恢复大对象 |
run_name/parent_run_id/group_id/tags/metadata | None | 运行分组与观测元信息 |
trace_sinks/tracing_disabled | None/False | 自定义 Tracing 输出 |
3.3 读懂RunResult
Runner.run_sync返回 RunResult,关键成员:
final_output:Agent 的最终输出(优先取最后一条有效内容消息);status:运行状态,取值completed/requires_input/max_turns_exceeded/failed;steps:完整步骤列表(agent、model_response、tool_call、tool_result等),每一步含trace_id、时间戳与数据;messages:运行后的完整消息历史;new_items:本轮新增消息;raw_responses:底层每次模型响应(含intermediate_responses);trace_id:整次运行的唯一追踪 ID。
RunResult还内置了可观测方法:print_trace()在终端打印结构化步骤摘要;write_trace_jsonl(path)将运行追踪追加写入 JSONL 文件。继续对话时可调用Runner.continue_run/Runner.continue_sync(传RunResult或Agent作为目标),无需手动拼接历史消息。
四、生产级治理组件
4.1 Toolkits:开箱即用的沙箱工具族
ai.toolkits提供三组预构建工具(见 aisuite/toolkits/init.py):
files —— 文件系统工具(aisuite/toolkits/files.py)
ai.toolkits.files( root=".", # 根目录,所有相对路径都基于它解析 allow_write=False, # 为 True 时额外暴露写工具 max_read_bytes=200_000, # 单文件最大读取字节数 max_search_bytes=1_000_000, # 单次搜索累计扫描上限 ignore=[".git", ".venv", "node_modules"], # 覆盖默认忽略列表 )- 只读工具:
list_files、read_file、read_file_lines、search_files(风险等级low); - 写工具(
allow_write=True时暴露):write_file、apply_unified_diff、apply_patch、replace_in_file,这些工具均标记risk_level="medium"且requires_approval=True,便于与审批策略联动; - 支持多根目录
roots=[{"path": ..., "writable": True}, ...],越界访问会被PermissionError拒绝。
git —— 只读 Git 工具(aisuite/toolkits/git.py)
ai.toolkits.git(root=".", max_output_chars=20000)提供git_status(git status --short --branch)与git_diff(支持path与staged参数),只读、沙箱化。
shell —— Shell 执行工具(aisuite/toolkits/shell.py):面向需要执行命令的场景,风险等级最高,务必配合工具策略使用。
工具本身通过tool()装饰器(aisuite/agents/policies.py)挂载ToolMetadata,其中category、risk_level、capabilities、requires_approval字段会作为策略决策与 Tracing 的依据。
4.2 工具策略(Tool Policies)
策略接收一个 ToolPolicyContext(内含agent_name、tool_name、arguments、tags、metadata、messages等),返回bool或ToolPolicyDecision(allowed, reason)。内置策略:
AllowAllToolPolicy:放行所有工具;DenyAllToolPolicy(reason=None):拒绝所有工具;AllowToolsPolicy(allowed_tools, reason=None):仅放行白名单内的工具名;RequireApprovalPolicy(callback):回调返回True放行、False拒绝,也可直接返回ToolPolicyDecision——适合接入人工审批界面或外部审批系统。
也支持任意满足evaluate(context) -> bool | ToolPolicyDecision的可调用对象(ToolPolicy为 Protocol 类型,见 aisuite/agents/types.py)。策略通过Runner.run_sync(agent, input, tool_policy=...)注入,被拒绝的工具调用会记录为tool.denied追踪事件。
4.3 状态存储(State Stores):跨进程持久化与续跑
存储协议(aisuite/agents/state_store.py)要求实现save_state/load_state/delete_state,并提供带乐观锁的revision机制(冲突时抛StateConflictError)。内置实现:
InMemoryStateStore():进程内字典,适合单进程测试;FileStateStore(root=".aisuite/state"):按thread_id落盘为 JSON 文件,写入采用临时文件 +os.replace的原子替换方式;PostgresStateStore:生产环境推荐,支持并发与持久化(见 aisuite/agents/postgres_state_store.py,含CompactionRecord压缩记录)。
使用方式:首次运行传入state_store与thread_id;再次运行时,直接以相同thread_id调用Runner.continue_run(agent, input, state_store=..., thread_id=...)即可从上次状态续跑。注意:state_store与thread_id必须成对出现,且新线程名冲突会抛ThreadAlreadyExistsError,缺失线程会抛StateNotFoundError。
4.4 Artifacts:Agent 产物的存取
aisuite/agents/artifact_store.py 定义了ArtifactStore协议(put/get/delete)与两种实现:
InMemoryArtifactStore():内存存储;FileArtifactStore(root=".aisuite/artifacts"):每个 Artifact 一个目录,含data文件与metadata.json(记录ref、created_at,并自动计算sha256摘要)。
Artifact提供text()方法便捷读取文本内容。Artifact 与状态存储配合使用时,Runner会自动对消息中的大对象做"脱水/回水"(dehydrate / hydrate),避免把二进制内容直接塞进会话状态。
4.5 Tracing:每次运行都可观测
每个RunResult都携带trace_id、完整steps与原始响应。除了print_trace()/write_trace_jsonl()之外,Runner会向配置的 Trace Sinks 发射run.started、model.send、model.response、tool.allowed、tool.denied、tool.completed、tool.failed、run.completed等事件(相关机制见 aisuite/tracing/sinks.py 与 aisuite/tracing/viewer.py),可对接本地观测面板或自建采集链路。
五、MCP 工具:接入 Model Context Protocol 服务器
任何 Model Context Protocol 服务器的工具都可以接入 aisuite(需安装 MCP 支持:pip install 'aisuite[mcp]')。
5.1 内联配置(简单场景)
直接在tools里声明一个type: "mcp"的工具:
response = client.chat.completions.create( model="openai:gpt-4o", messages=[{"role": "user", "content": "List the files in the current directory"}], tools=[{ "type": "mcp", "name": "filesystem", "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/directory"] }], max_turns=3 )5.2 显式 MCPClient(可复用、带安全过滤)
对于需要复用连接、过滤工具或做名称前缀隔离的场景,使用 aisuite/mcp/client.py 中的MCPClient:
from aisuite.mcp import MCPClient mcp = MCPClient( command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/directory"] ) response = client.chat.completions.create( model="openai:gpt-4o", messages=[{"role": "user", "content": "List the files"}], tools=mcp.get_callable_tools(), max_turns=3 ) mcp.close()MCPClient核心方法:
list_tools():查看服务器暴露的全部工具 Schema;get_callable_tools(allowed_tools=None, use_tool_prefix=False):将工具包装成 aisuite 可调用的 Python 函数。allowed_tools只暴露白名单内的工具(安全过滤);use_tool_prefix=True时为工具名加"{client_name}__"前缀,避免多服务器工具名冲突;get_tool(tool_name):按名获取单个工具;close():关闭服务器连接。
MCPClient同时支持 stdio 与 HTTP 两种传输:stdio 传command/args/env,HTTP 传server_url/headers/timeout(默认 30 秒),二者互斥。更多用法见 examples/mcp_tools_example.ipynb。
六、组合示例:把上述能力串成一个生产级 Agent
以下示例演示 Agent + Toolkits + 审批策略 + 文件状态存储 + Artifact 存储 + 手动续跑的完整组合:
import aisuite as ai from aisuite import Agent, Runner from aisuite.agents import ( AllowToolsPolicy, FileStateStore, FileArtifactStore, ) agent = Agent( name="doc-writer", model="openai:gpt-4o", instructions=( "You edit files in the repo. Only use allowed tools. " "Report what you changed." ), tools=[*ai.toolkits.files(root=".", allow_write=True)], ) state_store = FileStateStore(root=".aisuite/state") artifact_store = FileArtifactStore(root=".aisuite/artifacts") policy = AllowToolsPolicy( allowed_tools=["list_files", "read_file", "search_files", "write_file"], reason="only read + write_file are permitted", ) result = Runner.run_sync( agent, "Add a NOTES.md describing the repo layout, then summarize.", tool_policy=policy, state_store=state_store, thread_id="doc-writer-1", artifact_store=artifact_store, ) print(result.status, result.trace_id) result.print_trace() # 同一线程续跑,模型能感知上一次的全部工具历史 result2 = Runner.continue_sync( agent, "Now append today's date to NOTES.md.", state_store=state_store, thread_id="doc-writer-1", artifact_store=artifact_store, ) print(result2.final_output)七、进一步学习
- 若尚未安装 aisuite 或未配置 API Key,先阅读 Chat Completions 快速入门;
- 可运行的 Notebook 示例集中在 examples/(工具调用抽象见 examples/tool_calling_abstraction.ipynb,MCP 用法见 examples/mcp_tools_example.ipynb);
- Agent 核心实现源码:Agent/RunResult 类型定义、Runner 运行器、工具策略、状态存储、Artifact 存储、MCP 客户端;
- 配套测试可参考 tests/agents/ 与 tests/mcp/,覆盖 Agent 集成流程、状态续跑、工具策略与 MCP 端到端调用;
- 想要现成的桌面 AI 协作工具而非自己搭建,可参考 OpenWorker 快速入门,其完整源码位于 platform/,是使用 aisuite 搭建完整 Agent 编排系统的可运行参考实现。
【免费下载链接】aisuiteSimple, unified interface to multiple Generative AI providers项目地址: https://gitcode.com/GitHub_Trending/ai/aisuite
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考