大模型开发实战:如何用Function_call和Agent构建智能对话系统(附代码示例)

在当今AI技术快速发展的背景下,大模型已经不再局限于简单的文本生成任务。开发者们正积极探索如何将这些强大的语言模型转化为能够执行复杂任务的智能系统。本文将深入探讨如何通过Function_call和Agent技术的有机结合,构建一个真正智能的对话系统。

对于有一定大模型开发经验的工程师来说,最大的挑战往往不在于模型本身的使用,而在于如何让模型与现实世界的工具和服务进行有效交互。这正是Function_call和Agent技术大显身手的地方——它们为模型提供了"手脚",使其不再只是"纸上谈兵"。

1. Function_call技术深度解析与实现

Function_call本质上是大模型与外部世界交互的桥梁。它允许语言模型在对话过程中识别需要调用外部功能的场景,并生成相应的调用请求。这种机制极大地扩展了大模型的能力边界。

1.1 Function_call的核心实现原理

现代大模型通过特殊的训练方式学会了识别何时需要调用函数以及如何格式化函数调用请求。以下是一个典型的Function_call工作流程:

  1. 模型分析用户输入,判断是否需要调用外部功能
  2. 如果确定需要调用,模型会生成结构化的函数调用请求
  3. 系统执行实际函数调用并获取结果
  4. 模型将结果整合到自然语言回复中
# 示例:定义可调用函数
functions = [
    {
        "name": "get_current_weather",
        "description": "获取指定位置的当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "城市和地区,例如:'北京海淀区'",
                },
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["location"],
        },
    }
]

1.2 实战:为对话系统添加Function_call能力

让我们通过一个天气查询的案例,看看如何实际实现Function_call功能。首先需要定义函数规范,然后处理模型的函数调用请求。

import openai
import json

def get_current_weather(location, unit="celsius"):
    """模拟天气查询函数"""
    # 实际应用中这里会调用天气API
    return f"{location}当前天气为22度{unit},晴天"

def run_conversation():
    messages = [{"role": "user", "content": "北京现在的天气怎么样?"}]
    
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages,
        functions=functions,
        function_call="auto",
    )
    
    response_message = response["choices"][0]["message"]
    
    if response_message.get("function_call"):
        function_name = response_message["function_call"]["name"]
        function_args = json.loads(response_message["function_call"]["arguments"])
        
        if function_name == "get_current_weather":
            function_response = get_current_weather(
                location=function_args.get("location"),
                unit=function_args.get("unit"),
            )
            
            messages.append(response_message)
            messages.append(
                {
                    "role": "function",
                    "name": function_name,
                    "content": function_response,
                }
            )
            
            second_response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
            )
            return second_response["choices"][0]["message"]["content"]
    
    return response_message["content"]

print(run_conversation())

提示:在实际项目中,应该将函数调用逻辑封装为可重用的组件,并添加适当的错误处理和日志记录。

2. Agent系统的设计与实现

Agent技术将大模型从一个被动的文本生成器转变为能够主动规划和执行任务的智能体。一个完整的Agent系统通常包含以下几个关键组件:

组件功能描述实现要点
记忆模块存储对话历史和上下文信息需要考虑记忆压缩和关键信息提取
规划模块分解复杂任务为可执行步骤可以使用思维链(CoT)等技术
工具集可调用的外部功能集合需要良好的文档和错误处理
决策引擎决定下一步行动的核心逻辑平衡自主性和可控性

2.1 构建基础Agent框架

下面是一个简化版Agent框架的实现代码,展示了核心决策循环:

class BasicAgent:
    def __init__(self, model="gpt-3.5-turbo"):
        self.model = model
        self.memory = []
        self.tools = {
            "search": self.search_web,
            "calculate": self.calculate,
            # 可以添加更多工具
        }
    
    def search_web(self, query):
        # 模拟网络搜索
        return f"关于'{query}'的搜索结果..."
    
    def calculate(self, expression):
        # 简单的计算功能
        try:
            return str(eval(expression))
        except:
            return "计算失败"
    
    def run(self, user_input):
        self.memory.append({"role": "user", "content": user_input})
        
        # 第一步:决定是否需要使用工具
        decision_prompt = f"""
        当前对话历史:
        {self.memory}
        
        用户最新输入:{user_input}
        
        是否需要使用工具?如果需要,请指定工具名称和参数。
        可用工具:{list(self.tools.keys())}
        响应格式:TOOL_CALL:工具名 参数 或 CONTINUE:回复内容
        """
        
        decision = openai.ChatCompletion.create(
            model=self.model,
            messages=[{"role": "user", "content": decision_prompt}],
            temperature=0,
        ).choices[0].message.content
        
        if decision.startswith("TOOL_CALL:"):
            _, tool_name, *args = decision.split()
            tool_input = " ".join(args)
            
            if tool_name in self.tools:
                tool_result = self.tools[tool_name](tool_input)
                self.memory.append({
                    "role": "assistant",
                    "content": f"使用了{tool_name}工具,结果:{tool_result}"
                })
                return self.run(f"工具执行结果:{tool_result}")
        
        # 不需要工具或工具调用失败时直接回复
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=self.memory,
        ).choices[0].message.content
        
        self.memory.append({"role": "assistant", "content": response})
        return response

