本地 Agent 系统搭建实录：LangChain + Ollama + 向量记忆

技术选型

一、环境搭建

1. 安装 Ollama 并拉取模型

# Windows 直接去 ollama.com 下载安装包
# 安装完后拉模型（选一个）
ollama pull qwen2.5:14b    # 中文好，16G内存可跑
ollama pull deepseek-r1:14b  # 推理强，同理
ollama pull qwen2.5:7b      # 机器差选这个

# 验证：跑起来看看
ollama list
ollama run qwen2.5:14b "你好"

2. 安装 Python 依赖

pip install langchain langchain-community langchain-core
pip install langchain-ollama       # Ollama 接入 LangChain
pip install chromadb               # 向量数据库（本地持久化）
pip install selenium openpyxl     # 工具：浏览器 + Excel
pip install docker                # 沙箱执行
pip install faiss-cpu             # 备选向量库（比 Chroma 快）

二、长期记忆模块

核心思路：每次 Agent 完成任务后，把"任务描述 + 执行步骤 + 结果"存进向量库。下次碰到类似任务，先检索历史经验，拼进 Prompt，Agent 就不用重新规划了。

实现代码

from langchain_ollama import ChatOllama
from langchain_community.vectorstores import Chroma
from langchain_ollama import OllamaEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
import json, os

# 1. 初始化嵌入模型（用 Ollama 的 nomic-embed-text 做向量化）
embeddings = OllamaEmbeddings(model="nomic-embed-text")
# 先拉取嵌入模型：ollama pull nomic-embed-text

# 2. 初始化 Chroma 向量库（持久化到磁盘）
persist_dir = "./agent_memory"
vectorstore = Chroma(
    collection_name="agent_experience",
    embedding_function=embeddings,
    persist_directory=persist_dir
)

# 3. 存储经验
def save_experience(task_desc: str, steps: list, result: str):
    """把一次任务执行记录存进向量库"""
    doc = f"""任务：{task_desc}
执行步骤：{json.dumps(steps, ensure_ascii=False)}
结果：{result}"""
    vectorstore.add_texts(
        texts=[doc],
        metadatas=[{"task": task_desc, "type": "experience"}]
    )

# 4. 检索相似经验
def recall_experience(task_desc: str, k=3):
    """根据任务描述，召回最相关的历史经验"""
    results = vectorstore.similarity_search(task_desc, k=k)
    return "\n---\n".join([r.page_content for r in results])

# 5. 拼进 Prompt 使用
prompt = ChatPromptTemplate.from_template("""你是一个智能助手。
以下是你之前处理过的类似任务的经验，请参考后执行当前任务：

{experience}

当前任务：{task}

请输出执行步骤：""")

chain = (
    {"experience": lambda x: recall_experience(x["task"]), "task": RunnablePassthrough()}
    | prompt
    | ChatOllama(model="qwen2.5:14b", temperature=0.1)
)

# 使用示例
result = chain.invoke({"task": "帮我从某网站抓取表格数据并存入Excel"})
print(result.content)

# 执行完后存经验
save_experience("从网站抓取表格数据存Excel", ["步骤1...", "步骤2..."], result.content)

三、多 Agent 协作

不用上 RabbitMQ 那么重——LangChain 的 AgentExecutor 和 create_openai_functions 风格的 Agent 已经支持多 Agent 编排。轻量方案是直接用 LangChain 的 Tool 把子 Agent 包装成工具。

架构设计

主 Agent（协调者）
  ├── 工具：爬虫 Agent（负责拿数据）
  ├── 工具：分析 Agent（负责处理数据）
  └── 工具：报告 Agent（负责输出结果）

实现代码

from langchain_ollama import ChatOllama
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import Tool
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOllama(model="qwen2.5:14b", temperature=0.1)

# ===== 子 Agent 1：爬虫 =====
def crawl_tool(query: str) -> str:
    """用 Selenium 爬取数据（简化版）"""
    from selenium import webdriver
    from selenium.webdriver.common.by import By
    import time
    # 实际项目里这里封装你的 MSD 爬虫逻辑
    return f"[爬虫结果] 查询「{query}」得到数据XXX条"

