本文的目的

本文将为一系列旨在开发聊天机器人的文章奠定基础，聊天机器人可以作为小型企业的单点交互来支持和执行业务流程，或者聊天机器人可以在你的个人生活中组织你跟踪所需的一切的。从数据、例程、文件到图片，我们只想与我们的助手聊天，让它找出在哪里存储和检索你的数据。

从人工智能未来的宏大愿景过渡到实际应用，让我们来放大创建一个原型代理。我们将着手开发一个 "费用跟踪 "（Expense Tracking）代理，这是一项简单而又重要的任务，展示人工智能如何协助高效管理金融交易。

这个 "费用跟踪 "（Expense Tracking）原型不仅将展示人工智能在日常任务自动化方面的潜力，还将阐明设计一个能与数据库无缝交互的人工智能系统所面临的挑战和需要考虑的因素。通过关注这个例子，我们可以探索代理设计、输入验证和人工智能与现有系统集成的复杂性，为将来更复杂的应用奠定坚实的基础。

实际操作：测试 OpenAI 工具调用

为了让我们的代理原型栩栩如生，并找出潜在的瓶颈，我们正在冒险测试 OpenAI 的工具调用功能。从费用跟踪的基本示例开始，我们正在模拟真实世界的应用，奠定基础。这一阶段包括创建基础模型，并使用 langchain 库的 convert_to_openai_tool 函数将其转换为 OpenAI 工具模式。此外，制作报告工具（report_tool）可让我们未来的代理传达结果，或突出缺失的信息或问题：

from pydantic.v1 import BaseModel, validator  
from datetime import datetime
from langchain_core.utils.function_calling import convert_to_openai_tool
  
  
class Expense(BaseModel):    
   description: str    
   net_amount: float    
   gross_amount: float    
   tax_rate: float    
   date: datetime

class Report(BaseModel):
   report: str
add_expense_tool = convert_to_openai_tool(Expense)
report_tool = convert_to_openai_tool(Report)

有了数据模型和工具，下一步就是使用 OpenAI 客户端 SDK 启动简单的工具调用。在这个初始测试中，我们故意向模型提供不充分的信息，看看它是否能正确指出缺失的信息。这种方法不仅能测试代理的功能能力，还能测试其交互和错误处理能力。

调用 OpenAI API

现在，我们将使用 OpenAI 客户端 SDK 启动一个简单的工具调用。在第一次测试中，我们故意向模型提供不充分的信息，看它能否通知我们缺少的细节。

from openai import OpenAI  
from langchain_core.utils.function_calling import convert_to_openai_tool  
  
SYSTEM_MESSAGE = """You are tasked with completing specific objectives and 
must report the outcomes. At your disposal, you have a variety of tools, 
each specialized in performing a distinct type of task.  
  
For successful task completion:  
Thought: Consider the task at hand and determine which tool is best suited 
based on its capabilities and the nature of the work.  
  
Use the report_tool with an instruction detailing the results of your work.  
If you encounter an issue and cannot complete the task:  
  
Use the report_tool to communicate the challenge or reason for the 
task's incompletion.  
You will receive feedback based on the outcomes of 
each tool's task execution or explanations for any tasks that 
couldn't be completed. This feedback loop is crucial for addressing 
and resolving any issues by strategically deploying the available tools.  
"""  
user_message = "I have spend 5$ on a coffee today please track my expense. The tax rate is 0.2."
  
client = OpenAI()  
model_name = "gpt-3.5-turbo-0125"  
  
messages = [  
    {"role":"system", "content": SYSTEM_MESSAGE},  
    {"role":"user", "content": user_message}  
]  
  
response = client.chat.completions.create(  
            model=model_name,  
            messages=messages,  
            tools=[  
                convert_to_openai_tool(Expense),  
                convert_to_openai_tool(ReportTool)]  
        )

接下来，我们需要一个新函数来从响应中读取函数调用的参数：

def parse_function_args(response):
    message = response.choices[0].message
    return json.loads(message.tool_calls[0].function.arguments)
