阶段 6：成熟 Agent 框架实践

📖 本阶段目标

掌握主流 Agent 开发框架
学习 LangChain/LlamaIndex 的核心功能
理解框架的工具、记忆、执行器等组件
实践用框架构建复杂 Agent 应用

🚀 主流框架概览

框架对比

框架	特点	适用场景	学习曲线
LangChain	功能全面，生态丰富	通用 Agent 开发	中等
LlamaIndex	专注数据索引和 RAG	知识库问答	较低
AutoGen	多 Agent 对话	复杂协作系统	中等
CrewAI	角色化 Agent 团队	任务导向协作	较低
LangGraph	状态图流程控制	复杂工作流	较高

🔗 LangChain 实践

1. 基础概念

# examples/stage6/langchain_basics.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough

# 1. 创建 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# 2. 创建 Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}。"),
    ("user", "{input}")
])

# 3. 创建输出解析器
output_parser = StrOutputParser()

# 4. 构建链（Chain）
chain = prompt | llm | output_parser

# 5. 运行
result = chain.invoke({
    "role": "Python 编程助手",
    "input": "如何读取 JSON 文件？"
})

print(result)

2. 工具集成

# examples/stage6/langchain_tools.py
from langchain.agents import Tool, AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
import requests

# 定义工具函数
def get_weather(city: str) -> str:
    """获取天气信息"""
    # 实际应调用真实API
    return f"{city}的天气：晴，温度25°C"

def search_web(query: str) -> str:
    """网络搜索"""
    # 实际应调用搜索API
    return f"关于'{query}'的搜索结果..."

def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except:
        return "计算错误"

# 创建工具列表
tools = [
    Tool(
        name="get_weather",
        func=get_weather,
        description="获取指定城市的天气信息。输入应该是城市名称。"
    ),
    Tool(
        name="search_web",
        func=search_web,
        description="在网络上搜索信息。输入应该是搜索查询。"
    ),
    Tool(
        name="calculate",
        func=calculate,
        description="计算数学表达式。输入应该是数学表达式，如 '2 + 2'。"
    )
]

# 创建 Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手，可以使用工具来回答问题。"),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 使用 Agent
if __name__ == "__main__":
    result = agent_executor.invoke({
        "input": "北京的天气怎么样？如果温度超过20度，帮我计算 (25 + 5) * 2"
    })

    print("\n最终结果：")
    print(result["output"])

3. 记忆系统

# examples/stage6/langchain_memory.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chains import ConversationChain

# 1. 缓冲记忆（保存所有历史）
def demo_buffer_memory():
    llm = ChatOpenAI(temperature=0.7)

    memory = ConversationBufferMemory()
    conversation = ConversationChain(
        llm=llm,
        memory=memory,
        verbose=True
    )

    # 对话
    print(conversation.predict(input="你好，我叫张三"))
    print(conversation.predict(input="我喜欢编程"))
    print(conversation.predict(input="我叫什么名字？"))

    # 查看记忆
    print("\n记忆内容：")
    print(memory.load_memory_variables({}))

# 2. 摘要记忆（自动总结历史）
def demo_summary_memory():
    llm = ChatOpenAI(temperature=0)

    memory = ConversationSummaryMemory(llm=llm)
    conversation = ConversationChain(
        llm=llm,
        memory=memory,
        verbose=True
    )

    # 长对话
    conversation.predict(input="介绍一下 Python")
    conversation.predict(input="它有什么优点？")
    conversation.predict(input="适合做什么项目？")

    # 查看摘要
    print("\n对话摘要：")
    print(memory.load_memory_variables({}))

if __name__ == "__main__":
    print("=" * 60)
    print("缓冲记忆示例")
    print("=" * 60)
    demo_buffer_memory()

    print("\n" + "=" * 60)
    print("摘要记忆示例")
    print("=" * 60)
    demo_summary_memory()

4. RAG 实现

