架构师AI杜

Appearance

第6天：AI Agent兴起（2023-2024） ​

学习目标 ​

  • 理解AI Agent的概念和架构
  • 掌握LangChain框架的核心组件
  • 了解AutoGPT和BabyAGI的原理
  • 理解RAG（检索增强生成）技术
  • 掌握AI Agent的核心能力

课程内容 ​

1. AI Agent的概念 ​

1.1 什么是AI Agent ​

定义：AI Agent是一个能够感知环境、进行推理、执行行动并实现目标的智能系统。

核心特征：

  • 感知（Perception）：从环境中获取信息
  • 推理（Reasoning）：基于信息进行思考和规划
  • 行动（Action）：执行具体的操作
  • 记忆（Memory）：存储和检索信息
  • 学习（Learning）：从经验中改进

与传统AI的区别：

传统AI：
输入 → 模型 → 输出

AI Agent：
环境 → 感知 → 推理 → 行动 → 环境
              ↑         ↓
              ← 记忆 ←

1.2 AI Agent的架构 ​

核心组件：

1.2.1 感知模块（Perception）

作用：从环境中获取信息。

输入类型：

  • 文本
  • 图像
  • 音频
  • 传感器数据
  • API响应

示例：

python
class PerceptionModule:
    def __init__(self):
        self.text_input = None
        self.image_input = None
        self.audio_input = None
    
    def perceive(self, input_data):
        """感知环境"""
        if isinstance(input_data, str):
            self.text_input = input_data
        elif isinstance(input_data, Image):
            self.image_input = input_data
        elif isinstance(input_data, Audio):
            self.audio_input = input_data
        
        return {
            "text": self.text_input,
            "image": self.image_input,
            "audio": self.audio_input
        }

1.2.2 推理模块（Reasoning）

作用：基于感知信息进行思考和规划。

推理类型：

  • 演绎推理（Deductive）：从一般到特殊
  • 归纳推理（Inductive）：从特殊到一般
  • 溯因推理（Abductive）：从结果到原因

规划方法：

  • 目标分解：将大目标分解为小目标
  • 任务规划：确定执行顺序
  • 资源分配：分配时间和资源

示例：

python
class ReasoningModule:
    def __init__(self, llm):
        self.llm = llm
        self.memory = []
    
    def reason(self, perception, goal):
        """推理和规划"""
        # 从记忆中检索相关信息
        relevant_info = self.retrieve_from_memory(perception)
        
        # 构建推理Prompt
        prompt = f"""
        目标：{goal}
        感知信息：{perception}
        相关信息：{relevant_info}
        
        请分析当前情况，制定行动计划。
        """
        
        # 调用LLM进行推理
        response = self.llm.generate(prompt)
        
        # 解析行动计划
        action_plan = self.parse_action_plan(response)
        
        return action_plan
    
    def retrieve_from_memory(self, query):
        """从记忆中检索"""
        # 实现记忆检索逻辑
        pass
    
    def parse_action_plan(self, response):
        """解析行动计划"""
        # 实现解析逻辑
        pass

1.2.3 行动模块（Action）

作用：执行具体的操作。

行动类型：

  • 文本生成：生成回复、文章等
  • API调用：调用外部API
  • 文件操作：读写文件
  • 网络操作：发送请求、爬取数据
  • 工具使用：使用各种工具

示例：

python
class ActionModule:
    def __init__(self):
        self.tools = {
            "text_generation": self.text_generation,
            "api_call": self.api_call,
            "file_operation": self.file_operation,
            "web_search": self.web_search
        }
    
    def execute(self, action_type, action_params):
        """执行行动"""
        if action_type in self.tools:
            return self.tools[action_type](**action_params)
        else:
            raise ValueError(f"Unknown action type: {action_type}")
    
    def text_generation(self, prompt, **kwargs):
        """文本生成"""
        # 实现文本生成逻辑
        pass
    
    def api_call(self, url, method="GET", **kwargs):
        """API调用"""
        # 实现API调用逻辑
        pass
    
    def file_operation(self, operation, file_path, **kwargs):
        """文件操作"""
        # 实现文件操作逻辑
        pass
    
    def web_search(self, query, **kwargs):
        """网络搜索"""
        # 实现网络搜索逻辑
        pass