print(parse_function_args(response))

{'description': 'Coffee',
 'net_amount': 5,
 'gross_amount': None,
 'tax_rate': 0.2,
 'date': '2023-10-06T12:00:00Z'}

我们可以看到，在执行过程中遇到了几个问题：

• gross_amount未计算。
• 日期是幻觉。

有鉴于此。让我们尝试解决这些问题，优化我们的代理工作流程。

优化工具处理

要优化代理工作流程，我认为工作流程优先于提示工程至关重要。对提示进行微调，让代理学会完美地使用所提供的工具并不犯错，这可能很诱人，但更可取的做法是首先调整工具和流程。当出现典型错误时，首先要考虑的应该是如何通过代码来解决。

处理缺失信息

有效处理缺失信息是创建稳健可靠的代理必不可少的主题。在前面的示例中，为代理提供类似 "get_current_date "的工具是特定情况下的一种变通方法。但是，我们必须假设在各种情况下都会出现信息缺失的情况，而且我们不能仅仅依靠提示工程和添加更多工具来防止模型产生信息缺失的幻觉。

解决这种情况的一个简单办法是修改工具模式，将所有参数都视为可选参数。这种方法可以确保代理只提交它知道的参数，从而避免不必要的幻觉。

因此，让我们来看看 openai 工具模式：

add_expense_tool = convert_to_openai_tool(Expense)
print(add_expense_tool)

{'type': 'function',
 'function': {'name': 'Expense',
  'description': '',
  'parameters': {'type': 'object',
   'properties': {'description': {'type': 'string'},
    'net_amount': {'type': 'number'},
    'gross_amount': {'type': 'number'},
    'tax_rate': {'type': 'number'},
    'date': {'type': 'string', 'format': 'date-time'}},
   'required': ['description',
    'net_amount',
    'gross_amount',
    'tax_rate',
    'date']}}}

正如我们所看到的，我们需要删除特殊键 required。下面是如何调整 add_expense_tool 模式，通过删除必填键使参数成为可选参数：

del add_expense_tool["function"]["parameters"]["required"]

设计工具类

接下来，我们可以设计一个工具类，用于初步检查输入参数是否有缺失值。我们创建的工具类包含两个方法：.run()、.validate_input()，以及一个 openai_tool_schema 属性，在该属性中，我们通过删除所需的参数来操作工具模式。此外，我们还定义了带有 content 和 success 字段的 ToolResult BaseModel，作为每次工具运行的输出对象。

from pydantic import BaseModel
from typing import Type, Callable, Dict, Any, List
class ToolResult(BaseModel):  
    content: str  
    success: bool  
  
class Tool(BaseModel):  
    name: str  
    model: Type[BaseModel]  
    function: Callable  
    validate_missing: bool = False  
  
    class Config:  
        arbitrary_types_allowed = True  
  
    def run(self, **kwargs) -> ToolResult:
        if self.validate_missing:
            missing_values = self.validate_input(**kwargs)  
            if missing_values:  
                content = f"Missing values: {', '.join(missing_values)}"  
                return ToolResult(content=content, success=False)  
        result = self.function(**kwargs)  
        return ToolResult(content=str(result), success=True)  
      
    def validate_input(self, **kwargs) -> List[str]:  
        missing_values = []  
        for key in self.model.__fields__.keys():  
            if key not in kwargs:  
                missing_values.append(key)  
        return missing_values
    @property
    def openai_tool_schema(self) -> Dict[str, Any]:
        schema = convert_to_openai_tool(self.model)
        if "required" in schema["function"]["parameters"]:
            del schema["function"]["parameters"]["required"]
        return schema

tool是人工智能代理工作流程的重要组成部分，是创建和管理各种工具的蓝图，代理可以利用这些工具来执行特定任务。它旨在处理输入验证、执行工具功能并以标准格式返回结果。

tool的关键组件：

1. name：工具名称。
2. model：模型： 定义工具输入模式的 Pydantic BaseModel。
3. function：函数： 工具执行的可调用函数。
4. validate_missing： 布尔标志，表示是否验证缺失的输入值（默认为 False）。

