1. 项目概述:一个轻量级、可扩展的AI智能体框架

最近在AI应用开发领域,一个名为“Kani”的开源项目引起了我的注意。它不是一个具体的应用,而是一个框架,一个专门用于构建和运行AI智能体(AI Agent)的Python库。简单来说,Kani提供了一个结构化的“脚手架”,让开发者能够更轻松地创建出那些能够理解复杂指令、调用工具、并持续与环境交互的智能程序。如果你对让AI不只是聊天,而是能帮你订餐、分析数据、甚至管理服务器这类任务感兴趣,那么理解Kani的运作机制将非常有价值。

在当前的AI浪潮中,大语言模型(LLM)的能力已经毋庸置疑,但如何让这些“大脑”真正“动手做事”,一直是应用落地的关键瓶颈。直接调用API得到的往往是文本回复,而我们需要的是能执行一连串动作的智能体。Kani正是为了解决这个问题而生。它抽象了智能体交互中的核心循环——接收用户输入、模型思考、决定行动、执行工具、整合结果、再次思考——并将这个过程标准化、模块化。这意味着,开发者无需从零开始处理状态管理、上下文窗口、函数调用等繁琐细节,可以专注于定义智能体的“技能”(工具)和“个性”(提示词工程)。

这个框架特别适合两类人:一是希望快速原型化一个AI智能体应用的开发者;二是研究智能体行为、需要稳定实验平台的研究者。它内置了对主流模型(如OpenAI GPT、Anthropic Claude、本地LM Studio服务器)的支持,并且设计上强调轻量级和可扩展性,你可以在几分钟内启动一个基础智能体,也可以基于其引擎构建高度定制化的复杂多智能体系统。接下来,我将深入拆解Kani的设计哲学、核心组件,并分享如何从零开始构建一个具备实际功能的智能体。

2. 核心架构与设计哲学拆解

要理解Kani,不能只看代码,得先理解它背后要解决的根本问题。AI智能体开发不是简单的“问-答”,而是一个有状态的、持续的过程。一个能处理“帮我查查这周天气,如果周末晴天就推荐几个户外活动,并生成一个出行清单”这样任务的智能体,需要分解任务、记忆上下文、调用天气API、进行逻辑判断、最后组织回复。Kani的架构正是围绕优雅地处理这个流程而设计的。

2.1 引擎(Engine):智能体的“中央处理器”

Kani的核心是 Engine 类。你可以把它想象成智能体的运行时环境和调度中心。它不直接是智能体本身,而是驱动智能体运作的“引擎”。当你初始化一个引擎时,你需要至少指定两样东西:一个AI模型(比如 OpenAIClient )和一个系统提示词( system_prompt )。系统提示词定义了智能体的角色、行为准则和初始知识,这就像是给智能体注入灵魂。

引擎的核心职责是管理“对话轮次”(Turn)。每一轮交互中,引擎会做以下几件事:

  1. 组装上下文 :它自动维护一个对话历史列表,并智能地处理上下文窗口限制。当对话历史太长,超出模型令牌限制时,Kani会采用一种“流式摘要”或“滑动窗口”策略(取决于配置),将遥远的记忆进行摘要,保留最近的关键对话,确保模型始终在有效的上下文内工作。这是手动处理时极易出错的地方。
  2. 调用模型 :将组装好的上下文(系统提示+历史摘要+最新用户消息)发送给指定的AI模型。
  3. 解析函数调用 :如果模型的回复中包含一个或多个工具调用请求(例如, {"name": "get_weather", "arguments": {"city": "Beijing"}} ),引擎会解析这些请求。
  4. 分派工具执行 :引擎在其已知的“工具库”中查找对应的函数,并以正确的参数执行它。
  5. 生成下一步指令 :将工具执行的结果(成功或失败)作为新的消息追加到上下文,然后再次调用模型,让模型基于工具结果进行下一步的思考或生成最终回复给用户。

这个过程循环往复,直到模型决定不再调用工具,直接给出面向用户的答案为止。引擎抽象了这个循环,开发者只需定义好工具和提示词。