1.2.4 记忆模块（Memory）

作用：存储和检索信息。

记忆类型：

  • 短期记忆（Short-term Memory）：当前对话上下文
  • 长期记忆（Long-term Memory）：持久化存储
  • 工作记忆（Working Memory）：临时存储推理过程
  • 向量记忆（Vector Memory）：基于向量相似度的记忆

示例：

python
class MemoryModule:
    def __init__(self):
        self.short_term = []
        self.long_term = {}
        self.vector_memory = VectorDatabase()
    
    def add_to_short_term(self, information):
        """添加到短期记忆"""
        self.short_term.append({
            "content": information,
            "timestamp": time.time()
        })
    
    def add_to_long_term(self, key, value):
        """添加到长期记忆"""
        self.long_term[key] = {
            "value": value,
            "timestamp": time.time()
        }
    
    def add_to_vector_memory(self, text, metadata=None):
        """添加到向量记忆"""
        embedding = self.get_embedding(text)
        self.vector_memory.add(
            text=text,
            embedding=embedding,
            metadata=metadata
        )
    
    def retrieve_from_short_term(self, query):
        """从短期记忆检索"""
        # 实现检索逻辑
        pass
    
    def retrieve_from_long_term(self, key):
        """从长期记忆检索"""
        return self.long_term.get(key)
    
    def retrieve_from_vector_memory(self, query, top_k=5):
        """从向量记忆检索"""
        query_embedding = self.get_embedding(query)
        results = self.vector_memory.search(
            query_embedding,
            top_k=top_k
        )
        return results
    
    def get_embedding(self, text):
        """获取文本嵌入"""
        # 实现嵌入逻辑
        pass

1.3 AI Agent的核心能力 ​

1.3.1 感知能力

文本感知：

  • 理解自然语言
  • 提取关键信息
  • 识别意图

图像感知：

  • 识别物体
  • 理解场景
  • 提取信息

音频感知：

  • 语音识别
  • 情感分析
  • 音频特征提取

1.3.2 推理能力

逻辑推理：

  • 演绎推理
  • 归纳推理
  • 溯因推理

规划能力：

  • 目标分解
  • 任务规划
  • 资源分配

决策能力：

  • 多方案比较
  • 风险评估
  • 最优选择

1.3.3 行动能力

工具使用：

  • 调用API
  • 使用工具
  • 执行操作

内容生成：

  • 文本生成
  • 图像生成
  • 代码生成

交互能力：

  • 与用户交互
  • 与其他Agent交互
  • 与环境交互

1.3.4 记忆能力

信息存储：

  • 短期记忆
  • 长期记忆
  • 向量记忆

信息检索：

  • 关键词检索
  • 语义检索
  • 关联检索

信息更新：

  • 新信息添加
  • 旧信息更新
  • 无效信息删除

1.3.5 学习能力

经验学习：

  • 从成功中学习
  • 从失败中学习
  • 持续改进

反馈学习：

  • 用户反馈
  • 环境反馈
  • 自我反思

2. LangChain框架 ​

2.1 LangChain简介 ​

定义：LangChain是一个用于构建基于大语言模型应用的框架。

核心思想：

  • 将LLM与其他组件结合
  • 构建复杂的应用
  • 提供标准化的接口

主要组件：

  • Models：LLM和Chat模型
  • Prompts：提示管理
  • Memory：记忆管理
  • Chains：链式调用
  • Agents：智能体
  • Tools：工具集成

2.2 LangChain核心组件 ​

2.2.1 Models

LLM（Language Model）：

python
from langchain.llms import OpenAI

# 创建LLM
llm = OpenAI(temperature=0.9)

# 调用LLM
text = "What would be a good company name for a company that makes colorful socks?"
print(llm(text))

Chat Model：

python
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

# 创建Chat模型
chat = ChatOpenAI(temperature=0)

