哈喽！各位小伙伴，咱们的LangChain入门实战系列又更新啦！前面七章我们一步步夯实基础，尤其是之前的一章sqlagent，已经成功用SQL Agent打通了自家MySQL服务，实现了自然语言查询数据库的核心功能，告别了繁琐的SQL手写，效率直接拉满。

但光在本地终端跑Agent还不够，日常办公我们大多离不开飞书，要是能把这个智能Agent搬进飞书，做成专属机器人，同事们在群里、私聊里发句话就能查数据、办业务，才是真正的AI办公落地。

所以本章咱们直奔主题：手把手把LangChain SQL Agent接入飞书机器人，打造专属的聊天式办公AI，零基础也能跟着抄作业！

先搞懂：飞书机器人到底能帮我们做什么？

很多刚接触飞书开发的小伙伴，可能只把机器人当成“自动发通知”的工具，其实它的能力远不止于此。对于咱们LangChain开发者来说，飞书机器人是AI Agent的最佳办公入口：

• 双向交互：不仅能主动推送消息，还能接收用户提问、解析指令，调用LangChain Agent完成复杂任务
• 无缝融入办公流：不用切换软件，飞书私聊、群组里@机器人就能用，全员零成本上手
• 企业级权限管控：依托飞书开放平台，支持权限细分、消息加密，适配企业数据安全要求
• 适配多种场景：除了咱们本章的数据库查询，后续还能扩展知识库问答、日程安排、多维表格操作等

简单说，接入飞书机器人后，咱们的LangChain Agent就从“本地工具”变成了“全员可用的办公AI”，实用性直接翻倍。

核心前置：飞书机器人接入总流程

整个接入过程不复杂，咱们拆成两大模块，逻辑清晰不迷路：飞书开放平台配置 + LangChain代码对接，全程不用啃晦涩文档，跟着步骤走即可。

第一步：飞书开放平台创建机器人（基础配置）

这一步是打通飞书通道的关键，所有权限、凭证都在这里搞定：

登录飞书开放平台：用企业飞书账号登录，进入开发者后台，选择企业自建应用

添加机器人能力：创建应用后，在应用详情页找到「应用能力」，开启机器人功能，设置机器人头像、名称，打造专属AI助手

获取核心凭证：在「凭证与基础信息」页面，复制App ID和App Secret，这两个是代码对接的“钥匙”，务必妥善保管

第二步：权限与事件配置

机器人要接收消息、回复消息，必须开通对应权限，不然会出现“收不到消息”“发不出回复”的问题：

• 权限开通：进入「权限管理」，搜索并开通im:message（接收消息）、im:message.group_at_msg（群内@接收）、im:message.send（发送消息）三大核心权限

• 事件订阅：切换到「事件与回调」，添加im.message.receive_v1接收消息事件，选择长连接模式

事件配置中，我这里没有公网ip，所以使用了长连接方式。

这里按照官网的提示，一步一步操作

1，安装依赖

pip install lark-oapi -U

2，下载示例代码

import lark_oapi as lark## P2ImMessageReceiveV1 为接收消息 v2.0；CustomizedEvent 内的 message 为接收消息 v1.0。def do_p2_im_message_receive_v1(data: lark.im.v1.P2ImMessageReceiveV1) -> None:    print(f'[ do_p2_im_message_receive_v1 access ], data: {lark.JSON.marshal(data, indent=4)}')def do_message_event(data: lark.CustomizedEvent) -> None:    print(f'[ do_customized_event access ], type: message, data: {lark.JSON.marshal(data, indent=4)}')event_handler = lark.EventDispatcherHandler.builder("", "") \    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \    .register_p1_customized_event("这里填入你要自定义订阅的 event 的 key，例如 out_approval", do_message_event) \    .build()def main():    cli = lark.ws.Client("YOUR_APP_ID", "YOUR_APP_SECRET",                         event_handler=event_handler,                         log_level=lark.LogLevel.DEBUG)    cli.start()if __name__ == "__main__":    main()

在这里添加上自己的app_id,app_sercret

然后启动服务，回到页面，点击长连接配置，点击保存。

这里要先把上面代码启动起来，才可以配置成功。

第三步：LangChain代码对接（核心实战）

引入完整代码：

