纲要

• 项目概览：基于 LangChain 的单智能体“小浪助手”
• 核心能力拆解
  • RAG 知识库问答
  • 钉钉日程与待办管理
  • 多轮对话与上下文记忆
  • 情绪检测与动态优先级调整
• 系统架构
  • 前端：钉钉机器人
  • 调度层：LangChain Agent
  • 工具集：知识库检索、在线搜索、钉钉 API
  • 记忆系统：基于 Redis 的长短期记忆
  • 模型：大语言模型（LLM）生成与意图识别
• 单智能体概念与 LangChain 实现要点
• 实战代码详解
  • 项目结构
  • 工具定义（Tool 封装）
  • 记忆组件（RedisChatMessageHistory）
  • Agent 构建与执行流程
  • 情绪检测链
• 完整可运行示例
• 总结与相关度说明

项目概览

我们要构建的是一个名为“小浪助手”的单智能体，它通过钉钉机器人对外提供服务，核心功能包括：

• 基于 RAG 的专业知识问答
• 自然语言驱动的日程安排与待办管理
• 多轮对话能力，能记住上下文
• 实时检测用户情绪，根据负面程度自动创建高优先级待办

所有逻辑都使用 LangChain 编排，工具调用由 LLM 自主决策，记忆系统持久化在 Redis 中。

核心概念与架构

什么是单智能体

在 AI 应用语境中，智能体是一个能够感知环境、自主使用工具并做出决策的程序。单智能体指整个系统中只有一个决策中心，它根据用户输入决定调用哪些工具、如何回应，而多智能体则涉及多个协作或竞争的决策单元。本案例集中展示单智能体的设计、工具集成与记忆管理。

系统架构

架构分为输入、智能体决策、工具执行和记忆四个层次。用户消息进入后，Agent 先分析情绪和意图，再决定调用哪些工具，工具返回结果后结合记忆与 LLM 生成回答。

技术选型与关键模块

模块	技术/工具
智能体框架	LangChain
大模型	OpenAI / 兼容接口的 LLM
知识库	文档加载 → 文本分割 → 向量嵌入 → 向量数据库（Chroma / FAISS）
记忆	Redis (RedisChatMessageHistory)
在线搜索	搜索引擎 API 封装为 Tool
钉钉 API	钉钉开放平台 SDK，封装日程、待办调用
情绪检测	基于 LLM 的链，输出情绪标签和分值

实战代码

项目结构

langchain-dingtalk-agent/
├── agent.py              # 主程序，Agent 定义与启动
├── tools/
│   ├── dingtalk.py       # 钉钉日程、待办工具
│   ├── knowledge.py      # RAG 知识库检索
│   └── search.py         # 在线搜索工具
├── chains/
│   └── emotion.py        # 情绪检测链
├── memory/
│   └── redis_memory.py   # Redis 记忆配置
├── config.py             # 环境变量与密钥
└── requirements.txt

依赖安装

pip install langchain langchain-openai langchain-community redis chromadb tiktoken python-dotenv

配置文件 config.py

import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
DINGTALK_APP_KEY = os.getenv("DINGTALK_APP_KEY")
DINGTALK_APP_SECRET = os.getenv("DINGTALK_APP_SECRET")

记忆组件 memory/redis_memory.py

from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain.memory import ConversationBufferMemory
from config import REDIS_URL

def get_memory(session_id: str) -> ConversationBufferMemory:
    """为每个会话创建独立的 Redis 记忆"""
    history = RedisChatMessageHistory(
        session_id=session_id,
        url=REDIS_URL,
    )
    return ConversationBufferMemory(
        memory_key="chat_history",
        chat_memory=history,
        return_messages=True,
    )

工具定义

钉钉工具 tools/dingtalk.py

from langchain.tools import tool
import requests
from config import DINGTALK_APP_KEY, DINGTALK_APP_SECRET

def get_access_token():
    # 钉钉获取 access_token（示例实现）
    resp = requests.get(
        "https://oapi.dingtalk.com/gettoken",
        params={"appkey": DINGTALK_APP_KEY, "appsecret": DINGTALK_APP_SECRET},
    )
    return resp.json().get("access_token")

@tool
def create_calendar_event(summary: str, start_time: str, end_time: str) -> str:
    """在钉钉日历中创建事件。参数 summary: 事件标题, start_time: ISO 格式开始时间, end_time: ISO 格式结束时间。"""
    token = get_access_token()
    # 调用钉钉日历创建 API（示例使用预留接口）
    # 实际需使用钉钉日程接口 https://open.dingtalk.com/document/orgapp/create-calendar-event
    return f"已成功创建日程：{summary}，时间 {start_time} 至 {end_time}"

@tool
def create_task(content: str, priority: str = "normal") -> str:
    """创建钉钉待办。参数 content: 待办内容, priority: 优先级 low/medium/high。"""
    token = get_access_token()
    # 调用钉钉待办创建 API
    return f"已创建待办：{content}，优先级 {priority}"