注意 :Kani的引擎设计是“无状态”管理“有状态”对话。引擎对象本身是状态化的,它保存了对话历史。但你的工具函数应该是相对纯净的,避免在工具内部维护复杂的全局状态,这有利于智能体的可靠性和可测试性。

2.2 函数(工具)与装饰器:赋予智能体“手脚”

智能体光会思考不行,必须能行动。在Kani中,行动的能力通过“函数”(也称为工具)来体现。任何Python函数都可以通过一个简单的装饰器 @ai_function() 注册为智能体可用的工具。

这个装饰器会做一件关键事情:根据函数的名称、文档字符串(docstring)和参数注解,自动生成一个符合OpenAI Function Calling格式的JSON Schema描述。这个描述会被注入到每次请求模型的系统提示中,告诉模型“你现在拥有以下可用的功能”。例如:

from kani import ai_function

@ai_function()
def get_current_time(timezone: str = "UTC") -> str:
    """Get the current time in the specified timezone.
    
    Args:
        timezone: The IANA timezone name (e.g., "America/New_York", "Asia/Shanghai"). Defaults to "UTC".
    """
    from datetime import datetime
    import pytz
    tz = pytz.timezone(timezone)
    return datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S %Z")

当你把这个函数添加到引擎后,模型在思考时,就可能生成一个调用 get_current_time 的请求。引擎会自动将JSON参数映射到Python函数参数(这里 timezone 会映射为 "Asia/Shanghai" ),执行函数,并将返回值(一个时间字符串)反馈给模型。

这里有一个至关重要的实操细节 :函数文档字符串( """Get the current time...""" )的质量直接决定了模型调用工具的准确性。文档字符串需要清晰、无歧义地描述函数的功能、每个参数的意义和格式、以及返回值的含义。模型完全依赖这段文本来理解何时以及如何调用该工具。含糊的文档会导致模型错误调用或拒绝调用。

2.3 消息与状态管理:智能体的“记忆”

Kani内部使用 ChatMessage 对象来表示对话中的每一条消息。消息有不同的角色: System (系统)、 User (用户)、 Assistant (助手)、 FunctionCall (函数调用请求)、 FunctionReturn (函数返回结果)。引擎负责将这些消息对象组织成一个有序的列表,即对话历史。

状态管理的一个高级特性是 Kani 类本身。如果说 Engine 是通用引擎,那么 Kani 类就是一个预配置好的、具有特定身份和工具的“智能体实例”。你可以通过继承 Kani 类来创建不同类型的智能体,每个子类可以有自己的默认系统提示词、一组绑定的工具函数、以及自定义的行为钩子(例如,在每次模型调用前修改消息)。

例如,你可以创建一个 ChefKani 类,其系统提示是“你是一位法国大厨”,并绑定 search_recipe (搜索菜谱)、 convert_units (转换单位)等工具。这种面向对象的方式使得创建和管理多个各司其职的智能体变得非常清晰。

3. 从零构建一个天气出行助手智能体

理论说得再多,不如动手做一个。我们来构建一个实用的智能体:天气出行助手。它的功能是:用户告诉它一个城市和日期,它能查询天气,并根据天气情况推荐活动,最后生成一个简单的出行建议清单。

3.1 环境准备与依赖安装

首先,确保你的Python环境在3.8以上。创建一个新的虚拟环境是一个好习惯。

# 创建并激活虚拟环境(可选,但推荐)
python -m venv kani-env
source kani-env/bin/activate  # Linux/macOS
# kani-env\Scripts\activate  # Windows

# 安装Kani核心库及我们需要的额外依赖
pip install kani[openai]  # 安装Kani并包含OpenAI支持
pip install requests pytz  # 用于HTTP请求和时区处理

这里我们选择 kani[openai] 变体,因为它包含了与OpenAI API交互所需的客户端。如果你使用其他模型(如通过LM Studio部署的本地模型),则需要安装对应的依赖,如 kani[lmstudio]

3.2 定义核心工具函数

智能体的能力源于其工具。我们需要三个工具:查询天气、推荐活动、生成清单。我们将使用一个免费的开放天气API(例如Open-Meteo)来获取数据。

import requests
from datetime import datetime
from kani import ai_function
from typing import List, Optional

