Agent框架
--- title: "Agent框架" description: "掌握AI Agent框架的设计原理，包括LangChain、AutoGPT等主流框架的使用与架构设计" tags: ["Agent框架", "LangChain", "AutoGPT", "智能体设计"] category: "llm" icon: "🧠"
Agent框架
什么是AI Agent
AI Agent是能够感知环境、做出决策并采取行动的智能系统。与简单的聊天机器人不同，Agent具备自主规划、使用工具、记忆和反思的能力。现代AI Agent通常以大语言模型为核心，结合各种工具和记忆系统来完成复杂任务。
LangChain框架
LangChain是最流行的Agent开发框架之一，提供了完整的工具链。
1. 基础Agent结构
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
class BasicAgent:
def __init__(self, model_name="gpt-4"):
self.llm = ChatOpenAI(model=model_name)
self.tools = []
self.agent_executor = None
def add_tool(self, name, func, description):
"""添加工具"""
tool = Tool(name=name, func=func, description=description)
self.tools.append(tool)
def setup_agent(self, prompt_template=None):
"""设置Agent"""
if prompt_template is None:
prompt_template = PromptTemplate.from_template(
"""Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:{agent_scratchpad}"""
)
self.agent_executor = AgentExecutor(
agent=create_react_agent(self.llm, self.tools, prompt_template),
tools=self.tools,
verbose=True,
max_iterations=10
)
def run(self, query):
"""运行Agent"""
return self.agent_executor.invoke({"input": query})
# 使用示例
agent = BasicAgent()
agent.add_tool(
name="Search",
func=lambda q: f"搜索结果：关于'{q}'的信息",
description="当需要搜索信息时使用此工具"
)
agent.add_tool(
name="Calculator",
func=lambda expr: str(eval(expr)),
description="当需要进行数学计算时使用此工具"
)
agent.setup_agent()
result = agent.run("今天天气怎么样？帮我计算一下2+3")
2. 工具定义
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional
class SearchInput(BaseModel):
query: str = Field(description="搜索查询")
class SearchTool(BaseTool):
name = "search"
description = "搜索互联网获取信息"
args_schema = SearchInput
def _run(self, query: str) -> str:
"""执行搜索"""
# 这里可以接入实际的搜索API
return f"搜索'{query}'的结果：这里是模拟的搜索结果"
async def _arun(self, query: str) -> str:
"""异步执行搜索"""
return self._run(query)
class DatabaseInput(BaseModel):
query: str = Field(description="SQL查询")
class DatabaseTool(BaseTool):
name = "database"
description = "查询数据库获取信息"
args_schema = DatabaseInput
def _run(self, query: str) -> str:
"""执行数据库查询"""
return f"执行SQL: {query} - 返回结果：模拟数据"
# 创建工具集
tools = [SearchTool(), DatabaseTool()]
AutoGPT架构
import json
from typing import List, Dict
from dataclasses import dataclass
@dataclass
class Task:
id: str
description: str
status: str # pending, in_progress, completed, failed
result: Optional[str] = None
class AutoGPTAgent:
def __init__(self, llm, tools, memory=None):
self.llm = llm
self.tools = {tool.name: tool for tool in tools}
self.memory = memory or []
self.task_queue: List[Task] = []
self.completed_tasks: List[Task] = []
def think(self, context: str) -> Dict:
"""思考下一步行动"""
prompt = f"""你是一个自主AI助手。根据以下上下文，决定下一步行动。
上下文：
{context}
可用工具：{list(self.tools.keys())}
请以JSON格式输出你的思考：
{{
"thought": "你的思考过程",
"action": "要执行的动作",
"action_input": "动作的输入参数",
"next_step": "接下来的计划"
}}"""
response = self.llm.predict(prompt)
return json.loads(response)
def execute_action(self, action: str, action_input: str) -> str:
"""执行动作"""
if action in self.tools:
return self.tools[action]._run(action_input)
elif action == "finish":
return action_input
else:
return f"未知动作: {action}"
def run(self, goal: str, max_iterations: int = 10):
"""运行Agent"""
context = f"目标：{goal}\n\n历史记录："
for i in range(max_iterations):
print(f"\n=== 迭代 {i+1} ===")
# 思考
thought = self.think(context)
print(f"思考: {thought['thought']}")
print(f"动作: {thought['action']}")
# 执行
result = self.execute_action(thought['action'], thought['action_input'])
print(f"结果: {result}")
# 更新上下文
context += f"\n思考: {thought['thought']}"
context += f"\n动作: {thought['action']}"
context += f"\n结果: {result}"
# 存入记忆
self.memory.append({
"iteration": i + 1,
"thought": thought,
"result": result
})
# 检查是否完成
if thought['action'] == 'finish':
return result
return "达到最大迭代次数"
记忆系统
from collections import deque
import numpy as np
class MemorySystem:
def __init__(self, max_size=100):
self.short_term_memory = deque(maxlen=max_size)
self.long_term_memory = []
self.embeddings = []
def add(self, content: str, embedding=None):
"""添加记忆"""
entry = {
"content": content,
"timestamp": len(self.short_term_memory)
}
self.short_term_memory.append(entry)
if embedding is not None:
self.embeddings.append(embedding)
if len(self.embeddings) > max_size:
self.embeddings.pop(0)
def search(self, query_embedding, top_k=5):
"""基于语义搜索记忆"""
if not self.embeddings:
return []
# 计算余弦相似度
similarities = []
for i, emb in enumerate(self.embeddings):
sim = np.dot(query_embedding, emb) / (
np.linalg.norm(query_embedding) * np.linalg.norm(emb)
)
similarities.append((i, sim))
# 排序并返回top-k
similarities.sort(key=lambda x: x[1], reverse=True)
return [
self.short_term_memory[i]["content"]
for i, _ in similarities[:top_k]
]
def get_recent(self, k=5):
"""获取最近k条记忆"""
return [
entry["content"]
for entry in list(self.short_term_memory)[-k:]
]
def summarize(self):
"""总结所有记忆"""
return "\n".join([
f"- {entry['content']}"
for entry in self.short_term_memory
])
总结
AI Agent框架使得构建复杂的智能系统变得更加容易。掌握LangChain、AutoGPT等框架的使用，以及记忆系统的设计，是构建高效Agent的关键。