crawl_agent = Tool(
    name="CrawlAgent",
    func=crawl_tool,
    description="当需要获取网页数据时使用。输入：查询关键词或URL"
)

# ===== 子 Agent 2：分析 =====
def analyze_tool(data: str) -> str:
    """调用 LLM 做数据分析"""
    prompt = f"请分析以下数据，给出关键结论：\n{data}"
    result = llm.invoke(prompt)
    return result.content

analyze_agent = Tool(
    name="AnalyzeAgent",
    func=analyze_tool,
    description="当需要分析数据、提取结论时使用。输入：原始数据文本"
)

# ===== 子 Agent 3：写报告 =====
def report_tool(conclusion: str) -> str:
    """生成结构化报告"""
    prompt = f"请将以下结论整理成结构化报告（含摘要、数据、建议）：\n{conclusion}"
    result = llm.invoke(prompt)
    return result.content

report_agent = Tool(
    name="ReportAgent",
    func=report_tool,
    description="当需要生成正式报告时使用。输入：分析结论"
)

# ===== 主 Agent 编排 =====
tools = [crawl_agent, analyze_agent, report_agent]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个任务协调助手。根据用户的请求，决定调用哪个子Agent（CrawlAgent/AnalyzeAgent/ReportAgent）。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行
result = executor.invoke({"input": "帮我查最近抖音电商数据趋势，并给我一份报告"})
print(result["output"])

# 对 Ollama 更友好的多 Agent 方案：用 ReAct
from langchain.agents import create_react_agent
from langchain import hub

# 拉取 ReAct 提示词模板
prompt = hub.pull("hwchase17/react")  # 需要 LANGCHAIN_API_KEY，或者手写

# 手写简化版 ReAct prompt（推荐，不依赖 LangSmith）
react_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个助手，可以使用以下工具：
{tools}
使用格式：
Thought: 思考过程
Action: 工具名
Action Input: 工具输入
Observation: 工具返回结果
...（重复直到完成任务）
Thought: 我知道答案了
Final Answer: 最终答案"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_react_agent(llm, tools, react_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)
# handle_parsing_errors=True 很重要！本地模型输出格式不稳定时自动重试

四、沙箱环境

Agent 生成代码后直接执行是很危险的。用 Docker 创建一个隔离容器，代码只在容器内跑，跑完即删。

实现代码

import docker
import tempfile
import os

client = docker.from_env()

def run_code_in_sandbox(code: str, timeout: int = 30) -> str:
    """
    在 Docker 容器内执行 Python 代码，返回输出。
    容器用完即删。
    """
    # 把代码写进临时文件
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        tmp_path = f.name

    try:
        # 启动容器，挂载代码文件，执行后自动删除
        result = client.containers.run(
            image="python:3.11-slim",
            command=f"python /code/{os.path.basename(tmp_path)}",
            volumes={tmp_path: {"bind": f"/code/{os.path.basename(tmp_path)}", "mode": "ro"}},
            remove=True,          # 跑完自动删容器
            mem_limit="256m",    # 限制内存
            cpu_period=100000,
            cpu_quota=50000,     # 限制 CPU 50%
            network_disabled=True,# 禁网，防止恶意请求
            stdout=True,
            stderr=True,
            timeout=timeout
        )
        return result.decode("utf-8", errors="ignore")
    except docker.errors.ContainerError as e:
        return f"执行出错：{e.stderr.decode() if e.stderr else str(e)}"
    except Exception as e:
        return f"沙箱错误：{str(e)}"
    finally:
        os.unlink(tmp_path)

# 封装成 LangChain Tool
def sandbox_tool(code: str) -> str:
    return run_code_in_sandbox(code)

sandbox_agent = Tool(
    name="CodeSandbox",
    func=sandbox_tool,
    description="当需要执行 Python 代码时使用。输入：完整Python代码字符串。代码在沙箱内执行，安全隔离。"
)

五、性能优化

1. 用 Function Calling 代替 Prompt 解析