# 工具1:查询天气
@ai_function()
def get_weather_forecast(city: str, date: str) -> dict:
    """Get the weather forecast for a specific city and date.
    
    Args:
        city: The name of the city (e.g., "London", "Tokyo").
        date: The date in YYYY-MM-DD format (e.g., "2024-05-20"). Only supports forecasts up to 7 days ahead.
    
    Returns:
        A dictionary containing weather details. Example:
        {
            "city": "London",
            "date": "2024-05-20",
            "condition": "Sunny",
            "max_temp_c": 22.5,
            "min_temp_c": 12.0,
            "precipitation_mm": 0.0
        }
    """
    # 这里需要替换为真实的API调用。以Open-Meteo为例(需先获取坐标,此处简化)
    # 实际应用中,你应该先通过一个地理编码API将城市名转换为经纬度。
    base_url = "https://api.open-meteo.com/v1/forecast"
    # 假设我们有一个简单的城市到坐标的映射(生产环境应用更健壮的方法)
    city_coords = {
        "london": {"latitude": 51.5074, "longitude": -0.1278},
        "tokyo": {"latitude": 35.6762, "longitude": 139.6503},
        "new york": {"latitude": 40.7128, "longitude": -74.0060},
        "beijing": {"latitude": 39.9042, "longitude": 116.4074},
    }
    
    city_lower = city.lower()
    if city_lower not in city_coords:
        return {"error": f"Coordinates for city '{city}' not found in database."}
    
    coords = city_coords[city_lower]
    params = {
        "latitude": coords["latitude"],
        "longitude": coords["longitude"],
        "daily": ["weather_code", "temperature_2m_max", "temperature_2m_min", "precipitation_sum"],
        "timezone": "auto",
        "start_date": date,
        "end_date": date
    }
    
    try:
        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # 简化解析过程,实际应根据API响应结构调整
        daily = data.get("daily", {})
        # 将天气代码转换为文字描述(WMO代码)
        wmo_code = daily.get("weather_code", [0])[0]
        condition_map = {
            0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy",
            3: "Overcast", 45: "Foggy", 61: "Light rain"
        }
        condition = condition_map.get(wmo_code, "Unknown")
        
        return {
            "city": city,
            "date": date,
            "condition": condition,
            "max_temp_c": daily.get("temperature_2m_max", [0])[0],
            "min_temp_c": daily.get("temperature_2m_min", [0])[0],
            "precipitation_mm": daily.get("precipitation_sum", [0])[0],
        }
    except requests.exceptions.RequestException as e:
        return {"error": f"Failed to fetch weather data: {str(e)}"}

# 工具2:基于天气推荐活动
@ai_function()
def recommend_activities(weather_condition: str, temperature_c: float) -> List[str]:
    """Recommend a list of activities based on weather conditions and temperature.
    
    Args:
        weather_condition: Description of the weather (e.g., "Sunny", "Rainy", "Overcast").
        temperature_c: The maximum temperature in degrees Celsius.
    
    Returns:
        A list of suggested activity strings.
    """
    activities = []
    weather_lower = weather_condition.lower()
    
    if "clear" in weather_lower or "sunny" in weather_lower:
        activities.append("Go for a hike or walk in the park.")
        if temperature_c > 20:
            activities.append("Have a picnic or outdoor barbecue.")
            activities.append("Visit an outdoor swimming pool or beach.")
        elif temperature_c > 10:
            activities.append("Enjoy outdoor cycling or jogging.")
    elif "rain" in weather_lower or "drizzle" in weather_lower:
        activities.append("Visit a museum, art gallery, or library.")
        activities.append("Go to the cinema or watch a movie at home.")
        activities.append("Try indoor rock climbing or visit a cafe with a good book.")
    elif "cloud" in weather_lower or "overcast" in weather_lower:
        activities.append("Explore indoor markets or shopping centers.")
        activities.append("Attend a workshop or cooking class.")
        activities.append("Go for a brisk walk (carry an umbrella just in case).")
    else:
        activities.append("Plan a flexible day with indoor and outdoor options.")
    
    # 基于温度微调
    if temperature_c < 5:
        activities.append("Consider indoor activities to stay warm.")
    elif temperature_c > 30:
        activities.append("Stay hydrated and seek air-conditioned places during midday.")
    
    return activities[:5]  # 返回最多5条建议