tool有两个主要方法：

1. run(self, **kwargs) -> ToolResult： 该方法负责使用提供的输入参数执行工具函数。它首先检查 validate_missing 是否设置为 True。如果是，它就会调用 validate_input() 方法来检查输入值是否丢失。如果发现任何缺失值，它将返回一个带有错误信息的 ToolResult 对象，并将成功设置为 False。如果所有必需的输入值都存在，它将继续使用所提供的参数执行工具函数，并返回一个包含结果的 ToolResult 对象，同时将成功设置为 True。
2. validate_input(self, **kwargs) -> List[str]： 此方法将传递给工具的输入参数与模型中定义的预期输入模式进行比较。它会遍历模型中定义的字段，并检查输入参数中是否存在每个字段。如果缺少任何字段，它就会将字段名追加到缺失值列表中。最后，它会返回缺失值列表。

tool还有一个名为 openai_tool_schema 的属性，用于返回工具的 OpenAI 工具模式。它使用 convert_to_openai_tool() 函数将模型转换为 OpenAI 工具模式格式。此外，它还删除了模式中的 "required "键，使所有输入参数都成为可选参数。这样，代理就可以只提供可用的信息，而无需对缺失值产生幻觉。

通过封装工具的功能、输入验证和模式生成，工具类为在人工智能代理工作流中创建和管理工具提供了一个简洁、可重复使用的接口。它抽象化了处理缺失值的复杂性，确保代理可以优雅地处理不完整信息，同时根据可用输入执行适当的工具。

测试缺失信息处理

接下来，我们将扩展 OpenAI API 调用。我们希望客户端利用我们的工具和响应对象直接触发 tool.run()。为此，我们需要在新创建的工具类中初始化我们的工具。我们定义了两个返回成功消息字符串的虚函数。

def add_expense_func(**kwargs):  
    return f"Added expense: {kwargs} to the database."
add_expense_tool = Tool(  
    name="add_expense_tool",  
    model=Expense,  
    function=add_expense_func  
)  
  
def report_func(report: str = None):  
    return f"Reported: {report}"  
  
report_tool = Tool(  
    name="report_tool",  
    model=ReportTool,  
    function=report_func  
)  
  
tools = [add_expense_tool, report_tool]

接下来，我们定义辅助函数，每个函数都将客户端的响应作为输入，帮助与外部工具进行交互。

def get_tool_from_response(response, tools=tools):  
    tool_name = response.choices[0].message.tool_calls[0].function.name  
    for t in tools:  
        if t.name == tool_name:  
            return t  
    raise ValueError(f"Tool {tool_name} not found in tools list.")
def parse_function_args(response):  
    message = response.choices[0].message  
    return json.loads(message.tool_calls[0].function.arguments)
def run_tool_from_response(response, tools=tools):  
    tool = get_tool_from_response(response, tools)  
    tool_kwargs = parse_function_args(response)  
    return tool.run(**tool_kwargs)

现在，我们可以使用新工具执行客户端，并使用 run_tool_from_response 函数。

response = client.chat.completions.create(  
            model=model_name,  
            messages=messages,  
            tools=[tool.openai_tool_schema for tool in tools]  
        )
tool_result = run_tool_from_response(response, tools=tools)
print(tool_result)

content='Missing values: gross_amount, date' success=False

现在，我们看到我们的工具完美地显示了缺失值。由于我们采用了将所有参数作为可选参数发送的技巧，现在我们可以避免出现参数幻觉了。

构建代理工作流程

目前，我们的流程还不能代表一个真正的代理。到目前为止，我们只执行了一次 API 工具调用。要将其转化为代理工作流，我们需要引入一个迭代流程，将工具执行的结果反馈给客户端。基本流程如下：

让我们开始创建一个新的 OpenAIAgent 类：

class StepResult(BaseModel):  
    event: str   
    content: str  
    success: bool