# 调用Chat模型
messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="What is the capital of France?")
]
print(chat(messages))

2.2.2 Prompts

PromptTemplate：

python
from langchain.prompts import PromptTemplate

# 创建Prompt模板
prompt = PromptTemplate(
    input_variables=["product"],
    template="What is a good name for a company that makes {product}?"
)

# 使用Prompt模板
formatted_prompt = prompt.format(product="colorful socks")
print(formatted_prompt)

ChatPromptTemplate：

python
from langchain.prompts.chat import ChatPromptTemplate

# 创建Chat Prompt模板
template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "{user_input}")
])

# 使用Chat Prompt模板
formatted_template = template.format_messages(
    user_input="What is the capital of France?"
)
print(formatted_template)

2.2.3 Memory

ConversationBufferMemory：

python
from langchain.memory import ConversationBufferMemory

# 创建记忆
memory = ConversationBufferMemory()

# 添加对话
memory.save_context({"input": "Hi"}, {"output": "What's up?"})
memory.save_context({"input": "Not much, just hanging"}, {"output": "Cool"})

# 获取对话历史
print(memory.load_memory_variables({}))

ConversationSummaryMemory：

python
from langchain.memory import ConversationSummaryMemory

# 创建摘要记忆
memory = ConversationSummaryMemory(llm=llm)

# 添加对话
memory.save_context({"input": "Hi"}, {"output": "What's up?"})

# 获取摘要
print(memory.load_memory_variables({}))

2.2.4 Chains

Simple Chain：

python
from langchain.chains import LLMChain

# 创建链
chain = LLMChain(llm=llm, prompt=prompt)

# 运行链
print(chain.run("colorful socks"))

Sequential Chain：

python
from langchain.chains import SimpleSequentialChain

# 创建第一个链
first_chain = LLMChain(llm=llm, prompt=prompt)

# 创建第二个链
second_prompt = PromptTemplate(
    input_variables=["company_name"],
    template="Write a catchphrase for the company: {company_name}"
)
second_chain = LLMChain(llm=llm, prompt=second_prompt)

# 创建顺序链
overall_chain = SimpleSequentialChain(
    chains=[first_chain, second_chain],
    verbose=True
)

# 运行顺序链
print(overall_chain.run("colorful socks"))

2.2.5 Agents

ReAct Agent：

python
from langchain.agents import initialize_agent, Tool, AgentType
from langchain.tools import SerpAPIWrapper

# 创建工具
search = SerpAPIWrapper()
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="useful for when you need to answer questions about current events"
    )
]

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 运行Agent
print(agent.run("What is the weather in Paris?"))

2.2.6 Tools

内置工具：

python
from langchain.tools import (
    SerpAPIWrapper,
    PythonREPL,
    ShellTool,
    RequestsGetTool,
    # ... 更多工具
)

# 搜索工具
search = SerpAPIWrapper()

# Python工具
python_repl = PythonREPL()

# Shell工具
shell = ShellTool()

# HTTP请求工具
requests_get = RequestsGetTool()

自定义工具：

python
from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

class CalculatorTool(BaseTool):
    name = "Calculator"
    description = "useful for when you need to answer questions about math"
    args_schema = CalculatorInput
    
    def _run(self, a: int, b: int):
        return a + b

# 使用自定义工具
calculator = CalculatorTool()
print(calculator.run(a=1, b=2))

2.3 LangChain应用示例 ​

2.3.1 简单问答系统

python
from langchain.chains import ConversationalRetrievalChain
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_texts(
    ["Python is a programming language", "JavaScript is also a programming language"],
    embeddings
)

# 创建检索链
retriever = vectorstore.as_retriever()
qa_chain = ConversationalRetrievalChain.from_llm(
    ChatOpenAI(temperature=0),
    retriever=retriever
)

# 问答
chat_history = []
query = "What is Python?"
result = qa_chain({"question": query, "chat_history": chat_history})
print(result["answer"])

2.3.2 文档问答系统

python
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains import RetrievalQA