# 工具3:生成出行清单
@ai_function()
def generate_packing_list(activities: List[str], weather_condition: str, temperature_c: float) -> str:
    """Generate a concise packing list based on planned activities and weather.
    
    Args:
        activities: A list of activity strings recommended.
        weather_condition: Description of the weather.
        temperature_c: The maximum temperature in degrees Celsius.
    
    Returns:
        A formatted string representing the packing list.
    """
    essentials = ["Phone/Wallet/Keys", "Water bottle"]
    clothing = []
    gear = []
    
    # 根据天气和温度添加衣物
    if "rain" in weather_condition.lower():
        clothing.append("Waterproof jacket or umbrella")
        gear.append("Waterproof bag cover")
    if temperature_c < 15:
        clothing.append("Warm layers (sweater, jacket)")
    if temperature_c > 25:
        clothing.append("Lightweight, breathable clothing")
        gear.append("Sunscreen and sunglasses")
    if "hike" in " ".join(activities).lower() or "walk" in " ".join(activities).lower():
        gear.append("Comfortable walking shoes")
    
    # 根据活动添加特定装备
    for activity in activities:
        act_lower = activity.lower()
        if "swim" in act_lower or "beach" in act_lower:
            gear.append("Swimsuit and towel")
        if "cinema" in act_lower:
            gear.append("Movie tickets (pre-booked)")
    
    # 格式化输出
    list_str = "**Packing List:**\n"
    if clothing:
        list_str += f"- **Clothing:** {', '.join(clothing)}\n"
    if gear:
        list_str += f"- **Gear & Misc:** {', '.join(gear)}\n"
    list_str += f"- **Essentials:** {', '.join(essentials)}"
    
    return list_str

实操心得 :在定义工具时,务必确保函数参数类型提示(Type Hints)清晰准确(如 str , int , List[str] )。Kani和底层的模型会利用这些类型信息来更好地理解和生成参数。返回类型也尽量明确,这有助于模型理解工具执行的结果。

3.3 组装智能体并运行交互

工具定义好后,我们需要创建一个引擎,将这些工具“装配”上去,并提供一个系统提示词来塑造智能体的行为。

import asyncio
from kani import Kani, chat_in_terminal
from kani.engines.openai import OpenAIClient

# 1. 初始化AI客户端(这里使用OpenAI,需要设置API_KEY环境变量)
# 请在运行前设置环境变量:export OPENAI_API_KEY='your-key'
engine = OpenAIClient(model="gpt-4-turbo-preview")  # 或使用 "gpt-3.5-turbo"

# 2. 创建自定义的Kani智能体类
class WeatherTravelAssistant(Kani):
    # 定义默认系统提示词
    SYSTEM_PROMPT = """You are a helpful and concise weather and travel assistant. Your goal is to help users plan their day based on the weather.
    You have access to tools that can:
    1. Get the weather forecast for a city and date.
    2. Recommend activities based on weather conditions.
    3. Generate a packing list for the recommended activities.
    
    When a user asks about weather or plans:
    1. FIRST, always use the `get_weather_forecast` tool to get accurate weather data. Do not assume or invent weather.
    2. THEN, based on the actual weather data, use the `recommend_activities` tool to suggest suitable activities.
    3. FINALLY, use the `generate_packing_list` tool to create a practical packing list based on the weather and activities.
    
    Present the information in a clear, structured, and friendly manner. If the user's request is vague (e.g., just a city name), ask for clarification (e.g., which date?)."""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 将工具函数注册到这个智能体实例
        self.register_function(get_weather_forecast)
        self.register_function(recommend_activities)
        self.register_function(generate_packing_list)

# 3. 实例化智能体
async def main():
    assistant = WeatherTravelAssistant(engine=engine)
    
    # 方式一:在代码中进行单轮对话
    print("Testing single turn...")
    reply = await assistant.chat_round_str("What's the weather like in Tokyo this Saturday? And what can I do?")
    print(f"Assistant: {reply}")
    
    # 方式二:启动一个交互式终端聊天(更直观)
    print("\n--- Starting interactive terminal chat ---")
    await chat_in_terminal(assistant)

