LLM Agent:从聊天机器人到智能助手的进化

LLM Agent是具有自主决策和执行能力的AI系统,它能够理解目标、制定计划、调用工具、 并根据反馈调整策略。本文将详细介绍如何构建强大的Agent系统。

Agent核心架构

智能Agent的关键组件

🧠 大脑(LLM)

  • • 理解用户意图
  • • 制定执行计划
  • • 推理和决策
  • • 生成响应

🔧 工具(Tools)

  • • API调用
  • • 数据库查询
  • • 代码执行
  • • 文件操作

💾 记忆(Memory)

  • • 短期记忆(对话历史)
  • • 长期记忆(知识库)
  • • 工作记忆(任务状态)
  • • 情景记忆(经验)

📋 规划(Planning)

  • • 任务分解
  • • 步骤排序
  • • 资源调度
  • • 策略优化

ReAct模式:思考与行动

Reasoning + Acting = ReAct

class ReActAgent:
    """ReAct模式的Agent实现"""
    
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.max_steps = 10
        
    def run(self, task):
        """执行任务"""
        prompt = f"""你是一个智能助手,需要完成以下任务:{task}

可用工具:
{self.get_tools_description()}

请按照以下格式回复:
Thought: 我需要思考下一步做什么
Action: 工具名称
Action Input: 工具输入参数
Observation: 工具返回结果
... (重复Thought/Action/Action Input/Observation)
Thought: 我已经知道最终答案
Final Answer: 最终答案

开始:"""
        
        messages = [{"role": "user", "content": prompt}]
        steps = 0
        
        while steps < self.max_steps:
            # LLM思考
            response = self.llm.generate(messages)
            
            # 解析响应
            if "Final Answer:" in response:
                return self.extract_final_answer(response)
            
            # 提取动作
            action, action_input = self.parse_action(response)
            
            if action and action in self.tools:
                # 执行工具
                observation = self.tools[action](action_input)
                
                # 添加观察结果
                messages.append({
                    "role": "assistant",
                    "content": response
                })
                messages.append({
                    "role": "user", 
                    "content": f"Observation: {observation}"
                })
            else:
                # 处理错误
                messages.append({
                    "role": "user",
                    "content": "无效的动作,请重新思考"
                })
            
            steps += 1
        
        return "达到最大步骤限制"
    
    def parse_action(self, text):
        """解析动作和输入"""
        import re
        
        action_match = re.search(r'Action:\s*(.+)', text)
        input_match = re.search(r'Action Input:\s*(.+)', text)
        
        if action_match and input_match:
            return action_match.group(1), input_match.group(1)
        
        return None, None

# 使用示例
agent = ReActAgent(
    llm=llm_client,
    tools={
        "search": web_search_tool,
        "calculator": calculator_tool,
        "database": database_query_tool
    }
)

result = agent.run("北京今天的天气怎么样,温度是多少摄氏度?")
print(result)

Function Calling实现

让LLM调用外部函数

import json
from typing import List, Dict, Any

class FunctionCallingAgent:
    """基于Function Calling的Agent"""
    
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.functions = []
        self.function_map = {}
    
    def register_function(self, func, description: str, parameters: Dict):
        """注册可调用函数"""
        function_spec = {
            "name": func.__name__,
            "description": description,
            "parameters": parameters
        }
        
        self.functions.append(function_spec)
        self.function_map[func.__name__] = func
    
    def run(self, user_input: str):
        """执行用户请求"""
        messages = [
            {"role": "system", "content": "你是一个智能助手"},
            {"role": "user", "content": user_input}
        ]
        
        # 第一次调用:决定是否需要使用函数
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            functions=self.functions,
            function_call="auto"
        )
        
        message = response.choices[0].message
        
        # 检查是否需要调用函数
        if message.function_call:
            # 解析函数调用
            function_name = message.function_call.name
            function_args = json.loads(message.function_call.arguments)
            
            # 执行函数
            if function_name in self.function_map:
                function_result = self.function_map[function_name](**function_args)
                
                # 将函数结果添加到对话
                messages.append(message)
                messages.append({
                    "role": "function",
                    "name": function_name,
                    "content": str(function_result)
                })
                
                # 第二次调用:基于函数结果生成最终回复
                final_response = self.client.chat.completions.create(
                    model="gpt-4",
                    messages=messages
                )
                
                return final_response.choices[0].message.content
        
        return message.content