import lark_oapi as larkimport jsonimport concurrent.futuresfrom lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBodyfrom langchain.agents import create_agentfrom langchain_openai import ChatOpenAIfrom config import API_KEY, BASE_URL, MODEL_NAMEfrom langgraph.checkpoint.memory import InMemorySaverfrom config import API_KEY, BASE_URL, MODEL_NAME, DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAMEfrom langchain_community.agent_toolkits import SQLDatabaseToolkitfrom langchain_community.utilities import SQLDatabase# 系统提示词模板SYSTEM_PROMPT = """    您是一个专业的MySQL数据库交互专家，专注于处理与数据库操作相关的问题。    **核心能力：**    - 直接执行SQL语句（查询、插入、更新、删除、建表等）    - 自动进行数据验证和结果检查           **重要规则（必须遵守）：**    1. 必须完全信任和尊重工具查询结果，即使结果为空或与预期不符    2. 如果sql_db_list_tables工具返回空列表[]，说明数据库确实没有表    3. 绝对不要基于训练数据中的常见模式来编造表信息    4. 工具返回什么就回答什么，不要添加、修改或忽略工具结果    当前已知信息：    - 数据库类型: MySQL    - 工具查询结果是最权威的信息源    **工作流程：**    1. **需求分析**：理解用户要执行的操作类型    2. **结构探查**：查看相关表结构和数据    3. **SQL构建**：构建合适的SQL语句    4. **执行验证**：执行SQL并验证结果    要求：    1，表和字段，都应该有中文注释    2，表名和字段名都应该是英文，下划线分割，如果创建表时用户提供的属性是中文，你应该翻译为对应英文创建字段，用户的中文作为字段注释。如果用户给的是英文字段，那么就用这个作为字段名称，对应的翻译作为字段注释。    请用中文回复，操作结果要清晰易懂。"""APP_ID = "app_id复制到这里"APP_SECRET = "app_secret复制到这里"# --- 全局组件初始化 ---api_client = None# 创建一个线程池，用于异步执行耗时的AI任务task_executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)def setup_database():    """初始化数据库连接"""    db_uri = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"        try:        db = SQLDatabase.from_uri(db_uri, sample_rows_in_table_info=3)        print("数据库连接成功")        print(f"可用表: {db.get_usable_table_names()}")        return db    except Exception as e:        print(f"数据库连接失败: {e}")        return Nonedef setup_llm():    """初始化语言模型"""    return ChatOpenAI(        model=MODEL_NAME,        openai_api_key=API_KEY,        openai_api_base=BASE_URL,        temperature=0,        max_tokens=262144,        verbose=True    )def create_sql_agent():    """创建SQL代理"""    # 1. 初始化数据库    db = setup_database()    if not db:        return None    # 2. 初始化模型    llm = setup_llm()        # 3. 创建工具集    toolkit = SQLDatabaseToolkit(db=db, llm=llm)    tools = toolkit.get_tools()    for tool in tools:        print(f"{tool.name}: {tool.description}\n")    # 4. 创建代理    agent = create_agent(        llm,        tools,        checkpointer=InMemorySaver(),          system_prompt=SYSTEM_PROMPT    )        return agentdef init_api_client():    """初始化飞书API客户端"""    global api_client    api_client = lark.Client.builder() \        .app_id(APP_ID) \        .app_secret(APP_SECRET) \        .log_level(lark.LogLevel.DEBUG) \        .build()# --- 核心异步处理函数 ---def do_p2_im_message_receive_v1(data: lark.im.v1.P2ImMessageReceiveV1) -> None:    """处理接收到的消息（主线程，必须快速返回）"""    # 获取消息发送者Id    sender_open_id = data.event.sender.sender_id.open_id    try:        # 取出消息内容        content_json = json.loads(data.event.message.content)        user_input = content_json.get("text", "")    except (json.JSONDecodeError, AttributeError) as e:        print(f"解析消息内容失败，使用原始文本: {e}")        user_input = data.event.message.content        print(f"来自用户 {sender_open_id[:8]}... 的输入: {user_input[:50]}...")        # 大模型对话延迟高，飞书回调需要3秒内处理，将这次对话任务提交到线程池    task_executor.submit(        async_process_and_reply,         sender_open_id,        user_input,        data.event.message.chat_type    )    print(f"已为用户 {sender_open_id[:8]}... 提交异步处理任务。主线程退出。")def async_process_and_reply(sender_open_id: str, user_input: str, chat_type: str):    """在后台线程中执行：调用AI并发送回复"""    print(f"[异步任务开始] 处理用户 {sender_open_id[:8]}... 的请求")        # 1. 调用大模型（可能耗时较长）    ai_response = "抱歉，我暂时无法处理这个问题。"  # 默认回复    try:        # 构造LangGraph Agent需要的输入        inputs = {"messages": [{"role": "user", "content": user_input}]}        # 修正：使用 sender_open_id 作为 thread_id，为每个用户维护独立对话历史        response = ai_agent.invoke(inputs, {"configurable": {"thread_id": sender_open_id}})        # 修正：从响应中安全地提取AI生成的内容        if isinstance(response, dict) and "messages" in response and response["messages"]:            last_message = response["messages"][-1]            if hasattr(last_message, 'content'):                ai_response = last_message.content            else:                ai_response = str(last_message)  # 备用方案    except Exception as e:        print(f"[异步任务] 调用AI Agent时出错: {e}")        ai_response = f"处理您的请求时遇到了内部错误: {str(e)[:100]}"        # 2. 发送回复    send_notification(sender_open_id, ai_response)    print(f"[异步任务完成] 已向用户 {sender_open_id[:8]}... 发送回复。")def send_notification(receive_id: str, message: str):    """发送消息函数（线程安全）"""    if not api_client:        print("❌ API客户端未初始化，无法发送消息")        return False    try:        body = CreateMessageRequestBody.builder() \            .receive_id(receive_id) \            .msg_type("text") \            .content(json.dumps({"text": message}, ensure_ascii=False)) \            .build()        request = CreateMessageRequest.builder() \            .receive_id_type("open_id") \            .request_body(body) \            .build()        response = api_client.im.v1.message.create(request)        if response.success():            print(f"✅ 消息发送成功！消息ID: {response.data.message_id}")            return True        else:            print(f"❌ 发送失败: {response.msg}, 请求ID: {response.request_id}")            return False    except Exception as e:        print(f"⚠️ 发送异常: {e}")        return False# --- 事件处理器与主函数 ---event_handler = lark.EventDispatcherHandler.builder("", "") \    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \    .build()def main():    global ai_agent    # 初始化    init_api_client()    print("飞书API客户端初始化完成")    print(f"AI模型 {MODEL_NAME} 初始化完成")    print("机器人启动中，等待连接...")        """主函数"""    ai_agent = create_sql_agent()    # 启动WebSocket客户端（会阻塞主线程）    cli = lark.ws.Client(        APP_ID,         APP_SECRET,         event_handler=event_handler,        log_level=lark.LogLevel.DEBUG    )    cli.start()if __name__ == "__main__":    main()

然后运行起来，就可以添加这个机器人聊天了。

回到开发者后台，提交版本，然后再飞书客户端中，就能看到开发者小助手推送的消息

点击打开应用，就能进入聊天界面

下面是我的测试截图