# 使用示例
agent = BasicAgent()
print(agent.run("计算一下3的平方加上4的平方等于多少?"))

3. Function_call与Agent的协同工作机制

将Function_call集成到Agent系统中可以创造出更强大的智能体。这种结合的关键在于设计合理的决策流程和控制机制。

3.1 协同工作流程设计

  1. 输入解析阶段:Agent分析用户输入,确定意图和可能的行动路径
  2. 工具选择阶段:根据需求选择合适的Function_call或直接生成回复
  3. 执行监控阶段:跟踪Function_call执行状态,处理可能的错误
  4. 结果整合阶段:将Function_call结果融入自然语言回复
class EnhancedAgent(BasicAgent):
    def __init__(self, model="gpt-3.5-turbo"):
        super().__init__(model)
        self.functions = functions  # 使用之前定义的functions
        
    def run_with_functions(self, user_input):
        self.memory.append({"role": "user", "content": user_input})
        
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=self.memory,
            functions=self.functions,
            function_call="auto",
        ).choices[0].message
        
        if hasattr(response, "function_call"):
            # 处理函数调用
            func_name = response.function_call.name
            func_args = json.loads(response.function_call.arguments)
            
            if func_name == "get_current_weather":
                weather = self.tools["weather"](func_args["location"])
                self.memory.append({
                    "role": "function",
                    "name": func_name,
                    "content": weather,
                })
                return self.run_with_functions(f"天气数据已更新:{weather}")
        
        # 无函数调用时直接回复
        self.memory.append({"role": "assistant", "content": response.content})
        return response.content

3.2 性能优化技巧

  • 函数调用缓存:对频繁调用的函数结果进行缓存
  • 并行执行:当多个Function_call互不依赖时可以并行执行
  • 超时控制:为每个Function_call设置合理的超时时间
  • 结果验证:对Function_call返回的结果进行基本验证

4. 实战案例:构建智能旅行助手

让我们综合运用所学知识,构建一个能够处理复杂旅行规划的智能Agent。这个助手需要能够:

  • 查询航班信息
  • 查询酒店信息
  • 进行货币换算
  • 提供旅行建议

4.1 系统架构设计

旅行助手Agent
├── 核心决策引擎
├── 记忆系统
│   ├── 短期记忆(当前会话)
│   └── 长期记忆(用户偏好)
├── 工具集
│   ├── 航班查询Function
│   ├── 酒店查询Function
│   ├── 货币换算Function
│   └── 景点推荐Function
└── 输出生成器

4.2 关键代码实现

class TravelAssistant(EnhancedAgent):
    def __init__(self):
        super().__init__()
        self.functions = [
            {
                "name": "search_flights",
                "description": "查询航班信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "departure": {"type": "string", "description": "出发城市"},
                        "destination": {"type": "string", "description": "目的地城市"},
                        "date": {"type": "string", "description": "出发日期,格式YYYY-MM-DD"},
                    },
                    "required": ["departure", "destination", "date"],
                },
            },
            # 其他函数定义类似
        ]
        
        self.tools.update({
            "search_flights": self.mock_search_flights,
            # 其他工具注册
        })
    
    def mock_search_flights(self, departure, destination, date):
        # 模拟航班查询
        return f"找到从{departure}到{destination}的航班,日期{date},价格约2000元"
    
    def run(self, user_input):
        # 重写run方法,添加旅行领域特定逻辑
        if "预算" in user_input:
            # 添加预算分析逻辑
            pass
        return super().run(user_input)

# 使用示例
assistant = TravelAssistant()
print(assistant.run("我想下个月从北京飞往上海,有什么航班选择?"))

在实际项目中,这种架构可以扩展为支持插件系统,允许动态加载新的Function_call模块,使系统能力可以不断增长。

更多推荐