架构师AI杜

Appearance

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

学习目标 ​

本节将带领读者理解AI Agent的概念和架构，掌握LangChain框架的核心组件，了解AutoGPT和BabyAGI的原理，理解RAG检索增强生成技术，并掌握AI Agent的核心能力。

课程内容 ​

1. AI Agent的概念 ​

1.1 什么是AI Agent ​

AI Agent是一个能够感知环境、进行推理、执行行动并实现目标的智能系统。核心特征包括感知从环境中获取信息、推理基于信息进行思考和规划、行动执行具体的操作、记忆存储和检索信息、学习从经验中改进。

与传统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）

作用是基于感知信息进行思考和规划。推理类型包括演绎推理从一般到特殊、归纳推理从特殊到一般、溯因推理从结果到原因。规划方法包括目标分解将大目标分解为小目标、任务规划确定执行顺序、资源分配分配时间和资源。

示例：

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）

作用是存储和检索信息。记忆类型包括短期记忆当前对话上下文、长期记忆持久化存储、工作记忆临时存储推理过程、向量记忆基于向量相似度的记忆。

示例：

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 记忆能力

记忆能力是AI Agent持续运行的重要基础。在信息存储方面，Agent需要维护短期记忆以保持当前对话上下文，建立长期记忆以持久化重要信息，并利用向量记忆实现基于语义的快速检索。在信息检索方面，Agent能够通过关键词检索、语义检索和关联检索等多种方式，从海量记忆中快速定位所需信息。在信息更新方面，Agent能够智能地添加新信息、更新旧信息、删除无效信息，确保记忆系统的准确性和时效性。

1.3.5 学习能力

学习能力是AI Agent不断进化的核心动力。在经验学习方面，Agent能够从成功的经验中提炼有效策略，从失败的尝试中吸取教训，通过持续改进不断提升自身性能。在反馈学习方面，Agent重视用户反馈以优化服务质量，关注环境反馈以适应外部变化，进行自我反思以发现并修正不足，从而实现全方位的学习和成长。

2. LangChain框架 ​

2.1 LangChain简介 ​

LangChain是一个专门用于构建基于大语言模型应用的强大框架，其核心思想是将LLM与其他组件有机结合，从而构建出功能复杂、逻辑清晰的应用程序。通过提供标准化的接口和丰富的组件库，LangChain极大地简化了AI应用的开发流程。其主要组件包括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，能够自动完成复杂的任务。其核心思想是给定一个目标后，自动分解任务、自动执行任务、自动评估结果、自动调整策略，从而实现全流程的自主化。AutoGPT的特点包括完全自主能够独立完成任务、目标导向始终围绕目标行动、自我反思能够评估和改进、持续改进不断优化性能。

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的应用场景非常广泛，涵盖了自动编程、自动研究、自动创作、自动决策等多个领域。通过自主的思考和执行能力，AutoGPT能够完成从简单到复杂的各种任务，极大地提升了工作效率和创新能力。

示例任务：

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

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

4. BabyAGI ​

4.1 BabyAGI简介 ​

BabyAGI是一个任务驱动的AI Agent，能够自动管理和执行任务。其核心思想是创建任务列表、按优先级执行任务、动态调整任务、持续优化，从而实现高效的任务管理。BabyAGI的特点包括任务驱动以任务为中心、优先级管理按重要性排序、动态调整根据情况变化、持续优化不断提升效率。

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）是一种创新的结合检索和生成的技术，通过检索相关文档来增强生成能力。其核心思想是检索相关文档、将文档作为上下文、基于上下文生成回答，从而实现更准确、更可靠的生成结果。RAG的优势在于能够减少幻觉现象、提高回答准确性、可追溯信息来源、易于更新知识库。

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的应用 ​

RAG技术的应用场景十分广泛，包括知识库问答、文档问答、客服系统、研究辅助等多个领域。通过将检索和生成相结合，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杜"公众号，获取更多技术内容和最新动态