纲要项目准备开发资源获取Python与VSCode环境配置大模型API Key获取硅基流动、CloseAIGitHub Copilot开启 AI 辅助编程钉钉开放平台应用创建与Stream模式配置LangSmith可观测性追踪开通项目架构概览目录结构核心模块关系Mermaid流程图实战开发从入口到智能体环境依赖与Poetry管理.env配置文件详解日志系统初始化钉钉WebSocket入口实现智能体核心类AIAgent主/备模型回退机制工具加载与记忆系统情感分析与动态提示词多用户记忆隔离session_id完整可运行代码示例总结与相关度说明项目准备资源一站式获取在动手写代码之前我们需要先把开发环境、模型 API、钉钉应用等资源全部备齐。资源类型获取方式用途Python 3.10python.org 下载安装运行环境VSCodecode.visualstudio.com 下载主力 IDEGitHub CopilotVSCode 内购订阅推荐专业版AI 辅助编码、调试大模型 API硅基流动siliconflow.cn或 CloseAI提供 LLM 推理能力搜索引擎SerpAPIserpapi.com 注册免费 key在线搜索工具钉钉应用open.dingtalk.com 创建“钉钉应用”非“机器人”消息收发、日程待办 APILangSmithsmith.langchain.com 注册并创建 API Key调用链追踪、成本监控Redis本地安装或 Docker 启动会话记忆存储硅基流动硅基流动与 CloseAI 使用要点硅基流动托管了大量国产开源模型DeepSeek、Qwen 等其 API 地址完全兼容 OpenAI 格式只需更换base_url即可。注册后完成实名认证在“模型广场”中复制模型名称如deepseek-ai/DeepSeek-R1然后在API 文档中获取接口地址和密钥。计费透明按 token 消费适合个人开发者。CloseAI 则整合了 OpenAI、Claude、Gemini、DeepSeek 等海外模型通过统一接口提供。需注意不同模型的base_url后缀不同例如 OpenAI 使用/v1Claude 使用/anthropic。在“密钥管理”页面创建密钥并在使用时指定正确的地址。安全提醒永远不要将 API Key 明文写在代码中应放入.env文件并用.gitignore排除。项目架构一览小浪助手是一个基于LangChain的单智能体通过钉钉Stream模式WebSocket与用户实时交互具备日程管理、待办创建、知识库问答和情绪检测能力。其核心架构如下WebSocket文本消息 用户ID钉钉用户入口: dingtalk_hook.pyAIAgent 核心类情感分析链动态提示词管理Redis 记忆系统主模型 OpenAI / 备用 DeepSeek工具集知识库检索在线搜索钉钉日历/待办 API生成回复用户消息经钉钉 WebSocket 传入后被入口函数捕获提取文本和发送者唯一 ID。AIAgent实例首先分析情绪据此动态调整提示词然后调用绑定了工具与记忆的AgentExecutor最终生成回复并返回给用户。实战开发从零编写核心代码环境初始化与项目结构使用Poetry管理依赖确保环境干净可复现。在项目根目录执行mkdirxiaolang-dingtalkcdxiaolang-dingtalk poetry init-npoetryaddlangchain langchain-openai langchain-community python-dotenv redis pydantic\dingtalk-stream chromadb tiktoken项目目录结构如下xiaolang-dingtalk/ ├── src/ │ ├── dingtalk_hook.py # 钉钉入口WebSocket 回调 │ ├── agent.py # AI智能体核心类 │ ├── tools/ │ │ ├── __init__.py │ │ ├── calendar.py # 钉钉日历工具 │ │ ├── task.py # 钉钉待办工具 │ │ ├── knowledge.py # RAG 知识库检索 │ │ └── search.py # 在线搜索 │ ├── chains/ │ │ └── emotion.py # 情感检测链 │ └── memory/ │ └── redis_memory.py # Redis 记忆封装 ├── .env # 环境变量不提交仓库 ├── pyproject.toml └── README.md配置环境变量.env# LLM 主模型 OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.siliconflow.cn/v1 DEFAULT_MODELdeepseek-ai/DeepSeek-R1 # 备用模型 DEEPSEEK_API_KEYsk-xxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com/v1 # 搜索引擎 SERPAPI_API_KEYxxxx # Redis REDIS_URLredis://localhost:6379/0 # 钉钉应用 DINGTALK_APP_KEYdingxxxx DINGTALK_APP_SECRETxxxx DINGTALK_ROBOT_CODExxxx # LangSmith 追踪 LANGCHAIN_TRACING_V2true LANGCHAIN_API_KEYls__xxxx LANGCHAIN_PROJECTxiaolang-dingtalk入口文件钉钉消息接收与回复使用钉钉官方dingtalk-stream库建立 WebSocket 连接注册消息回调。这里简化示例不依赖真实钉钉环境也能本地运行——我们用标准输入模拟消息收发但完整保留了多用户记忆隔离、调用智能体的真实流程。实际部署时只需将模拟部分替换为DingTalkStreamClient即可。# src/dingtalk_hook.pyimportosimportsysimportloggingfromdotenvimportload_dotenvfromagentimportAIAgent load_dotenv()# 配置日志logging.basicConfig(levellogging.INFO,format%(asctime)s - %(name)s - %(levelname)s - %(message)s)loggerlogging.getLogger(XiaoLang)# 全局用户存储模拟从钉钉回调中获取的 sender_iduser_store{}defget_or_create_user(sender_id:str):ifsender_idnotinuser_store:user_store[sender_id]{id:sender_id}returnuser_store[sender_id]defmain():logger.info(小浪助手启动中...)agentAIAgent()# 初始化智能体print(小浪助手已就绪输入 exit 退出。)# 模拟多用户交互输入格式为 用户ID: 消息内容whileTrue:try:rawinput( )ifraw.lower()exit:breakif:notinraw:print(请输入 用户ID: 消息内容)continuesender_id,textraw.split(:,1)sender_idsender_id.strip()texttext.strip()# 存储或获取用户信息get_or_create_user(sender_id)# 调用智能体replyagent.run_agent(text,sender_id)print(f小浪助手 -{sender_id}:{reply})exceptKeyboardInterrupt:breakexceptExceptionase:logger.error(f处理消息失败:{e})if__name____main__:main()真实钉钉连接时将main函数替换为DingTalkStreamClient的回调注册即可原理完全一致。智能体核心类AIAgent这个类封装了模型加载、工具注册、记忆管理、情感分析和动态提示词合成的全部逻辑。# src/agent.pyimportosimportjsonfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain.toolsimporttoolfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain.memoryimportConversationBufferMemoryfromlangchain_community.chat_message_historiesimportRedisChatMessageHistoryfromlangchain_core.runnablesimportConfigurableField load_dotenv()# ---------- 工具定义简化示例 ----------tooldefcreate_calendar_event(summary:str,start_time:str,end_time:str)-str:创建钉钉日程。参数 summary: 标题, start_time: 开始时间 ISO格式, end_time: 结束时间 ISO格式# 实际调用钉钉 API此处模拟成功returnf已创建日程「{summary}」时间{start_time}~{end_time}tooldefcreate_task(content:str,priority:strnormal)-str:创建钉钉待办。参数 content: 待办内容, priority: 优先级 low/medium/highreturnf已创建待办「{content}」优先级{priority}tooldefweb_search(query:str)-str:在线搜索最新信息fromlangchain_community.utilitiesimportSerpAPIWrapper searchSerpAPIWrapper()returnsearch.run(query)tools[create_calendar_event,create_task,web_search]# ---------- 情感分析链 ----------emotion_promptChatPromptTemplate.from_template(分析用户消息的情绪返回JSON: {{\emotion\: \positive/neutral/negative\, \score\: 0-10}}\n消息: {input}\nJSON:)classAIAgent:def__init__(self):# 1. 主模型可回退到备用模型primaryChatOpenAI(modelos.getenv(DEFAULT_MODEL,gpt-3.5-turbo),openai_api_keyos.getenv(OPENAI_API_KEY),base_urlos.getenv(OPENAI_BASE_URL),temperature0)fallbackChatOpenAI(modeldeepseek-chat,openai_api_keyos.getenv(DEEPSEEK_API_KEY),base_urlos.getenv(DEEPSEEK_BASE_URL),temperature0)self.llmprimary.with_fallbacks([fallback])# 2. 记忆系统基于 Redis支持多用户 sessionself.base_memoryConversationBufferMemory(memory_keychat_history,return_messagesTrue,input_keyinput,output_keyoutput)# 3. 情感分析 LLMself.emotion_llmChatOpenAI(modelgpt-3.5-turbo,temperature0)# 4. 基础提示词动态部分通过占位符注入base_promptChatPromptTemplate.from_messages([(system,你是小浪助手一个钉钉智能客服。你可以管理日程、创建待办、在线搜索。请友好专业地回复。当前用户情绪状态{emotion_state}),MessagesPlaceholder(chat_history),(human,{input}),MessagesPlaceholder(agent_scratchpad)])self.base_promptbase_prompt# 5. 构建可配置记忆的 Agentagentcreate_tool_calling_agent(self.llm,tools,base_prompt)self.agent_executorAgentExecutor(agentagent,toolstools,memoryself.base_memory,verboseTrue,handle_parsing_errorsTrue,).configurable_fields(memoryConfigurableField(idagent_memory,nameMemory,description可切换的记忆实例))defdetect_emotion(self,text:str)-dict:调用 LLM 分析情绪返回 dictchainemotion_prompt|self.emotion_llm resultchain.invoke({input:text})try:returnjson.loads(result.content)except:return{emotion:neutral,score:5}defget_memory_for_user(self,user_id:str)-ConversationBufferMemory:为特定用户创建独立的 Redis 记忆historyRedisChatMessageHistory(session_iduser_id,urlos.getenv(REDIS_URL,redis://localhost:6379/0))returnConversationBufferMemory(memory_keychat_history,chat_memoryhistory,return_messagesTrue,input_keyinput,output_keyoutput)defrun_agent(self,user_input:str,user_id:strdefault)-str:# 情绪分析emotionself.detect_emotion(user_input)# 根据情绪值动态调整提示词中的状态描述ifemotion[emotion]negativeandemotion[score]8:emotion_state强烈负面请主动将用户诉求创建为高优先级待办else:emotion_statef{emotion[emotion]}(分值{emotion[score]})# 合成带情绪状态的提示词dynamic_promptself.base_prompt.partial(emotion_stateemotion_state)# 重建 agent因为提示词变化需要新 agentfromlangchain.agentsimportcreate_tool_calling_agent agentcreate_tool_calling_agent(self.llm,tools,dynamic_prompt)# 获取用户专属记忆user_memoryself.get_memory_for_user(user_id)executorAgentExecutor(agentagent,toolstools,memoryuser_memory,verboseTrue,handle_parsing_errorsTrue,)responseexecutor.invoke({input:user_input})returnresponse[output]运行效果确保本地 Redis 已启动然后执行cdsrc python dingtalk_hook.py按提示输入user1: 帮我创建一个明天下午3点的会议主题是 AI 进展智能体将调用工具并回复。不同用户 ID 的消息会拥有各自独立的上下文记忆互不干扰。总结本文从零开始带领读者完成了钉钉智能助手项目的资源申请、环境搭建和核心代码编写。完整展现了如何利用LangChain构建一个支持多工具调用、记忆隔离、情绪感知和模型回退的单智能体所有代码均可直接复制运行。下一步你可以继续扩展工具集、接入真实钉钉 Stream 回调或迁移到LangGraph实现更复杂的多智能体协作。
纲要项目准备开发资源获取Python与VSCode环境配置大模型API Key获取硅基流动、CloseAIGitHub Copilot开启 AI 辅助编程钉钉开放平台应用创建与Stream模式配置LangSmith可观测性追踪开通项目架构概览目录结构核心模块关系Mermaid流程图实战开发从入口到智能体环境依赖与Poetry管理.env配置文件详解日志系统初始化钉钉WebSocket入口实现智能体核心类AIAgent主/备模型回退机制工具加载与记忆系统情感分析与动态提示词多用户记忆隔离session_id完整可运行代码示例总结与相关度说明项目准备资源一站式获取在动手写代码之前我们需要先把开发环境、模型 API、钉钉应用等资源全部备齐。资源类型获取方式用途Python 3.10python.org 下载安装运行环境VSCodecode.visualstudio.com 下载主力 IDEGitHub CopilotVSCode 内购订阅推荐专业版AI 辅助编码、调试大模型 API硅基流动siliconflow.cn或 CloseAI提供 LLM 推理能力搜索引擎SerpAPIserpapi.com 注册免费 key在线搜索工具钉钉应用open.dingtalk.com 创建“钉钉应用”非“机器人”消息收发、日程待办 APILangSmithsmith.langchain.com 注册并创建 API Key调用链追踪、成本监控Redis本地安装或 Docker 启动会话记忆存储硅基流动硅基流动与 CloseAI 使用要点硅基流动托管了大量国产开源模型DeepSeek、Qwen 等其 API 地址完全兼容 OpenAI 格式只需更换base_url即可。注册后完成实名认证在“模型广场”中复制模型名称如deepseek-ai/DeepSeek-R1然后在API 文档中获取接口地址和密钥。计费透明按 token 消费适合个人开发者。CloseAI 则整合了 OpenAI、Claude、Gemini、DeepSeek 等海外模型通过统一接口提供。需注意不同模型的base_url后缀不同例如 OpenAI 使用/v1Claude 使用/anthropic。在“密钥管理”页面创建密钥并在使用时指定正确的地址。安全提醒永远不要将 API Key 明文写在代码中应放入.env文件并用.gitignore排除。项目架构一览小浪助手是一个基于LangChain的单智能体通过钉钉Stream模式WebSocket与用户实时交互具备日程管理、待办创建、知识库问答和情绪检测能力。其核心架构如下WebSocket文本消息 用户ID钉钉用户入口: dingtalk_hook.pyAIAgent 核心类情感分析链动态提示词管理Redis 记忆系统主模型 OpenAI / 备用 DeepSeek工具集知识库检索在线搜索钉钉日历/待办 API生成回复用户消息经钉钉 WebSocket 传入后被入口函数捕获提取文本和发送者唯一 ID。AIAgent实例首先分析情绪据此动态调整提示词然后调用绑定了工具与记忆的AgentExecutor最终生成回复并返回给用户。实战开发从零编写核心代码环境初始化与项目结构使用Poetry管理依赖确保环境干净可复现。在项目根目录执行mkdirxiaolang-dingtalkcdxiaolang-dingtalk poetry init-npoetryaddlangchain langchain-openai langchain-community python-dotenv redis pydantic\dingtalk-stream chromadb tiktoken项目目录结构如下xiaolang-dingtalk/ ├── src/ │ ├── dingtalk_hook.py # 钉钉入口WebSocket 回调 │ ├── agent.py # AI智能体核心类 │ ├── tools/ │ │ ├── __init__.py │ │ ├── calendar.py # 钉钉日历工具 │ │ ├── task.py # 钉钉待办工具 │ │ ├── knowledge.py # RAG 知识库检索 │ │ └── search.py # 在线搜索 │ ├── chains/ │ │ └── emotion.py # 情感检测链 │ └── memory/ │ └── redis_memory.py # Redis 记忆封装 ├── .env # 环境变量不提交仓库 ├── pyproject.toml └── README.md配置环境变量.env# LLM 主模型 OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.siliconflow.cn/v1 DEFAULT_MODELdeepseek-ai/DeepSeek-R1 # 备用模型 DEEPSEEK_API_KEYsk-xxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com/v1 # 搜索引擎 SERPAPI_API_KEYxxxx # Redis REDIS_URLredis://localhost:6379/0 # 钉钉应用 DINGTALK_APP_KEYdingxxxx DINGTALK_APP_SECRETxxxx DINGTALK_ROBOT_CODExxxx # LangSmith 追踪 LANGCHAIN_TRACING_V2true LANGCHAIN_API_KEYls__xxxx LANGCHAIN_PROJECTxiaolang-dingtalk入口文件钉钉消息接收与回复使用钉钉官方dingtalk-stream库建立 WebSocket 连接注册消息回调。这里简化示例不依赖真实钉钉环境也能本地运行——我们用标准输入模拟消息收发但完整保留了多用户记忆隔离、调用智能体的真实流程。实际部署时只需将模拟部分替换为DingTalkStreamClient即可。# src/dingtalk_hook.pyimportosimportsysimportloggingfromdotenvimportload_dotenvfromagentimportAIAgent load_dotenv()# 配置日志logging.basicConfig(levellogging.INFO,format%(asctime)s - %(name)s - %(levelname)s - %(message)s)loggerlogging.getLogger(XiaoLang)# 全局用户存储模拟从钉钉回调中获取的 sender_iduser_store{}defget_or_create_user(sender_id:str):ifsender_idnotinuser_store:user_store[sender_id]{id:sender_id}returnuser_store[sender_id]defmain():logger.info(小浪助手启动中...)agentAIAgent()# 初始化智能体print(小浪助手已就绪输入 exit 退出。)# 模拟多用户交互输入格式为 用户ID: 消息内容whileTrue:try:rawinput( )ifraw.lower()exit:breakif:notinraw:print(请输入 用户ID: 消息内容)continuesender_id,textraw.split(:,1)sender_idsender_id.strip()texttext.strip()# 存储或获取用户信息get_or_create_user(sender_id)# 调用智能体replyagent.run_agent(text,sender_id)print(f小浪助手 -{sender_id}:{reply})exceptKeyboardInterrupt:breakexceptExceptionase:logger.error(f处理消息失败:{e})if__name____main__:main()真实钉钉连接时将main函数替换为DingTalkStreamClient的回调注册即可原理完全一致。智能体核心类AIAgent这个类封装了模型加载、工具注册、记忆管理、情感分析和动态提示词合成的全部逻辑。# src/agent.pyimportosimportjsonfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain.toolsimporttoolfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain.memoryimportConversationBufferMemoryfromlangchain_community.chat_message_historiesimportRedisChatMessageHistoryfromlangchain_core.runnablesimportConfigurableField load_dotenv()# ---------- 工具定义简化示例 ----------tooldefcreate_calendar_event(summary:str,start_time:str,end_time:str)-str:创建钉钉日程。参数 summary: 标题, start_time: 开始时间 ISO格式, end_time: 结束时间 ISO格式# 实际调用钉钉 API此处模拟成功returnf已创建日程「{summary}」时间{start_time}~{end_time}tooldefcreate_task(content:str,priority:strnormal)-str:创建钉钉待办。参数 content: 待办内容, priority: 优先级 low/medium/highreturnf已创建待办「{content}」优先级{priority}tooldefweb_search(query:str)-str:在线搜索最新信息fromlangchain_community.utilitiesimportSerpAPIWrapper searchSerpAPIWrapper()returnsearch.run(query)tools[create_calendar_event,create_task,web_search]# ---------- 情感分析链 ----------emotion_promptChatPromptTemplate.from_template(分析用户消息的情绪返回JSON: {{\emotion\: \positive/neutral/negative\, \score\: 0-10}}\n消息: {input}\nJSON:)classAIAgent:def__init__(self):# 1. 主模型可回退到备用模型primaryChatOpenAI(modelos.getenv(DEFAULT_MODEL,gpt-3.5-turbo),openai_api_keyos.getenv(OPENAI_API_KEY),base_urlos.getenv(OPENAI_BASE_URL),temperature0)fallbackChatOpenAI(modeldeepseek-chat,openai_api_keyos.getenv(DEEPSEEK_API_KEY),base_urlos.getenv(DEEPSEEK_BASE_URL),temperature0)self.llmprimary.with_fallbacks([fallback])# 2. 记忆系统基于 Redis支持多用户 sessionself.base_memoryConversationBufferMemory(memory_keychat_history,return_messagesTrue,input_keyinput,output_keyoutput)# 3. 情感分析 LLMself.emotion_llmChatOpenAI(modelgpt-3.5-turbo,temperature0)# 4. 基础提示词动态部分通过占位符注入base_promptChatPromptTemplate.from_messages([(system,你是小浪助手一个钉钉智能客服。你可以管理日程、创建待办、在线搜索。请友好专业地回复。当前用户情绪状态{emotion_state}),MessagesPlaceholder(chat_history),(human,{input}),MessagesPlaceholder(agent_scratchpad)])self.base_promptbase_prompt# 5. 构建可配置记忆的 Agentagentcreate_tool_calling_agent(self.llm,tools,base_prompt)self.agent_executorAgentExecutor(agentagent,toolstools,memoryself.base_memory,verboseTrue,handle_parsing_errorsTrue,).configurable_fields(memoryConfigurableField(idagent_memory,nameMemory,description可切换的记忆实例))defdetect_emotion(self,text:str)-dict:调用 LLM 分析情绪返回 dictchainemotion_prompt|self.emotion_llm resultchain.invoke({input:text})try:returnjson.loads(result.content)except:return{emotion:neutral,score:5}defget_memory_for_user(self,user_id:str)-ConversationBufferMemory:为特定用户创建独立的 Redis 记忆historyRedisChatMessageHistory(session_iduser_id,urlos.getenv(REDIS_URL,redis://localhost:6379/0))returnConversationBufferMemory(memory_keychat_history,chat_memoryhistory,return_messagesTrue,input_keyinput,output_keyoutput)defrun_agent(self,user_input:str,user_id:strdefault)-str:# 情绪分析emotionself.detect_emotion(user_input)# 根据情绪值动态调整提示词中的状态描述ifemotion[emotion]negativeandemotion[score]8:emotion_state强烈负面请主动将用户诉求创建为高优先级待办else:emotion_statef{emotion[emotion]}(分值{emotion[score]})# 合成带情绪状态的提示词dynamic_promptself.base_prompt.partial(emotion_stateemotion_state)# 重建 agent因为提示词变化需要新 agentfromlangchain.agentsimportcreate_tool_calling_agent agentcreate_tool_calling_agent(self.llm,tools,dynamic_prompt)# 获取用户专属记忆user_memoryself.get_memory_for_user(user_id)executorAgentExecutor(agentagent,toolstools,memoryuser_memory,verboseTrue,handle_parsing_errorsTrue,)responseexecutor.invoke({input:user_input})returnresponse[output]运行效果确保本地 Redis 已启动然后执行cdsrc python dingtalk_hook.py按提示输入user1: 帮我创建一个明天下午3点的会议主题是 AI 进展智能体将调用工具并回复。不同用户 ID 的消息会拥有各自独立的上下文记忆互不干扰。总结本文从零开始带领读者完成了钉钉智能助手项目的资源申请、环境搭建和核心代码编写。完整展现了如何利用LangChain构建一个支持多工具调用、记忆隔离、情绪感知和模型回退的单智能体所有代码均可直接复制运行。下一步你可以继续扩展工具集、接入真实钉钉 Stream 回调或迁移到LangGraph实现更复杂的多智能体协作。