# examples/stage6/langchain_rag.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader

# 1. 加载文档
def load_documents():
    """加载文档（示例使用文本）"""
    documents = [
        "Python 是一种高级编程语言，由 Guido van Rossum 创建于 1991 年。",
        "Python 的设计哲学强调代码的可读性和简洁的语法。",
        "Python 拥有丰富的标准库和第三方包生态系统。",
        "Django 是一个用 Python 编写的高级 Web 框架。",
        "Flask 是一个轻量级的 Python Web 框架。",
    ]

    from langchain.schema import Document
    return [Document(page_content=doc) for doc in documents]

# 2. 构建 RAG 系统
def build_rag_system():
    # 加载文档
    docs = load_documents()

    # 分割文档
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=100,
        chunk_overlap=20
    )
    splits = text_splitter.split_documents(docs)

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

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

    # 创建 QA 链
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )

    return qa_chain

# 使用示例
if __name__ == "__main__":
    qa_system = build_rag_system()

    questions = [
        "Python 是什么时候创建的？",
        "有哪些 Python Web 框架？",
        "Python 的设计哲学是什么？"
    ]

    for question in questions:
        print(f"\n问题: {question}")
        print("-" * 60)

        result = qa_system.invoke({"query": question})

        print(f"答案: {result['result']}")
        print(f"\n来源文档:")
        for i, doc in enumerate(result['source_documents'], 1):
            print(f"{i}. {doc.page_content}")
        print()

📊 LlamaIndex 实践

1. 基础索引

# examples/stage6/llamaindex_basics.py
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 配置
Settings.llm = OpenAI(model="gpt-4", temperature=0.1)
Settings.embed_model = OpenAIEmbedding()

# 1. 加载文档
def create_index_from_documents():
    """从文档创建索引"""
    # 方式1：从目录加载
    # documents = SimpleDirectoryReader('data').load_data()

    # 方式2：手动创建文档
    from llama_index.core import Document
    documents = [
        Document(text="Python 是一种高级编程语言，创建于 1991 年。"),
        Document(text="LlamaIndex 是一个用于构建 LLM 应用的数据框架。"),
        Document(text="向量数据库用于存储和检索文本嵌入。")
    ]

    # 创建索引
    index = VectorStoreIndex.from_documents(documents)

    return index

# 2. 查询索引
def query_index(index, question):
    """查询索引"""
    query_engine = index.as_query_engine()
    response = query_engine.query(question)

    return response

# 使用示例
if __name__ == "__main__":
    index = create_index_from_documents()

    questions = [
        "Python 是什么时候创建的？",
        "什么是 LlamaIndex？"
    ]

    for q in questions:
        print(f"问题: {q}")
        response = query_index(index, q)
        print(f"答案: {response}\n")

2. 高级 RAG

# examples/stage6/llamaindex_advanced.py
from llama_index.core import VectorStoreIndex, Document
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine

def create_advanced_rag():
    """创建高级 RAG 系统"""

    # 准备文档
    documents = [
        Document(text="""
        AI Agent 是一种能够感知环境、做出决策并采取行动的智能系统。
        它结合了大语言模型的推理能力和工具调用的执行能力。
        Agent 可以自主规划任务步骤，调用各种工具完成复杂任务。
        """),
        Document(text="""
        RAG（检索增强生成）是一种结合信息检索和文本生成的技术。
        它首先检索相关文档，然后基于这些文档生成答案。
        这种方法可以提高答案的准确性和可信度。
        """)
    ]

    # 自定义分块
    splitter = SentenceSplitter(chunk_size=128, chunk_overlap=20)

    # 创建索引
    index = VectorStoreIndex.from_documents(
        documents,
        transformations=[splitter]
    )

    # 配置检索器
    retriever = VectorIndexRetriever(
        index=index,
        similarity_top_k=3  # 返回最相似的3个块
    )

    # 创建查询引擎
    query_engine = RetrieverQueryEngine(retriever=retriever)

    return query_engine