class OpenAIAgent:  
      
    def __init__(  
            self,   
            tools: list[Tool],   
            client: OpenAI,   
            system_message: str = SYSTEM_MESSAGE,   
            model_name: str = "gpt-3.5-turbo-0125",  
            max_steps: int = 5,  
            verbose: bool = True  
    ):  
        self.tools = tools  
        self.client = client  
        self.model_name = model_name  
        self.system_message = system_message  
        self.step_history = []  
        self.max_steps = max_steps  
        self.verbose = verbose  
      
      
    def to_console(self, tag: str, message: str, color: str = "green"):  
        if self.verbose:  
            color_prefix = Fore.__dict__[color.upper()]  
            print(color_prefix + f"{tag}: {message}{Style.RESET_ALL}")

与 ToolResult 对象一样，我们为每个代理步骤定义了一个 StepResult 对象。然后，我们定义了 OpenAIAgent 类的 __init__ 方法和 to_console()方法，将中间步骤和工具调用打印到控制台，并使用 colorama 进行彩色打印。接下来，我们定义了代理的核心、run() 和 run_step() 方法。

class OpenAIAgent:
    # ... __init__...
    
    # ... to_console ...
    
    def run(self, user_input: str):  
        
        openai_tools = [tool.openai_tool_schema for tool in self.tools]    
        self.step_history = [    
            {"role":"system", "content":self.system_message},    
            {"role":"user", "content":user_input}    
        ]    
          
        step_result = None    
        i = 0
        
        self.to_console("START", f"Starting Agent with Input: {user_input}")
          
        while i < self.max_steps:  
            step_result = self.run_step(self.step_history, openai_tools)    
            
            if step_result.event == "finish":    
                break  
            elif step_result.event == "error":  
                self.to_console(step_result.event, step_result.content, "red")  
            else:  
                self.to_console(step_result.event, step_result.content, "yellow")  
            i += 1   
              
        self.to_console("Final Result", step_result.content, "green")  
        return step_result.content

在 run() 方法中，我们首先使用预定义的 system_message 和 user_input 初始化 step_history，它将作为我们的消息存储器。然后开始 while 循环，每次迭代都会调用 run_step，并返回一个 StepResult 对象。我们将确定代理是完成了任务还是发生了错误，错误信息也将传递到控制台。

class OpenAIAgent:
    # ... __init__...
    
    # ... to_console ...
    # ... run ...
    def run_step(self, messages: list[dict], tools):  
          
        # plan the next step  
        response = self.client.chat.completions.create(  
            model=self.model_name,  
            messages=messages,  
            tools=tools  
        )  
          
        # add message to history  
        self.step_history.append(response.choices[0].message)  
        
        # check if tool call is present  
        if not response.choices[0].message.tool_calls:  
            return StepResult(
                event="Error",
                content="No tool calls were returned.", 
                success=False
                )  
          
        tool_name = response.choices[0].message.tool_calls[0].function.name  
        tool_kwargs = parse_function_args(response)  
          
        # execute the tool call  
        self.to_console(
        "Tool Call", f"Name: {tool_name}\nArgs: {tool_kwargs}", "magenta"
        )  
        tool_result = run_tool_from_response(response, tools=self.tools)  
        tool_result_msg = self.tool_call_message(response, tool_result)  
        self.step_history.append(tool_result_msg)  
          
        if tool_result.success:  
            step_result = StepResult(  
                event="tool_result",   
                content=tool_result.content,   
                success=True)  
        else:  
            step_result = StepResult(  
                event="error",   
                content=tool_result.content,   
                success=False  
            )   
          
        return step_result  
          
      
    def tool_call_message(self, response, tool_result: ToolResult):  
        tool_call = response.choices[0].message.tool_calls[0]  
        return {  
            "tool_call_id": tool_call.id,  
            "role": "tool",  
            "name": tool_call.function.name,  
            "content": tool_result.content,  
        }