# 加载文档
loader = TextLoader("document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 创建向量存储
vectorstore = Chroma.from_documents(texts, embeddings)

# 创建问答链
qa_chain = RetrievalQA.from_chain_type(
    ChatOpenAI(temperature=0),
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)

# 问答
query = "What is the main topic of the document?"
result = qa_chain.run(query)
print(result)

3. AutoGPT ​

3.1 AutoGPT简介 ​

定义：AutoGPT是一个自主的AI Agent，能够自动完成复杂的任务。

核心思想：

  • 给定一个目标
  • 自动分解任务
  • 自动执行任务
  • 自动评估结果
  • 自动调整策略

特点：

  • 完全自主
  • 目标导向
  • 自我反思
  • 持续改进

3.2 AutoGPT的工作流程 ​

核心循环：

1. 思考（Thought）
   - 分析当前情况
   - 确定下一步行动

2. 推理（Reasoning）
   - 评估不同方案
   - 选择最优方案

3. 行动（Action）
   - 执行选定的行动
   - 调用工具或API

4. 观察（Observation）
   - 观察行动结果
   - 收集反馈信息

5. 反思（Reflection）
   - 评估行动效果
   - 调整策略

6. 重复（Repeat）
   - 返回步骤1
   - 直到目标达成

示例：

python
class AutoGPT:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.goal = None
    
    def set_goal(self, goal):
        """设置目标"""
        self.goal = goal
    
    def run(self, max_iterations=10):
        """运行AutoGPT"""
        for iteration in range(max_iterations):
            # 思考
            thought = self.think()
            print(f"Iteration {iteration + 1}: {thought}")
            
            # 推理
            reasoning = self.reason(thought)
            print(f"Reasoning: {reasoning}")
            
            # 行动
            action = self.decide_action(reasoning)
            print(f"Action: {action}")
            
            # 执行
            observation = self.execute_action(action)
            print(f"Observation: {observation}")
            
            # 反思
            reflection = self.reflect(observation)
            print(f"Reflection: {reflection}")
            
            # 检查是否达成目标
            if self.check_goal():
                print("Goal achieved!")
                break
    
    def think(self):
        """思考"""
        prompt = f"""
        目标：{self.goal}
        当前状态：{self.memory.get_current_state()}
        
        请分析当前情况，确定下一步应该做什么。
        """
        response = self.llm.generate(prompt)
        return response
    
    def reason(self, thought):
        """推理"""
        prompt = f"""
        思考：{thought}
        可用工具：{[tool.name for tool in self.tools]}
        
        请评估不同的行动方案，选择最优方案。
        """
        response = self.llm.generate(prompt)
        return response
    
    def decide_action(self, reasoning):
        """决定行动"""
        # 解析推理结果，确定行动
        pass
    
    def execute_action(self, action):
        """执行行动"""
        # 执行行动，返回观察结果
        pass
    
    def reflect(self, observation):
        """反思"""
        prompt = f"""
        观察结果：{observation}
        
        请评估行动效果，提出改进建议。
        """
        response = self.llm.generate(prompt)
        return response
    
    def check_goal(self):
        """检查是否达成目标"""
        # 检查目标是否达成
        pass

3.3 AutoGPT的应用 ​

应用场景：

  • 自动编程
  • 自动研究
  • 自动创作
  • 自动决策

示例任务：

目标：创建一个简单的网页应用

AutoGPT会自动：
1. 分析需求
2. 设计架构
3. 编写代码
4. 测试代码
5. 部署应用

4. BabyAGI ​

4.1 BabyAGI简介 ​

定义：BabyAGI是一个任务驱动的AI Agent，能够自动管理和执行任务。

核心思想：

  • 创建任务列表
  • 按优先级执行任务
  • 动态调整任务
  • 持续优化

特点：

  • 任务驱动
  • 优先级管理
  • 动态调整
  • 持续优化

4.2 BabyAGI的工作流程 ​

核心循环：

1. 创建任务（Create Tasks）
   - 基于目标创建初始任务
   - 将任务添加到任务列表

2. 选择任务（Select Task）
   - 从任务列表中选择优先级最高的任务

3. 执行任务（Execute Task）
   - 执行选定的任务
   - 调用工具或API

4. 评估结果（Evaluate Result）
   - 评估任务执行结果
   - 判断是否需要调整

5. 更新任务（Update Tasks）
   - 根据结果更新任务列表
   - 添加新任务或修改现有任务

6. 重复（Repeat）
   - 返回步骤2
   - 直到所有任务完成

示例：

python
class BabyAGI:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.task_list = []
    
    def set_goal(self, goal):
        """设置目标"""
        self.goal = goal
        self.create_initial_tasks()
    
    def create_initial_tasks(self):
        """创建初始任务"""
        prompt = f"""
        目标：{self.goal}
        
        请将这个目标分解为一系列具体的任务，并按优先级排序。
        """
        response = self.llm.generate(prompt)
        self.task_list = self.parse_tasks(response)
    
    def run(self):
        """运行BabyAGI"""
        while self.task_list:
            # 选择任务
            task = self.select_task()
            print(f"Executing task: {task}")
            
            # 执行任务
            result = self.execute_task(task)
            print(f"Result: {result}")
            
            # 评估结果
            evaluation = self.evaluate_result(result)
            print(f"Evaluation: {evaluation}")
            
            # 更新任务
            self.update_tasks(task, result, evaluation)
    
    def select_task(self):
        """选择任务"""
        # 选择优先级最高的任务
        return self.task_list[0]
    
    def execute_task(self, task):
        """执行任务"""
        # 执行任务，返回结果
        pass
    
    def evaluate_result(self, result):
        """评估结果"""
        prompt = f"""
        任务结果：{result}
        
        请评估任务执行效果，判断是否需要调整。
        """
        response = self.llm.generate(prompt)
        return response
    
    def update_tasks(self, task, result, evaluation):
        """更新任务列表"""
        # 根据评估结果更新任务列表
        pass
    
    def parse_tasks(self, response):
        """解析任务"""
        # 解析LLM返回的任务列表
        pass

5. RAG（检索增强生成） ​

5.1 RAG简介 ​

定义：RAG（Retrieval-Augmented Generation）是一种结合检索和生成的技术，通过检索相关文档来增强生成能力。

核心思想：

  • 检索相关文档
  • 将文档作为上下文
  • 基于上下文生成回答

优势：

  • 减少幻觉
  • 提高准确性
  • 可追溯来源
  • 易于更新知识

5.2 RAG的架构 ​

核心组件：

5.2.1 文档处理

文档加载：

python
from langchain.document_loaders import TextLoader, PDFLoader, WebLoader

# 加载文本文件
loader = TextLoader("document.txt")
documents = loader.load()

# 加载PDF文件
loader = PDFLoader("document.pdf")
documents = loader.load()

# 加载网页
loader = WebLoader("https://example.com")
documents = loader.load()

文档分割：

python
from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter

# 字符分割器
text_splitter = CharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
texts = text_splitter.split_documents(documents)

# 递归字符分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
texts = text_splitter.split_documents(documents)

5.2.2 向量化

文本嵌入：

python
from langchain.embeddings import OpenAIEmbeddings

# 创建嵌入模型
embeddings = OpenAIEmbeddings()

# 获取嵌入
text = "This is a sample text."
embedding = embeddings.embed_query(text)
print(f"Embedding dimension: {len(embedding)}")

向量存储：

python
from langchain.vectorstores import Chroma, FAISS, Pinecone

# Chroma向量存储
vectorstore = Chroma.from_documents(
    documents=texts,
    embedding=embeddings
)

# FAISS向量存储
vectorstore = FAISS.from_documents(
    documents=texts,
    embedding=embeddings
)

# Pinecone向量存储
vectorstore = Pinecone.from_documents(
    documents=texts,
    embedding=embeddings,
    index_name="my-index"
)

5.2.3 检索

相似度检索：

python
# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

# 检索相关文档
query = "What is the main topic?"
docs = retriever.get_relevant_documents(query)

for doc in docs:
    print(f"Content: {doc.page_content}")
    print(f"Metadata: {doc.metadata}")

MMR（Maximal Marginal Relevance）检索：

python
# 创建MMR检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "fetch_k": 10}
)

