aisuite Agents 快速上手:用统一接口在多个 LLM 上构建多轮工具调用 Agent
2026/9/14 4:07:35 网站建设 项目流程

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-4oanthropic: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)

运行流程大致如下:

  1. aisuite 从will_it_rain的函数签名解析出参数location: strtime_of_day: str,从 docstring 提取参数说明,生成 OpenAI 格式的function工具规范;
  2. 模型看到工具后,返回一次tool_calls请求(而不是直接回答);
  3. aisuite 在内部执行该函数,把返回值包装成tool角色的消息回传给模型;
  4. 模型基于工具结果继续推理,输出最终答案;若轮次达到max_turns则停止。

1.2 关键约定:Schema 从签名与 docstring 自动生成

函数返回值的类型不限,但参数建议用类型注解strintfloatbool等),并在 docstring 中为每个参数写出Args:说明——这些信息会直接映射到生成工具 Schema 的propertiesrequired字段中,直接影响模型调用工具的准确率。

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 编排

对于更长时、更结构化的任务,推荐使用声明式AgentRunner的组合:一次声明 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),核心字段:

字段类型说明
namestrAgent 名称,用于日志、Tracing 与上下文标记
modelstr模型标识,格式<provider>:<model-name>
instructionsstr | None系统指令,Runner 会自动将其作为首条system消息注入
toolslist[Callable]工具函数列表(含 Toolkit 生成的工具)
model_settingsdict透传给模型的额外参数(如temperature
tags/metadatalist[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_turns5工具循环最大轮数(与Clientmax_turns语义一致)
tool_policyNone工具执行策略,见下文第四节
state_store/thread_idNone持久化状态存储,二者必须同时提供
artifact_storeNoneArtifact 存储,用于保存/恢复大对象
run_name/parent_run_id/group_id/tags/metadataNone运行分组与观测元信息
trace_sinks/tracing_disabledNone/False自定义 Tracing 输出

3.3 读懂RunResult

Runner.run_sync返回 RunResult,关键成员:

  • final_output:Agent 的最终输出(优先取最后一条有效内容消息);
  • status:运行状态,取值completed/requires_input/max_turns_exceeded/failed
  • steps:完整步骤列表(agentmodel_responsetool_calltool_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(传RunResultAgent作为目标),无需手动拼接历史消息。

四、生产级治理组件

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_filesread_fileread_file_linessearch_files(风险等级low);
  • 写工具(allow_write=True时暴露):write_fileapply_unified_diffapply_patchreplace_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_statusgit status --short --branch)与git_diff(支持pathstaged参数),只读、沙箱化。

shell —— Shell 执行工具(aisuite/toolkits/shell.py):面向需要执行命令的场景,风险等级最高,务必配合工具策略使用。

工具本身通过tool()装饰器(aisuite/agents/policies.py)挂载ToolMetadata,其中categoryrisk_levelcapabilitiesrequires_approval字段会作为策略决策与 Tracing 的依据。

4.2 工具策略(Tool Policies)

策略接收一个 ToolPolicyContext(内含agent_nametool_nameargumentstagsmetadatamessages等),返回boolToolPolicyDecision(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_storethread_id;再次运行时,直接以相同thread_id调用Runner.continue_run(agent, input, state_store=..., thread_id=...)即可从上次状态续跑。注意:state_storethread_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(记录refcreated_at,并自动计算sha256摘要)。

Artifact提供text()方法便捷读取文本内容。Artifact 与状态存储配合使用时,Runner会自动对消息中的大对象做"脱水/回水"(dehydrate / hydrate),避免把二进制内容直接塞进会话状态。

4.5 Tracing:每次运行都可观测

每个RunResult都携带trace_id、完整steps与原始响应。除了print_trace()/write_trace_jsonl()之外,Runner会向配置的 Trace Sinks 发射run.startedmodel.sendmodel.responsetool.allowedtool.deniedtool.completedtool.failedrun.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),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询