# 使用示例
if __name__ == "__main__":
    engine = create_advanced_rag()

    response = engine.query("什么是 AI Agent？")
    print(f"答案: {response}")

🎯 实践项目

综合 Agent 应用

# examples/stage6/comprehensive_agent.py
from langchain_openai import ChatOpenAI
from langchain.agents import Tool, AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.chains import RetrievalQA

class ComprehensiveAgent:
    """综合能力的 Agent"""

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        self.tools = self._create_tools()
        self.agent_executor = self._create_agent()

    def _create_knowledge_base(self):
        """创建知识库"""
        documents = [
            "公司成立于2020年，主营AI产品开发。",
            "我们的产品包括智能客服、文档分析、代码助手。",
            "技术栈：Python, FastAPI, React, PostgreSQL。",
        ]

        from langchain.schema import Document
        docs = [Document(page_content=d) for d in documents]

        embeddings = OpenAIEmbeddings()
        vectorstore = FAISS.from_documents(docs, embeddings)

        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=vectorstore.as_retriever()
        )

        return qa_chain

    def _create_tools(self):
        """创建工具"""
        kb_chain = self._create_knowledge_base()

        tools = [
            Tool(
                name="knowledge_base",
                func=lambda q: kb_chain.invoke({"query": q})["result"],
                description="查询公司知识库，包括公司信息、产品、技术栈等。"
            ),
            Tool(
                name="calculate",
                func=lambda expr: str(eval(expr, {"__builtins__": {}}, {})),
                description="计算数学表达式。输入应该是有效的数学表达式。"
            )
        ]

        return tools

    def _create_agent(self):
        """创建 Agent"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个智能助手，可以使用工具回答问题并记住对话历史。"),
            MessagesPlaceholder(variable_name="chat_history"),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        agent = create_openai_functions_agent(
            self.llm,
            self.tools,
            prompt
        )

        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True
        )

    def chat(self, message: str):
        """对话"""
        result = self.agent_executor.invoke({"input": message})
        return result["output"]

# 使用示例
if __name__ == "__main__":
    agent = ComprehensiveAgent()

    print("Agent: 你好！我可以帮你查询公司信息、进行计算等。有什么可以帮助你的？\n")

    # 多轮对话
    conversations = [
        "公司是什么时候成立的？",
        "我们的产品有哪些？",
        "帮我算一下 (100 + 50) * 2",
        "我们之前讨论了什么？"  # 测试记忆
    ]

    for msg in conversations:
        print(f"用户: {msg}")
        response = agent.chat(msg)
        print(f"Agent: {response}\n")

📚 框架选择建议

选择 LangChain 当你需要：

✅ 丰富的工具和集成
✅ 灵活的链式组合
✅ 复杂的 Agent 逻辑
✅ 活跃的社区支持

选择 LlamaIndex 当你需要：

✅ 专注于数据索引和检索
✅ 简单的 RAG 应用
✅ 结构化数据处理
✅ 快速原型开发

选择 AutoGen/CrewAI 当你需要：

✅ 多 Agent 协作
✅ 角色化系统
✅ 复杂对话流程

✅ 阶段完成检查清单

[ ] 理解主流框架的特点和适用场景
[ ] 掌握 LangChain 的核心概念（Chain、Tool、Memory）
[ ] 能够使用 LangChain 创建 Agent
[ ] 了解 LlamaIndex 的索引和查询机制
[ ] 实现了基于框架的 RAG 系统
[ ] 完成综合 Agent 应用项目
[ ] 能够根据需求选择合适的框架

🎯 下一步

完成本阶段后，进入阶段 7：工程化、评估与安全，学习如何将 Agent 产品化。

💡 小贴士：框架是工具，理解底层原理更重要。先用框架快速实现，再根据需要优化和定制。不要被框架限制思路！