# 检索相关文档
docs = retriever.get_relevant_documents(query)

5.2.4 生成

RAG链：

python
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 创建Prompt模板
prompt_template = """
Use the following pieces of context to answer the question at the end.
If you don't know the answer, just say that you don't know, don't try to make up an answer.

Context: {context}

Question: {question}

Answer:
"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 创建RAG链
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0),
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    chain_type_kwargs={"prompt": PROMPT}
)

# 问答
query = "What is the main topic?"
result = qa_chain.run(query)
print(result)

5.3 RAG的应用 ​

应用场景：

  • 知识库问答
  • 文档问答
  • 客服系统
  • 研究辅助

示例：企业知识库问答系统

python
from langchain.chains import ConversationalRetrievalChain

# 创建对话式RAG链
qa_chain = ConversationalRetrievalChain.from_llm(
    ChatOpenAI(temperature=0),
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

# 对话
chat_history = []
while True:
    query = input("You: ")
    if query.lower() == "quit":
        break
    
    result = qa_chain({
        "question": query,
        "chat_history": chat_history
    })
    
    print(f"AI: {result['answer']}")
    print(f"Sources: {[doc.metadata for doc in result['source_documents']]}")
    
    chat_history.append((query, result['answer']))

6. AI Agent的应用 ​

6.1 个人助理 ​

功能：

  • 日程管理
  • 邮件处理
  • 信息查询
  • 任务提醒

示例：

python
class PersonalAssistant:
    def __init__(self, llm):
        self.llm = llm
        self.memory = MemoryModule()
        self.tools = {
            "calendar": CalendarTool(),
            "email": EmailTool(),
            "search": SearchTool(),
            "reminder": ReminderTool()
        }
    
    def assist(self, user_input):
        """协助用户"""
        # 感知用户意图
        intent = self.perceive_intent(user_input)
        
        # 推理和规划
        action_plan = self.reason(intent)
        
        # 执行行动
        result = self.execute(action_plan)
        
        # 更新记忆
        self.memory.add_to_short_term({
            "user_input": user_input,
            "intent": intent,
            "result": result
        })
        
        return result
    
    def perceive_intent(self, user_input):
        """感知用户意图"""
        prompt = f"""
        用户输入：{user_input}
        
        请分析用户的意图，确定需要执行什么操作。
        可用操作：{list(self.tools.keys())}
        """
        response = self.llm.generate(prompt)
        return response
    
    def reason(self, intent):
        """推理和规划"""
        # 根据意图制定行动计划
        pass
    
    def execute(self, action_plan):
        """执行行动"""
        # 执行行动计划
        pass

6.2 代码助手 ​

功能：

  • 代码生成
  • 代码审查
  • Bug修复
  • 代码解释

示例：

python
class CodeAssistant:
    def __init__(self, llm):
        self.llm = llm
        self.memory = MemoryModule()
        self.tools = {
            "code_generation": CodeGenerationTool(),
            "code_review": CodeReviewTool(),
            "bug_fix": BugFixTool(),
            "code_explanation": CodeExplanationTool()
        }
    
    def assist(self, user_input, code=None):
        """协助编程"""
        # 感知用户需求
        intent = self.perceive_intent(user_input)
        
        # 推理和规划
        action_plan = self.reason(intent, code)
        
        # 执行行动
        result = self.execute(action_plan)
        
        return result
    
    def perceive_intent(self, user_input):
        """感知用户需求"""
        prompt = f"""
        用户输入：{user_input}
        
        请分析用户的需求，确定需要执行什么操作。
        可用操作：{list(self.tools.keys())}
        """
        response = self.llm.generate(prompt)
        return response
    
    def reason(self, intent, code):
        """推理和规划"""
        # 根据意图和代码制定行动计划
        pass
    
    def execute(self, action_plan):
        """执行行动"""
        # 执行行动计划
        pass

6.3 研究助手 ​

功能：

  • 文献检索
  • 信息总结
  • 数据分析
  • 报告生成

示例：

python
class ResearchAssistant:
    def __init__(self, llm):
        self.llm = llm
        self.memory = MemoryModule()
        self.tools = {
            "search": SearchTool(),
            "summarize": SummarizeTool(),
            "analyze": AnalyzeTool(),
            "report": ReportTool()
        }
    
    def assist(self, research_topic):
        """协助研究"""
        # 检索文献
        literature = self.search_literature(research_topic)
        
        # 总结文献
        summary = self.summarize_literature(literature)
        
        # 分析数据
        analysis = self.analyze_data(summary)
        
        # 生成报告
        report = self.generate_report(analysis)
        
        return report
    
    def search_literature(self, topic):
        """检索文献"""
        # 使用搜索工具检索文献
        pass
    
    def summarize_literature(self, literature):
        """总结文献"""
        # 使用总结工具总结文献
        pass
    
    def analyze_data(self, summary):
        """分析数据"""
        # 使用分析工具分析数据
        pass
    
    def generate_report(self, analysis):
        """生成报告"""
        # 使用报告工具生成报告
        pass

实践任务 ​

任务1：实现简单的AI Agent ​

目标：使用LangChain实现一个简单的AI Agent。

要求：

  1. 定义Agent的目标
  2. 实现感知、推理、行动模块
  3. 集成工具
  4. 测试Agent的能力

代码框架：

python
from langchain.agents import initialize_agent, Tool, AgentType
from langchain.llms import OpenAI
from langchain.tools import SerpAPIWrapper, PythonREPL

# 创建LLM
llm = OpenAI(temperature=0)

# 创建工具
search = SerpAPIWrapper()
python_repl = PythonREPL()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="useful for when you need to answer questions about current events"
    ),
    Tool(
        name="Python",
        func=python_repl.run,
        description="useful for when you need to run Python code"
    )
]

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试Agent
print(agent.run("What is the weather in Paris?"))
print(agent.run("Calculate 123 * 456"))

任务2：实现RAG系统 ​

目标：使用LangChain实现一个RAG系统。

要求：

  1. 加载文档
  2. 创建向量存储
  3. 实现检索和生成
  4. 测试问答效果

代码框架：

python
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI

# 加载文档
loader = TextLoader("document.txt")
documents = loader.load()

# 分割文档
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)

# 创建RAG链
qa_chain = RetrievalQA.from_chain_type(
    ChatOpenAI(temperature=0),
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)

# 测试问答
query = "What is the main topic of the document?"
result = qa_chain.run(query)
print(result)

任务3：实现AutoGPT简化版 ​

目标：实现一个简化版的AutoGPT。

要求：

  1. 定义目标
  2. 实现思考-推理-行动-观察-反思循环
  3. 集成工具
  4. 测试自主执行能力

代码框架：

python
class SimpleAutoGPT:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.memory = []
    
    def set_goal(self, goal):
        """设置目标"""
        self.goal = goal
    
    def run(self, max_iterations=5):
        """运行"""
        for iteration in range(max_iterations):
            print(f"\n=== Iteration {iteration + 1} ===")
            
            # 思考
            thought = self.think()
            print(f"Thought: {thought}")
            
            # 推理
            reasoning = self.reason(thought)
            print(f"Reasoning: {reasoning}")
            
            # 行动
            action, params = self.decide_action(reasoning)
            print(f"Action: {action}")
            print(f"Params: {params}")
            
            # 执行
            observation = self.execute(action, params)
            print(f"Observation: {observation}")
            
            # 反思
            reflection = self.reflect(observation)
            print(f"Reflection: {reflection}")
            
            # 检查是否达成目标
            if self.check_goal():
                print("\nGoal achieved!")
                break
    
    def think(self):
        """思考"""
        prompt = f"""
        目标：{self.goal}
        记忆：{self.memory[-5:] if self.memory else 'None'}
        
        请分析当前情况，确定下一步应该做什么。
        """
        response = self.llm.generate(prompt)
        self.memory.append(("thought", response))
        return response
    
    def reason(self, thought):
        """推理"""
        prompt = f"""
        思考：{thought}
        可用工具：{[tool.name for tool in self.tools]}
        
        请评估不同的行动方案，选择最优方案。
        """
        response = self.llm.generate(prompt)
        self.memory.append(("reasoning", response))
        return response
    
    def decide_action(self, reasoning):
        """决定行动"""
        # 解析推理结果，确定行动和参数
        pass
    
    def execute(self, action, params):
        """执行行动"""
        # 执行行动，返回观察结果
        pass
    
    def reflect(self, observation):
        """反思"""
        prompt = f"""
        观察结果：{observation}
        
        请评估行动效果，提出改进建议。
        """
        response = self.llm.generate(prompt)
        self.memory.append(("reflection", response))
        return response
    
    def check_goal(self):
        """检查是否达成目标"""
        prompt = f"""
        目标：{self.goal}
        当前记忆：{self.memory[-5:] if self.memory else 'None'}
        
        请判断目标是否已经达成。
        """
        response = self.llm.generate(prompt)
        return "yes" in response.lower()

课后作业 ​

作业1：AI Agent设计 ​

题目：设计一个特定领域的AI Agent。

要求：

  1. 选择一个应用领域（如客服、教育、医疗等）
  2. 设计Agent的架构
  3. 定义Agent的能力
  4. 实现原型系统
  5. 测试Agent的能力

作业2：RAG系统优化 ​

题目：优化RAG系统的性能。

要求：

  1. 实现一个RAG系统
  2. 测试不同的检索策略
  3. 测试不同的Prompt
  4. 优化系统性能
  5. 总结优化经验

作业3：AutoGPT应用 ​

题目：使用AutoGPT解决实际问题。

要求：

  1. 选择一个实际任务
  2. 使用AutoGPT自动执行
  3. 分析执行过程
  4. 评估执行效果
  5. 提出改进建议

参考资料 ​

必读文献 ​

  1. Wei, J., et al. (2022). "Chain-of-Thought Prompting Elicits Reasoning in Large Language Models". NeurIPS.

    • Chain-of-Thought论文

  2. Lewis, P., et al. (2020). "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks". NeurIPS.

    • RAG原始论文

  3. Nakano, R., et al. (2021). "WebGPT: Browser-assisted question-answering with human feedback". arXiv.

    • WebGPT论文

推荐阅读 ​

  1. LangChain Documentation: https://python.langchain.com/

    • LangChain官方文档

  2. AutoGPT GitHub: https://github.com/Significant-Gravitas/Auto-GPT

    • AutoGPT项目

  3. BabyAGI GitHub: https://github.com/yoheinakajima/babyagi

    • BabyAGI项目

在线资源 ​

  1. LangChain Tutorials: https://python.langchain.com/docs/get_started/introduction

    • LangChain教程

  2. RAG Tutorial: https://python.langchain.com/docs/use_cases/question_answering

    • RAG教程

  3. Agent Tutorial: https://python.langchain.com/docs/modules/agents/

    • Agent教程

扩展阅读 ​

AI Agent前沿 ​

  • Park, J. S., et al. (2023). "Generative Agents: Interactive Simulacra of Human Behavior". arXiv.

    • 生成式Agent

  • Shinn, N., et al. (2023). "Reflexion: Language Agents with Verbal Reinforcement Learning". arXiv.

    • Reflexion Agent

RAG前沿 ​

  • Lewis, P., et al. (2020). "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks". NeurIPS.

    • RAG论文

  • Karpukhin, V., et al. (2020). "Dense Passage Retrieval for Open-Domain Question Answering". EMNLP.

    • 密集段落检索

下节预告 ​

下一节我们将学习MCP协议发布（2024年底），了解MCP（Model Context Protocol）的诞生背景、核心思想、架构设计和应用场景，以及它如何标准化AI Agent与外部系统的交互。

架构师AI杜公众号二维码

扫描二维码关注"架构师AI杜"公众号，获取更多技术内容和最新动态