# 定义工具函数
def get_weather(location: str, unit: str = "celsius"):
    """获取天气信息"""
    # 实际应调用天气API
    return f"{location}的天气是晴天,温度是25{unit}"

def calculate(expression: str):
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果:{result}"
    except:
        return "计算错误"

def search_database(query: str, table: str):
    """查询数据库"""
    # 实际应执行数据库查询
    return f"从{table}表查询'{query}'的结果:..."

# 创建Agent并注册函数
agent = FunctionCallingAgent(api_key="your-key")

agent.register_function(
    get_weather,
    "获取指定地点的天气信息",
    {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "地点名称"
            },
            "unit": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "description": "温度单位"
            }
        },
        "required": ["location"]
    }
)

agent.register_function(
    calculate,
    "计算数学表达式",
    {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "数学表达式"
            }
        },
        "required": ["expression"]
    }
)

# 使用Agent
result = agent.run("北京的天气怎么样?")
print(result)  # 自动调用get_weather函数

result = agent.run("计算 123 * 456 + 789")
print(result)  # 自动调用calculate函数

多Agent协作系统

构建Agent团队完成复杂任务

class MultiAgentSystem:
    """多Agent协作系统"""
    
    def __init__(self):
        self.agents = {}
        self.message_queue = []
    
    def register_agent(self, name: str, agent):
        """注册Agent"""
        self.agents[name] = agent
    
    def run_task(self, task: str):
        """执行协作任务"""
        # 主管Agent分解任务
        manager = self.agents["manager"]
        plan = manager.create_plan(task)
        
        results = {}
        for step in plan["steps"]:
            agent_name = step["agent"]
            subtask = step["task"]
            
            # 分配任务给特定Agent
            if agent_name in self.agents:
                agent = self.agents[agent_name]
                result = agent.execute(subtask, context=results)
                results[step["id"]] = result
                
                # 广播结果给其他Agent
                self.broadcast_message({
                    "from": agent_name,
                    "type": "result",
                    "content": result
                })
        
        # 汇总结果
        return manager.summarize(results)
    
    def broadcast_message(self, message):
        """广播消息"""
        for agent_name, agent in self.agents.items():
            if hasattr(agent, "receive_message"):
                agent.receive_message(message)

# 专门化的Agent类
class ResearchAgent:
    """研究Agent - 负责信息收集"""
    def execute(self, task, context=None):
        # 搜索和收集信息
        return f"研究结果:{task}的相关信息..."

class WriterAgent:
    """写作Agent - 负责内容创作"""
    def execute(self, task, context=None):
        # 基于研究结果写作
        research = context.get("research", "")
        return f"基于{research}创作的内容..."

class ReviewerAgent:
    """审核Agent - 负责质量检查"""
    def execute(self, task, context=None):
        # 审核内容质量
        content = context.get("content", "")
        return f"审核意见:{content}需要改进的地方..."

class ManagerAgent:
    """管理Agent - 负责任务规划和协调"""
    def create_plan(self, task):
        return {
            "steps": [
                {"id": "research", "agent": "researcher", "task": f"研究{task}"},
                {"id": "write", "agent": "writer", "task": f"撰写{task}"},
                {"id": "review", "agent": "reviewer", "task": f"审核{task}"}
            ]
        }
    
    def summarize(self, results):
        return f"任务完成,最终成果:{results}"

# 构建多Agent系统
system = MultiAgentSystem()
system.register_agent("manager", ManagerAgent())
system.register_agent("researcher", ResearchAgent())
system.register_agent("writer", WriterAgent())
system.register_agent("reviewer", ReviewerAgent())

# 执行复杂任务
result = system.run_task("写一篇关于人工智能的文章")
print(result)

Agent记忆系统

让Agent记住和学习

短期记忆实现

class ShortTermMemory:
    def __init__(self, max_size=10):
        self.buffer = []
        self.max_size = max_size
    
    def add(self, item):
        self.buffer.append(item)
        if len(self.buffer) > self.max_size:
            self.buffer.pop(0)
    
    def get_context(self):
        return "\n".join(self.buffer[-5:])

长期记忆实现

class LongTermMemory:
    def __init__(self, vector_db):
        self.vector_db = vector_db
    
    def store(self, content, metadata):
        embedding = create_embedding(content)
        self.vector_db.insert(
            embedding, content, metadata
        )
    
    def recall(self, query, k=5):
        results = self.vector_db.search(query, k)
        return results

Agent框架对比