# 运行异步主函数
if __name__ == "__main__":
    asyncio.run(main())

运行这段代码,你会进入一个交互式终端。尝试输入:“帮我看看北京下周三的天气,然后推荐点活动”。智能体会自动执行以下步骤:

  1. 调用 get_weather_forecast("Beijing", "2024-05-29") (假设下周三是这个日期)。
  2. 拿到返回的天气字典后,调用 recommend_activities(condition, max_temp)
  3. 拿到活动列表后,调用 generate_packing_list(activities, condition, max_temp)
  4. 最后,模型会综合所有工具返回的结果,生成一段连贯、友好的回复呈现给你。

关键设计点 :注意系统提示词( SYSTEM_PROMPT )的编写。它明确规定了智能体的决策流程(先查天气,再推荐,最后列清单),并约束其行为(必须使用工具,不能编造天气)。好的提示词是引导智能体可靠工作的关键,它减少了模型“胡思乱想”的可能。

4. 高级特性与实战技巧

掌握了基础用法后,我们可以探索Kani的一些高级特性,这些特性能让你的智能体更强大、更稳定。

4.1 流式输出与实时反馈

默认情况下, chat_round_str 会等待整个推理循环(可能包含多次模型调用和工具执行)全部完成,才返回最终结果。对于长时间运行的任务,用户会陷入等待。Kani支持流式输出,你可以实时看到模型的“思考过程”。

async def stream_chat_example():
    assistant = WeatherTravelAssistant(engine=engine)
    # 使用 chat_round_stream 方法
    async for msg in assistant.chat_round_stream("What's the weather in London tomorrow?"):
        # msg是一个流式事件,可能是模型开始思考、生成文本、调用工具等
        if msg.content:  # 如果是文本内容
            print(msg.content, end="", flush=True)  # 逐块打印
        elif msg.event == "function_call":
            print(f"\n[Calling tool: {msg.name}]...")
        elif msg.event == "function_return":
            print(f"\n[Tool returned]")
    print()  # 最后换行

流式输出不仅提升了用户体验,对于调试也极其有用。你可以清晰地看到模型何时决定调用工具、调用了哪个工具、以及工具返回的结果,这有助于你优化提示词和工具设计。

4.2 上下文管理与摘要策略

长对话是智能体的核心挑战。Kani引擎内置了智能的上下文窗口管理。当对话历史(包括系统提示、所有消息、工具调用)的总令牌数超过模型上限(如GPT-4的128K)时,Kani不会简单地截断最早的消息,而是采用更高级的策略。

你可以在创建引擎时配置 retention_policy 。一个常用的策略是 LastMessageRetentionPolicy 结合 SummaryRetentionPolicy

  • LastMessageRetentionPolicy : 保留最近的N条消息。
  • SummaryRetentionPolicy : 当需要腾出空间时,将最早的一部分消息压缩成一个摘要。
from kani.engines.openai import OpenAIClient
from kani.retrievers import LastMessageRetentionPolicy, SummaryRetentionPolicy

# 创建一个组合策略:保留最近10条完整消息,更早的消息进行摘要。
retention_policy = LastMessageRetentionPolicy(keep_messages=10)
# 注意:原版Kani中,SummaryRetentionPolicy可能需要自定义或使用其他方式实现上下文摘要。
# 更常见的做法是依赖模型自身的上下文长度,或使用外挂向量数据库进行长期记忆管理。
engine = OpenAIClient(model="gpt-4", api_key="...")
# 策略通常通过创建自定义的Kani类或中间件实现,此处展示概念。

对于超长程记忆,Kani的设计允许集成向量数据库(如Chroma、Weaviate)。你可以将历史对话片段存入向量库,在需要时通过语义搜索召回相关记忆,再注入到当前上下文中。这属于更进阶的用法,需要你实现一个自定义的 Retriever

4.3 自定义模型与本地部署集成

Kani的引擎是抽象的,这意味着它不绑定于OpenAI。你可以轻松集成本地部署的大模型。例如,使用LM Studio提供的本地API服务器:

from kani.engines.httpclient import BaseClient

