← 返回首页
🧠

LLM Agent:让大语言模型使用工具和执行任务

📂 llm ⏱ 4 min 606 words

LLM Agent:让大语言模型使用工具和执行任务

什么是LLM Agent?

LLM Agent是基于大语言模型的智能代理,能够:

Agent vs 普通LLM

特性 普通LLM LLM Agent
输入 文本 文本 + 工具结果
输出 文本 文本 + 工具调用
能力 生成文本 执行任务
交互 单轮/多轮 自主循环

Agent架构

1. 感知-思考-行动循环

class LLMAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.memory = []
    
    def run(self, user_input, max_iterations=10):
        self.memory.append({"role": "user", "content": user_input})
        
        for _ in range(max_iterations):
            # 思考:LLM决定下一步
            response = self.llm.chat(self.memory)
            
            # 检查是否有工具调用
            if response.tool_calls:
                self.memory.append(response.message)
                
                # 执行工具
                for tool_call in response.tool_calls:
                    result = self.execute_tool(tool_call)
                    self.memory.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
            else:
                # 没有工具调用,返回最终答案
                return response.content
        
        return "达到最大迭代次数"
    
    def execute_tool(self, tool_call):
        tool = self.tools[tool_call.function.name]
        args = json.loads(tool_call.function.arguments)
        return tool.execute(**args)

2. ReAct模式

ReAct(Reasoning + Acting)让LLM交替进行推理和行动。

REACT_PROMPT = """
你是一个AI助手,可以使用以下工具:

{tools}

请按照以下格式回答问题:
Thought: 我需要思考什么
Action: 工具名称
Action Input: 工具输入

如果你已经有了答案,请使用:
Thought: 我现在知道答案了
Final Answer: 最终答案

问题:{question}
"""

class ReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
    
    def run(self, question):
        prompt = REACT_PROMPT.format(
            tools=self.format_tools(),
            question=question
        )
        
        while True:
            response = self.llm.generate(prompt)
            
            # 解析响应
            thought, action, action_input = self.parse_response(response)
            
            if action == "Final Answer":
                return action_input
            
            # 执行工具
            tool_result = self.execute_tool(action, action_input)
            
            # 更新prompt
            prompt += f"\n{response}\nObservation: {tool_result}\n"

工具系统

工具定义

from typing import Callable, Any
from dataclasses import dataclass

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict
    execute: Callable

# 定义工具
search_tool = Tool(
    name="search",
    description="搜索互联网获取信息",
    parameters={
        "query": {"type": "string", "description": "搜索关键词"}
    },
    execute=lambda query: search_internet(query)
)

calculator_tool = Tool(
    name="calculator",
    description="执行数学计算",
    parameters={
        "expression": {"type": "string", "description": "数学表达式"}
    },
    execute=lambda expression: eval(expression)
)

python_tool = Tool(
    name="python",
    description="执行Python代码",
    parameters={
        "code": {"type": "string", "description": "Python代码"}
    },
    execute=lambda code: exec(code)
)

工具注册

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, tool: Tool):
        self.tools[tool.name] = tool
    
    def get_tool_schemas(self):
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            }
            for tool in self.tools.values()
        ]
    
    def execute(self, name: str, arguments: dict):
        if name not in self.tools:
            raise ValueError(f"未知工具: {name}")
        
        tool = self.tools[name]
        return tool.execute(**arguments)

记忆系统

短期记忆

class ShortTermMemory:
    def __init__(self, max_tokens=4000):
        self.messages = []
        self.max_tokens = max_tokens
    
    def add(self, message):
        self.messages.append(message)
        self.trim()
    
    def trim(self):
        # 保留最近的消息,确保不超过token限制
        while self.count_tokens() > self.max_tokens:
            self.messages.pop(0)
    
    def count_tokens(self):
        return sum(len(m.get("content", "")) for m in self.messages) // 4
    
    def get_context(self):
        return self.messages

长期记忆

class LongTermMemory:
    def __init__(self, vector_db):
        self.vector_db = vector_db
    
    def store(self, content, metadata=None):
        embedding = self.get_embedding(content)
        self.vector_db.insert(embedding, content, metadata)
    
    def search(self, query, top_k=5):
        query_embedding = self.get_embedding(query)
        return self.vector_db.search(query_embedding, top_k)
    
    def get_embedding(self, text):
        # 使用embedding模型
        return embedding_model.encode(text)

实际应用案例

1. 代码助手

class CodeAssistant:
    def __init__(self, llm):
        self.llm = llm
        self.tools = [
            Tool("read_file", "读取文件内容", ...),
            Tool("write_file", "写入文件", ...),
            Tool("execute_code", "执行代码", ...),
            Tool("search_docs", "搜索文档", ...)
        ]
    
    def solve_coding_problem(self, problem_description):
        # 分析问题
        analysis = self.llm.chat(f"""
        分析以下编程问题:
        {problem_description}
        
        请提供:
        1. 问题理解
        2. 解决方案思路
        3. 需要的步骤
        """)
        
        # 逐步实现
        solution = self.implement_solution(analysis)
        
        # 测试验证
        test_results = self.test_solution(solution)
        
        return solution, test_results

2. 研究助手

class ResearchAssistant:
    def __init__(self, llm):
        self.llm = llm
        self.tools = [
            Tool("search_paper", "搜索学术论文", ...),
            Tool("read_paper", "阅读论文内容", ...),
            Tool("summarize", "总结摘要", ...),
            Tool("create_report", "生成报告", ...)
        ]
    
    def research_topic(self, topic):
        # 搜索相关论文
        papers = self.search_papers(topic)
        
        # 阅读和总结
        summaries = []
        for paper in papers[:10]:
            content = self.read_paper(paper)
            summary = self.summarize(content)
            summaries.append(summary)
        
        # 生成研究报告
        report = self.create_report(topic, summaries)
        
        return report

Agent框架

LangChain Agent

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI

# 定义工具
tools = [
    Tool(
        name="Search",
        func=search_function,
        description="搜索互联网获取最新信息"
    ),
    Tool(
        name="Calculator",
        func=calculator_function,
        description="执行数学计算"
    )
]

# 初始化Agent
llm = OpenAI(temperature=0)
agent = initialize_agent(tools, llm, agent="zero-shot-react-description")

# 运行
result = agent.run("今天北京的天气怎么样?")

AutoGPT

class AutoGPT:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.memory = []
        self.goals = []
    
    def set_goals(self, goals):
        self.goals = goals
    
    def think(self):
        prompt = f"""
        当前目标:{self.goals}
        记忆:{self.memory[-5:]}
        
        请决定下一步行动。
        """
        return self.llm.generate(prompt)
    
    def act(self, action):
        # 解析并执行行动
        pass
    
    def run(self):
        while not self.goals_achieved():
            thought = self.think()
            action = self.parse_thought(thought)
            result = self.act(action)
            self.memory.append({"thought": thought, "action": action, "result": result})

Agent的挑战

1. 可靠性

Agent可能做出错误决策或调用错误工具。

2. 安全性

需要防止Agent执行危险操作。

3. 效率

多轮交互可能消耗大量token。

4. 可解释性

Agent的决策过程可能难以理解。

总结

LLM Agent是大语言模型的高级应用形态,通过工具使用和自主决策,能够完成复杂的实际任务。理解Agent的架构设计、工具系统和记忆机制,对于构建实用的AI系统至关重要。