框架特点易用性扩展性适用场景
LangChain功能全面,生态丰富⭐⭐⭐⭐⭐⭐⭐⭐通用开发
AutoGPT完全自主⭐⭐⭐⭐⭐自动化任务
AutoGen多Agent协作⭐⭐⭐⭐⭐⭐⭐⭐团队协作
CrewAI角色定义清晰⭐⭐⭐⭐⭐⭐⭐⭐流程自动化
BabyAGI任务驱动⭐⭐⭐⭐⭐⭐目标导向

Agent评估指标

如何评价Agent的性能

⚡ 效率指标

  • • 任务完成时间
  • • 步骤数量
  • • 资源消耗
  • • API调用次数

✅ 准确性指标

  • • 任务成功率
  • • 错误率
  • • 结果质量
  • • 一致性

🤝 交互指标

  • • 响应时间
  • • 理解准确度
  • • 用户满意度
  • • 澄清次数

实战案例:客服Agent

构建智能客服Agent

class CustomerServiceAgent:
    """智能客服Agent"""
    
    def __init__(self, llm, knowledge_base, order_system):
        self.llm = llm
        self.knowledge_base = knowledge_base
        self.order_system = order_system
        self.conversation_history = []
        
    def handle_query(self, user_query):
        """处理用户查询"""
        # 意图识别
        intent = self.identify_intent(user_query)
        
        # 根据意图路由
        if intent == "order_tracking":
            return self.track_order(user_query)
        elif intent == "product_inquiry":
            return self.product_info(user_query)
        elif intent == "complaint":
            return self.handle_complaint(user_query)
        else:
            return self.general_response(user_query)
    
    def identify_intent(self, query):
        """识别用户意图"""
        prompt = f"""分析以下用户查询的意图:
        
查询:{query}

可能的意图:
- order_tracking: 订单查询
- product_inquiry: 产品咨询  
- complaint: 投诉建议
- general: 一般问题

返回最匹配的意图类型:"""
        
        response = self.llm.generate(prompt)
        return response.strip().lower()
    
    def track_order(self, query):
        """订单追踪"""
        # 提取订单号
        order_id = self.extract_order_id(query)
        
        if order_id:
            # 查询订单系统
            order_info = self.order_system.get_order(order_id)
            
            if order_info:
                return f"""您的订单 {order_id} 当前状态:
📦 状态:{order_info['status']}
📍 位置:{order_info['location']}  
📅 预计送达:{order_info['eta']}

如需其他帮助,请随时告诉我。"""
            else:
                return f"抱歉,未找到订单号 {order_id}。请确认订单号是否正确。"
        else:
            return "请提供您的订单号,我来帮您查询。"
    
    def product_info(self, query):
        """产品信息查询"""
        # 从知识库检索
        relevant_info = self.knowledge_base.search(query, top_k=3)
        
        if relevant_info:
            response = self.llm.generate(f"""
基于以下产品信息回答用户问题:

信息:{relevant_info}

用户问题:{query}

请提供准确、有帮助的回答:""")
            
            return response
        else:
            return "抱歉,暂时没有找到相关产品信息。我可以为您转接人工客服。"
    
    def handle_complaint(self, query):
        """处理投诉"""
        # 记录投诉
        complaint_id = self.log_complaint(query)
        
        # 生成同理心回复
        response = self.llm.generate(f"""
用户投诉:{query}

请生成一个富有同理心的回复,包括:
1. 表达理解和歉意
2. 说明处理流程
3. 提供投诉编号:{complaint_id}
4. 承诺跟进

回复:""")
        
        return response
    
    def general_response(self, query):
        """一般问题回复"""
        # 结合历史对话
        context = "\n".join(self.conversation_history[-5:])
        
        response = self.llm.generate(f"""
对话历史:
{context}

用户问题:{query}

请提供友好、专业的回答:""")
        
        self.conversation_history.append(f"用户:{query}")
        self.conversation_history.append(f"客服:{response}")
        
        return response

# 使用示例
agent = CustomerServiceAgent(
    llm=llm_client,
    knowledge_base=vector_db,
    order_system=order_api
)

# 处理各种查询
print(agent.handle_query("我的订单12345什么时候到?"))
print(agent.handle_query("这个产品的保修期是多久?"))
print(agent.handle_query("产品质量有问题,我要投诉!"))

构建您的智能Agent

LLM Agent技术让AI从简单的问答工具进化为能够自主完成复杂任务的智能助手。 通过LLM API,您可以快速构建强大的Agent系统。

开始构建Agent