现在我们已经定义了每个步骤的逻辑。我们首先通过之前使用工具测试过的客户端 API 调用来获取响应对象。我们将响应消息对象附加到我们的step_history.然后，我们验证响应对象中是否包含工具调用，否则，我们会在 StepResult 中返回错误。然后，我们将工具调用记录到控制台，并使用之前定义的方法运行选定的工具run_tool_from_response()。我们还需要将工具结果附加到我们的消息历史记录中。 OpenAI 为此定义了一种特定的格式，以便模型通过将 tool_call_id 传递到我们的消息字典中来知道哪个工具调用引用哪个输出。这是通过我们的方法完成的tool_call_message()，该方法将响应对象和 tool_result 作为输入参数。在每个步骤结束时，我们将工具结果分配给 StepResult 对象，该对象还指示该步骤是否成功，并将其返回到 中的循环run()。

运行代理

现在，我们可以用之前的示例来测试我们的代理，并直接为其配备 get_current_date_tool 工具。在这里，我们可以将之前定义的 validate_missing 属性设置为 False，因为该工具不需要任何输入参数。

class DateTool(BaseModel):  
    x: str = None  
get_date_tool = Tool(  
    name="get_current_date",  
    model=DateTool,  
    function=lambda: datetime.now().strftime("%Y-%m-%d"),  
    validate_missing=False  
)  
      
tools = [  
    add_expense_tool,   
    report_tool,  
    get_date_tool  
]  
  
agent = OpenAIAgent(tools, client)
agent.run("I have spent 5$ on a coffee today please track my expense. The tax rate is 0.2.")

START: Starting Agent with Input: 
"I have spend 5$ on a coffee today please track my expense. The tax rate is 0.2."

Tool Call: get_current_date
Args: {}
tool_result: 2024-03-15

Tool Call: add_expense_tool
Args: {'description': 'Coffee expense', 'net_amount': 5, 'tax_rate': 0.2, 'date': '2024-03-15'}
error: Missing values: gross_amount

Tool Call: add_expense_tool
Args: {'description': 'Coffee expense', 'net_amount': 5, 'tax_rate': 0.2, 'date': '2024-03-15', 'gross_amount': 6}
tool_result: Added expense: {'description': 'Coffee expense', 'net_amount': 5, 'tax_rate': 0.2, 'date': '2024-03-15', 'gross_amount': 6} to the database.
Error: No tool calls were returned.

Tool Call: Name: report_tool
Args: {'report': 'Expense successfully tracked for coffee purchase.'}
tool_result: Reported: Expense successfully tracked for coffee purchase.

Final Result: Reported: Expense successfully tracked for coffee purchase.

在我们的原型代理成功执行后，强调一下代理如何根据计划有效地使用了指定的工具。首先，它调用了get_current_date_tool，为费用记录建立了一个基本的时间戳。随后，在尝试通过add_expense_tool记录费用时，我们智能设计的工具类识别出了一个关键的缺失信息——gross_amount，这对于准确的财务跟踪至关重要。令人印象深刻的是，代理自主地通过使用提供的税率计算出了gross_amount。

值得一提的是，在我们的测试中，输入费用的性质——无论是花在咖啡上的5美元是净额还是毛额——并没有明确说明。在这个阶段，代理执行任务并不需要这样的明确性。然而，这为我们改进代理的理解和交互能力提供了有价值的见解：在初始系统提示中加入这些详细信息可以显著提高代理在处理费用记录时的准确性和效率。这样的调整将确保从一开始就更全面地掌握财务数据。

总结

1. 迭代式开发：该项目强调了迭代式开发周期的关键性质，通过反馈促进持续改进。在人工智能领域，变化是常态，因此需要灵活和响应迅速的开发策略。
2. 处理不确定性：我们的经验凸显了优雅地处理模糊和错误的重要性。创新，如可选参数和严格的输入验证，对于提高代理的可靠性和用户体验至关重要。
3. 为特定任务定制代理工作流程：从这项工作中得出的一个关键见解是根据特定用例定制代理工作流程的重要性。除了组合一套工具外，工具交互和响应的战略设计至关重要。这种定制确保代理有效地应对特定挑战，从而实现更专注和高效的问题解决方法。