class LMStudioEngine(BaseClient):
    """一个适配LM Studio本地服务器的简易引擎"""
    def __init__(self, base_url="http://localhost:1234/v1"):
        super().__init__(base_url)
        # LM Studio的API端点与OpenAI基本兼容
        self.chat_url = f"{base_url}/chat/completions"
    
    async def chat_round(self, messages, functions=None, **kwargs):
        # 构建请求体,将Kani的消息格式转换为LM Studio兼容格式
        # 这里需要处理functions/tools的映射(如果模型支持函数调用)
        # ... 实现具体的HTTP请求和响应解析逻辑 ...
        pass

# 使用自定义引擎
local_engine = LMStudioEngine()
assistant = WeatherTravelAssistant(engine=local_engine)

这为隐私敏感、成本可控或需要特定模型调优的场景提供了可能。你需要根据本地模型API的具体规范来实现请求/响应的编解码。

4.4 多智能体协同与编排

单个智能体能力有限,复杂的任务可能需要多个智能体分工合作。Kani框架本身不强制规定多智能体架构,但其清晰的 Kani 实例化模式非常适合构建多智能体系统。

一种简单的模式是“管理者-工作者”(Manager-Worker)。创建一个 ManagerKani ,它的工具库里包含的是“调用其他智能体”。例如:

class ManagerKani(Kani):
    def __init__(self, engine, weather_agent, research_agent):
        super().__init__(engine)
        self.weather_agent = weather_agent
        self.research_agent = research_agent
        self.register_function(self.delegate_to_weather_agent)
        self.register_function(self.delegate_to_research_agent)
    
    @ai_function()
    async def delegate_to_weather_agent(self, query: str) -> str:
        """Delegate a weather-related query to the specialist agent."""
        response = await self.weather_agent.chat_round_str(query)
        return response
    
    @ai_function()
    async def delegate_to_research_agent(self, query: str) -> str:
        """Delegate a research-intensive query to the specialist agent."""
        response = await self.research_agent.chat_round_str(query)
        return response

当用户向管理者提出一个复合问题,如“北京下周天气如何,并查查那里有什么历史博物馆推荐?”管理者可以决定调用 delegate_to_weather_agent delegate_to_research_agent 两个工具,分别获取答案,然后综合回复。这实现了智能体间的简单编排。

5. 常见问题、调试技巧与性能优化

在实际开发中,你肯定会遇到各种问题。以下是我在多次使用Kani过程中积累的一些常见问题与解决思路。

5.1 模型不调用工具或错误调用

这是最常见的问题,根源通常在于提示词或工具定义。

  • 症状1:模型完全忽略工具,直接以文本形式回答。

    • 检查点1:系统提示词是否明确指令? 在系统提示中,必须清晰地告诉模型“你必须使用提供的工具来获取信息”。像我们之前例子中的“FIRST, always use the get_weather_forecast tool...”就是很强的指令。
    • 检查点2:工具描述是否清晰? 检查 @ai_function() 装饰器下的文档字符串。确保它对功能、参数(名称、类型、格式、可选/必选)、返回值的描述毫无歧义。模型完全依赖这段描述来理解工具。
    • 检查点3:模型能力是否支持? 确保你使用的模型版本支持函数调用(Function Calling)。例如, gpt-3.5-turbo 的某些旧版本可能不支持或支持不佳。
  • 症状2:模型调用了错误的工具,或参数格式错误。

    • 检查点1:参数类型提示(Type Hints)是否正确? Python的类型提示( str , int , List[str] , Optional[float] )会被Kani转换为JSON Schema。确保它们准确反映了参数期望的类型。对于复杂对象,使用 TypedDict BaseModel (Pydantic)可以提供更精确的模式。
    • 检查点2:是否提供了示例? 在工具的描述或系统提示中,为复杂参数提供示例值可以极大提高模型调用的准确性。例如: city: The city name (e.g., "San Francisco", "柏林")
    • 检查点3:上下文是否混乱? 如果对话历史很长且包含许多无关信息,模型可能会分心。尝试使用更严格的 retention_policy 或在新会话中测试。

5.2 处理工具执行失败与错误重试

工具执行可能因网络、API限制、无效输入等原因失败。健壮的智能体需要处理这些情况。