让本地模型输出工具调用格式，比让它"自由输出然后正则解析"稳定10倍：

# 好的做法：用 LangChain 的 tool 装饰器，让模型自己决定调用
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索网络。"""
    # 实现搜索逻辑
    return f"搜索结果：{query}"

@tool
def write_excel(data: str, filename: str) -> str:
    """把数据写入Excel文件。"""
    import openpyxl
    wb = openpyxl.Workbook()
    ws = wb.active
    ws['A1'] = data
    wb.save(filename)
    return f"已写入 {filename}"

tools = [search_web, write_excel]
# LangChain 会自动把 @tool 装饰的函数转成 Function Calling 格式发给模型

2. 加缓存：重复任务不重跑

from langchain_core.caches import InMemoryCache
from langchain_ollama import ChatOllama
from langchain.cache import SQLiteCache  # 持久化缓存

# 方案A：内存缓存（重启失效）
ChatOllama(model="qwen2.5:14b").cache = InMemoryCache()

# 方案B：SQLite 持久化缓存（推荐）
from langchain_community.cache import SQLiteCache
import langchain
langchain.llm_cache = SQLiteCache(database_path="./llm_cache.db")
# 相同 Prompt 第二次会直接返回缓存结果，省掉一次模型调用

3. 向量检索加 Rerank

Chroma 召回的 top-K 结果不一定最相关，加一层 Rerank 可以提升准确率：

# 用 Ollama 做轻量 Rerank（不用额外API）
def rerank_results(query: str, docs: list, top_n: int = 2) -> list:
    """用 LLM 对召回结果重新打分排序"""
    rank_prompt = f"""请对以下文档与查询的相关性打分（1-10分，只输出分数）：
查询：{query}
"""
    for doc in docs:
        score_result = llm.invoke(rank_prompt + doc.page_content[:500])
        # 解析分数（简化版）
        try:
            score = int(''.join(filter(str.isdigit, score_result.content)))
        except:
            score = 5
        doc.metadata["rerank_score"] = score

    docs.sort(key=lambda x: x.metadata.get("rerank_score", 0), reverse=True)
    return docs[:top_n]

六、完整集成示例

把上面所有模块串起来，一个能"记住历史经验"的本地 Agent：

from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
import docker, os, json

llm = ChatOllama(model="qwen2.5:14b", temperature=0.1)
embeddings = OllamaEmbeddings(model="nomic-embed-text")
memory = Chroma(collection_name="agent_mem", embedding_function=embeddings, persist_directory="./agent_memory")

# 工具定义
@tool
def remember(task: str) -> str:
    """检索历史经验。输入：任务描述。"""
    results = memory.similarity_search(task, k=2)
    return "\n".join([r.page_content for r in results]) if results else "无相关历史经验"

@tool
def sandbox_run(code: str) -> str:
    """在沙箱内执行Python代码。输入：完整代码。"""
    client = docker.from_env()
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code); tmp = f.name
    try:
        r = client.containers.run("python:3.11-slim",
            command=f"python /code/{os.path.basename(tmp)}",
            volumes={tmp: {"bind": f"/code/{os.path.basename(tmp)}", "mode": "ro"}},
            remove=True, mem_limit="256m", network_disabled=True,
            stdout=True, stderr=True, timeout=30)
        return r.decode()
    finally:
        os.unlink(tmp)

@tool
def save_task(task_desc: str, steps: str) -> str:
    """保存本次任务经验到记忆库。输入：任务描述+执行步骤（JSON格式）。"""
    memory.add_texts([f"任务：{task_desc}\n步骤：{steps}"])
    return "经验已保存"

tools = [remember, sandbox_run, save_task]

react_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有长期记忆的Agent。可用工具：{tools}\n{tool_names}"),
    ("human", "{input}\n{agent_scratchpad}"),
])

agent = create_react_agent(llm, tools, react_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

# 使用
result = executor.invoke({"input": "帮我写一个Selenium脚本，自动登录某网站并下载报表"})
print(result["output"])

遇到的问题汇总