知识库工具 tools/knowledge.py

from langchain.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from config import OPENAI_API_KEY

# 假设向量数据库已预先构建，存于 ./chroma_db
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(openai_api_key=OPENAI_API_KEY)
)

@tool
def search_knowledge_base(query: str) -> str:
    """搜索专业知识库，传入用户问题，返回相关文档片段。"""
    docs = vectorstore.similarity_search(query, k=3)
    if not docs:
        return "知识库中未找到相关信息。"
    return "\n\n".join([d.page_content for d in docs])

在线搜索工具 tools/search.py

from langchain.tools import tool
from langchain_community.utilities import SerpAPIWrapper
import os

@tool
def web_search(query: str) -> str:
    """在线搜索最新信息，返回搜索摘要。"""
    search = SerpAPIWrapper(serpapi_api_key=os.getenv("SERPAPI_API_KEY"))
    return search.run(query)

情绪检测链 chains/emotion.py

from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
import json

emotion_prompt = PromptTemplate(
    input_variables=["user_input"],
    template="""分析以下用户消息的情绪，返回 JSON 格式 {{"emotion": "positive/neutral/negative", "score": 0-10}}。
消息：{user_input}
JSON:"""
)

def detect_emotion(user_input: str) -> dict:
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    chain = emotion_prompt | llm
    result = chain.invoke({"user_input": user_input})
    try:
        return json.loads(result.content)
    except:
        return {"emotion": "neutral", "score": 5}

Agent 主程序 agent.py

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from tools.dingtalk import create_calendar_event, create_task
from tools.knowledge import search_knowledge_base
from tools.search import web_search
from chains.emotion import detect_emotion
from memory.redis_memory import get_memory
from config import OPENAI_API_KEY

# 1. 工具列表
tools = [create_calendar_event, create_task, search_knowledge_base, web_search]

# 2. LLM 实例
llm = ChatOpenAI(model="gpt-4", temperature=0, openai_api_key=OPENAI_API_KEY)

# 3. 系统提示词
system_prompt = """你是小浪助手，一个钉钉智能客服。你可以：
- 回答关于产品的专业知识（调用知识库）
- 搜索最新信息（调用在线搜索）
- 管理用户的钉钉日程和待办
请友好、专业地回应用户。如果检测到用户情绪非常负面（score >= 8），主动将用户当前诉求创建为高优先级待办。"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# 4. 构建 Agent
agent = create_openai_tools_agent(llm, tools, prompt)

# 5. 模拟执行函数（实际可接入钉钉 Webhook）
def process_message(user_input: str, session_id: str = "default") -> str:
    # 情绪检测
    emotion_result = detect_emotion(user_input)
    # 如果负面情绪高，自动追加创建待办的意图
    if emotion_result["emotion"] == "negative" and emotion_result["score"] >= 8:
        # 将需求描述直接传给 create_task 工具（让 Agent 自己决策调用）
        # 我们可以在提示里做引导，这里简单修改输入
        user_input += "\n（系统检测到强烈负面情绪，请主动将相关诉求创建为高优先级待办）"

    memory = get_memory(session_id)
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        memory=memory,
        verbose=True,
        handle_parsing_errors=True,
    )
    response = agent_executor.invoke({"input": user_input})
    return response["output"]

# 示例运行
if __name__ == "__main__":
    # 模拟多轮对话
    session = "user_123"
    print("小浪助手：", process_message("帮我预定后天上午10点的会议室，主题是智能体技术讨论", session))
    print("小浪助手：", process_message("我想起来后天早上有事，相关的安排调整一下吧", session))
    print("小浪助手：", process_message("算了，我特别生气，你们产品太难用了！", session))

运行说明

1. 准备环境变量文件 .env，填入 OPENAI_API_KEY、REDIS_URL、SERPAPI_API_KEY 及钉钉凭证。
2. 确保本地 Redis 服务已启动，并事先构建好知识库（向量数据库存于 ./chroma_db）。
3. 运行 python agent.py 即可看到控制台输出。实际接入钉钉时，将 process_message 作为 Webhook 处理函数即可。

多轮对话的实现

在 LangChain 中，通过在 AgentExecutor 里注入 memory 即可实现上下文记录。这里使用 ConversationBufferMemory 配合 RedisChatMessageHistory，使会话历史在服务重启后依然保留。每次调用时传入相同的 session_id，Agent 就能读取之前的对话，从而理解“相关的安排”这类指代。

情绪检测与自动待办

情绪检测链利用 LLM 输出结构化 JSON，解析出情感倾向和分值。当负面分值达到阈值时，主流程在原输入尾部追加提示，引导 Agent 调用 create_task 工具。这种方式保持了 Agent 的自主决策能力，同时又可植入业务规则。

总结

本博客基于“小浪助手”单智能体实战内容，完整复现了一个使用 LangChain 构建的钉钉智能助手，涵盖了架构设计、工具开发、记忆集成、情绪检测和可运行代码。

读者可依据此模板快速扩展到多智能体或更多业务工具。