@ai_function()
def get_weather_forecast(city: str, date: str) -> dict:
    """Get the weather forecast..."""
    try:
        # ... API调用逻辑 ...
        return result
    except requests.exceptions.Timeout:
        # 返回一个结构化的错误信息,让模型能够理解并可能重试或调整策略
        return {
            "error": "REQUEST_TIMEOUT",
            "message": "The weather service is temporarily unreachable. Please try again in a moment or specify a different city.",
            "suggestion": "You can ask the user to retry or try a nearby major city."
        }
    except KeyError:
        return {
            "error": "CITY_NOT_FOUND",
            "message": f"Could not find coordinates for '{city}'. Please ensure the city name is spelled correctly.",
            "suggestion": "Ask the user to clarify or provide a nearby major city name."
        }

在系统提示词中,可以加入指导:“如果工具返回包含 error 字段,说明执行失败。请根据错误信息中的 suggestion 向用户友好地解释问题,并引导他们提供更正的信息或采取替代方案。”这样,模型在收到错误返回时,就能生成得体的回复,而不是崩溃或胡言乱语。

5.3 性能优化与成本控制

对于生产环境,性能和成本是需要考虑的重要因素。

  • 缓存工具结果 :对于频繁查询且结果变化不快的工具(如城市坐标、静态信息查询),可以添加缓存层。
    from functools import lru_cache
    
    @lru_cache(maxsize=100)
    @ai_function()
    def get_city_coordinates(city: str) -> dict:
        # ... 地理编码API调用 ...
        pass
    
  • 精简上下文 :避免在系统提示词或消息历史中放入过多无关的示例或说明。每个令牌都在消耗成本和上下文窗口。
  • 使用更便宜的模型进行简单轮次 :对于不需要复杂推理或工具调用的简单确认、格式化回复,可以尝试让智能体在特定条件下切换到更便宜的模型(如 gpt-3.5-turbo )。这需要更复杂的引擎调度逻辑。
  • 异步并发 :如果智能体需要调用多个彼此独立的工具(例如,同时查询A城市和B城市的天气),可以考虑使用 asyncio.gather 来并发执行,减少总体等待时间。但需要注意,这要求工具函数本身是异步的(定义为 async def ),并且引擎和模型调用支持异步。

5.4 调试与日志记录

当智能体行为不符合预期时,详细的日志是救命稻草。Kani提供了日志接口。

import logging
# 设置Kani的日志级别为DEBUG,可以看到引擎内部的消息流转、工具调用详情。
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("kani")

运行你的智能体,控制台会输出类似这样的信息:

DEBUG:kani.engine: Assembling messages for model...
DEBUG:kani.engine: Messages: [SystemMessage(...), UserMessage(...)]
DEBUG:kani.engine: Calling model with functions: [...]
DEBUG:kani.engine: Model response: ChatCompletionMessage(content=None, function_call={'name': 'get_weather_forecast', 'arguments': '...'}, ...)
DEBUG:kani.engine: Executing function: get_weather_forecast with args: {...}
DEBUG:kani.engine: Function returned: {...}
DEBUG:kani.engine: Appending function result to context...

通过这些日志,你可以精确地看到模型收到了什么上下文、决定调用什么工具、传递了什么参数、以及工具返回了什么。这对于排查提示词问题、工具定义问题或模型理解偏差至关重要。

最后,我想分享的一点体会是,构建一个可靠的AI智能体,更像是在设计一个精密的“人机协作流程”。Kani这样的框架提供了优秀的自动化基础设施,但成功的关键依然在于开发者对任务本身的深刻理解、对工具边界的清晰定义、以及对模型“思维”方式的巧妙引导(通过提示词)。从一个小而专的智能体开始,逐步迭代其能力和可靠性,是避免陷入复杂性和不可控性的最佳路径。我们的天气出行助手就是一个很好的起点,你可以在此基础上,继续为它添加“预订餐厅”、“查询交通”等工具,让它真正成为一个实用的个人生活助理。

Logo

码道开发者社区,聚焦华为云码道 CodeArts 代码智能体,沉淀 Agent、Skill、鸿蒙开发实战内容,供开发者查阅资料、交流技术、分享工程实践

更多推荐