使用 LangChain 构建基于 DataHub 企业数据上下文的自主数据 Agent【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub在 DataHub 之上构建自主数据 Agent本文介绍如何通过datahub-agent-contextSDK 的 LangChain 集成将 DataHub 中沉淀的元数据上下文——数据归属ownership、血缘lineage、文档documentation、质量信号quality signals等——以可调用工具tool的形式接入 LangChain Agent使大模型能够检索数据集、解析 Schema、追踪血缘并可选执行元数据写操作。读完本文你将掌握从安装、快速搭建、模型提供商切换OpenAI / AWS Bedrock到 Agent 注册与排障的完整实战路径。概述Agent Context Kit 与 LangChain 集成datahub-agent-context是 DataHub 官方提供的 Agent 上下文工具包详见 Agent Context Kit 指南它把 DataHub 实例中沉淀的企业数据上下文以标准 MCP 工具的形式暴露给各类 AI Agent。LangChain 是其官方支持的三大 SDK 接入方式之一另见 Google ADK 与 MCP Server。LangChain 集成的核心价值在于你无需手工为每个工具编写 DataHub 连接代码datahub_agent_context.langchain_tools模块会基于DataHubClient自动构建一组封装好的 LangChainBaseTool且默认只读、可选写操作安全边界清晰。前置条件在开始之前请确认以下环境Python 3.10SDK 与 LangChain 1.x 的最低运行要求一个可访问的 DataHub 实例以及对应的 Personal Access Token用于认证一个支持工具调用tool-calling的大模型可以是 OpenAI API Key也可以是任何具备工具调用能力的 LLM 提供商如 AWS Bedrock 上的 Claude安装安装 DataHub Agent Context 的 LangChain 扩展及 OpenAI 模型接入包pip install datahub-agent-context[langchain] langchain-openai[langchain]extra 会拉取langchain与langchain-core两个核心依赖langchain-openai是独立安装项用于下文示例中的 OpenAI 模型提供商。从 setup.py 可以看到[langchain]extra 的完整依赖约束langchain1.0.0,2.0.0 langchain-core1.0.0,2.0.0 langchain-mcp-adapters0.1.0,1.0.0也就是说SDK 面向LangChain 1.x版本族设计并额外内置了langchain-mcp-adapters便于在需要时通过 MCP 适配器加载 DataHub MCP Server 的工具见后文 basic_agent.py 示例。快速开始三步接入 LangChain Agent第 1 步连接 DataHub 并构建工具集from datahub.sdk.main_client import DataHubClient from datahub_agent_context.langchain_tools import build_langchain_tools client DataHubClient.from_env() # 默认只读工具如需写操作添加标签、修改描述等设置 include_mutationsTrue tools build_langchain_tools(client, include_mutationsFalse)DataHubClient.from_env()会从环境变量读取 DataHub 连接配置GMS URL 与访问 Token。参考 simple_search.py 中的写法也可以显式指定import os datahub_gms_url os.getenv(DATAHUB_GMS_URL) if datahub_gms_url is None: client DataHubClient.from_env() else: client DataHubClient(serverdatahub_gms_url, tokenos.getenv(DATAHUB_GMS_TOKEN))第 2 步把工具接入 LangChain Agentfrom langchain_openai import ChatOpenAI from langchain.agents import create_agent llm ChatOpenAI(modelgpt-4o, temperature0) # 任何支持工具调用的模型均可 agent create_agent( llm, toolstools, system_promptYou are a data catalog assistant. Use the available tools to search for datasets, get entity details, and trace lineage. Always include URNs in your answers., ) response agent.invoke({messages: [{role: user, content: Find all datasets about customers}]}) print(response[messages][-1].content)注意这里使用的是 LangChain 1.x 的新版create_agentAPI而非旧版initialize_agent。system_prompt是引导 Agent 正确使用工具的抓手明确告知它何时搜索、何时取详情、何时追血缘并要求答案中始终附带 URN可显著提升回答质量。第 3 步切换模型提供商以 AWS Bedrock 为例LangChain 集成的价值之一在于模型无关。工具、Agent 构建与调用逻辑完全不变只需替换模型对象import boto3 from langchain_aws import ChatBedrock # pip install langchain-aws boto3 bedrock_runtime boto3.client(service_namebedrock-runtime, region_nameus-west-2) llm ChatBedrock( clientbedrock_runtime, model_idus.anthropic.claude-haiku-4-5-20251001-v1:0, model_kwargs{max_tokens: 2048, temperature: 0}, )完整可运行示例见 simple_search.py该示例即采用 Bedrock Claude Haiku 组合是“最小可用代码”的参考实现。内置工具清单读与写的安全边界build_langchain_tools的完整工具清单可在 builder.py 中确认。默认include_mutationsFalse只暴露只读工具能力类别工具说明搜索search、search_documents、grep_documents全文检索数据集与上下文文档实体详情get_entities、list_schema_fields获取实体信息与 Schema 字段上下文get_me当前用户/上下文信息血缘get_lineage、get_lineage_paths_between查询血缘关系及两实体间的血缘路径查询与质量get_dataset_queries、get_dataset_assertions数据集采样查询与质量断言事件list_incidents列出数据集事件/事故当include_mutationsTrue时会追加以下写操作工具元数据编辑update_description、save_document、set_domains/remove_domains归属管理add_owners/remove_owners、add_tags/remove_tags、add_glossary_terms/remove_glossary_terms、add_structured_properties/remove_structured_properties事件处理raise_incident、resolve_incident安全建议面向生产环境的 Agent 一律先以只读模式运行验证工具调用行为后再按需开启写操作并对DataHubClient使用的 Token 施加最小权限。另外DataHub Cloud 专属能力Ask DataHub AI 对话通过build_langchain_cloud_tools(client, ask_datahubTrue)获取可与你本地工具集合并使用tools build_langchain_tools(client, include_mutationsTrue) tools build_langchain_cloud_tools(client, ask_datahubTrue)源码原理工具如何与 DataHub 上下文自动绑定每个工具之所以能直接访问DataHubClient依赖 utils.py 中的create_context_wrapper它在工具执行前通过contextvars将客户端写入上下文执行结束后复位从而避免在工具函数间显式传递客户端对象同时把工具抛出的ItemNotFoundError捕获并转换为结构化字典响应保证 Agent/LLM 拿到的是可用消息而非未处理的异常。在 builder.py 中每个 MCP 工具都通过tool(create_context_wrapper(func, client))包装后追加到工具列表返回类型为list[BaseTool]可直接被 LangChain 的create_agent消费。进阶将 LangChain Agent 注册进 DataHub Agent Registry如果你希望 DataHub 能“看见”你的 LangChain Agent工具、模型、所用数据集SDK 提供了两种注册模式示例见 autoregister_agent.py。方式一回调处理器自动注册——把DataHubCallbackHandler挂到 AgentExecutor 上首次链执行完成chain_end时自动将 Agent 结构注册进 DataHubfrom datahub_agent_context._registration_core import AgentRegistrar from datahub_agent_context.langchain_registration import DataHubCallbackHandler registrar AgentRegistrar( frameworkLangChain, framework_versionNone, platformlangchain, agent_idorders-assistant, agent_nameOrders Assistant, descriptionAnswers questions about order data., ) handler DataHubCallbackHandler(registrar, tools[search_datasets, get_order_stats]) # 将 handler 挂到 AgentExecutor.callbacks 即可注册在首次 chain_end 时触发方式二一次性快照注册——Agent 构建完成后显式调用register_langchain_agent不侵入执行器from datahub_agent_context.langchain_registration import register_langchain_agent register_langchain_agent( executor, agent_idorders-assistant-v2, agent_nameOrders Assistant v2, descriptionSnapshot-registered agent using register_langchain_agent., )该函数会自动读取executor.tools中的工具形状名称、描述、输入 JSON Schema、executor.agent.llm的模型名以及通过importlib.metadata获取的langchain版本号然后一次性上报。核心实现见 handler.py 与 decorator.py。若要标记工具关联的数据集与技能可使用datahub_tool(datasets[...], skilldata-discovery)装饰器这些元数据会随注册一并写入 DataHub 图谱便于在 DataHub 中追踪每个 Agent 工具到底操作了哪些数据集。完整交互式示例basic_agent.py 提供了一个功能更完整的参考实现包含两个值得关注的点通过 MCP 适配器加载工具使用MultiServerMCPClient连接 DataHub MCP HTTP Server环境变量DATAHUB_MCP_URL、可选DATAHUB_MCP_TOKEN并从中拉取工具这为不直接依赖DataHubClient的场景提供了替代路径流式输出与工具调用可视化通过agent.astream(..., stream_modemessages)逐 token 打印模型输出并高亮展示每次工具调用的名称、参数与结果摘要。其系统提示词覆盖了搜索、Schema 解析、血缘追踪与元数据更新四大能力并内置了三条示例提问可直接体验1. Find all datasets related to customers 2. What are the schemas of the top 2 customer datasets? 3. Show me the upstream lineage for the first customer dataset you found故障排查工具执行报错先用client.test_connection()验证 DataHub 连通性再检查 Token 权限是否覆盖所需读/写范围。Agent 不使用工具强化system_prompt中对工具使用时机与输出格式的约束或将temperature设为0以获得更确定性的工具调用行为。导入错误ImportError确认已执行pip install datahub-agent-context[langchain] langchain-openai如使用 Bedrock还需pip install langchain-aws boto3。延伸阅读Agent Context Kit 总览了解 Data Analytics、Data Quality、Data Steward 三类 Agent 的构建场景MCP Server 指南以 MCP 方式将任意 Agent 接入 DataHubPersonal Access Token 文档配置 DataHub 认证凭据本集成源码langchain_tools 模块、langchain_registration 模块、examples/langchain 目录【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
使用 LangChain 构建基于 DataHub 企业数据上下文的自主数据 Agent
使用 LangChain 构建基于 DataHub 企业数据上下文的自主数据 Agent【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub在 DataHub 之上构建自主数据 Agent本文介绍如何通过datahub-agent-contextSDK 的 LangChain 集成将 DataHub 中沉淀的元数据上下文——数据归属ownership、血缘lineage、文档documentation、质量信号quality signals等——以可调用工具tool的形式接入 LangChain Agent使大模型能够检索数据集、解析 Schema、追踪血缘并可选执行元数据写操作。读完本文你将掌握从安装、快速搭建、模型提供商切换OpenAI / AWS Bedrock到 Agent 注册与排障的完整实战路径。概述Agent Context Kit 与 LangChain 集成datahub-agent-context是 DataHub 官方提供的 Agent 上下文工具包详见 Agent Context Kit 指南它把 DataHub 实例中沉淀的企业数据上下文以标准 MCP 工具的形式暴露给各类 AI Agent。LangChain 是其官方支持的三大 SDK 接入方式之一另见 Google ADK 与 MCP Server。LangChain 集成的核心价值在于你无需手工为每个工具编写 DataHub 连接代码datahub_agent_context.langchain_tools模块会基于DataHubClient自动构建一组封装好的 LangChainBaseTool且默认只读、可选写操作安全边界清晰。前置条件在开始之前请确认以下环境Python 3.10SDK 与 LangChain 1.x 的最低运行要求一个可访问的 DataHub 实例以及对应的 Personal Access Token用于认证一个支持工具调用tool-calling的大模型可以是 OpenAI API Key也可以是任何具备工具调用能力的 LLM 提供商如 AWS Bedrock 上的 Claude安装安装 DataHub Agent Context 的 LangChain 扩展及 OpenAI 模型接入包pip install datahub-agent-context[langchain] langchain-openai[langchain]extra 会拉取langchain与langchain-core两个核心依赖langchain-openai是独立安装项用于下文示例中的 OpenAI 模型提供商。从 setup.py 可以看到[langchain]extra 的完整依赖约束langchain1.0.0,2.0.0 langchain-core1.0.0,2.0.0 langchain-mcp-adapters0.1.0,1.0.0也就是说SDK 面向LangChain 1.x版本族设计并额外内置了langchain-mcp-adapters便于在需要时通过 MCP 适配器加载 DataHub MCP Server 的工具见后文 basic_agent.py 示例。快速开始三步接入 LangChain Agent第 1 步连接 DataHub 并构建工具集from datahub.sdk.main_client import DataHubClient from datahub_agent_context.langchain_tools import build_langchain_tools client DataHubClient.from_env() # 默认只读工具如需写操作添加标签、修改描述等设置 include_mutationsTrue tools build_langchain_tools(client, include_mutationsFalse)DataHubClient.from_env()会从环境变量读取 DataHub 连接配置GMS URL 与访问 Token。参考 simple_search.py 中的写法也可以显式指定import os datahub_gms_url os.getenv(DATAHUB_GMS_URL) if datahub_gms_url is None: client DataHubClient.from_env() else: client DataHubClient(serverdatahub_gms_url, tokenos.getenv(DATAHUB_GMS_TOKEN))第 2 步把工具接入 LangChain Agentfrom langchain_openai import ChatOpenAI from langchain.agents import create_agent llm ChatOpenAI(modelgpt-4o, temperature0) # 任何支持工具调用的模型均可 agent create_agent( llm, toolstools, system_promptYou are a data catalog assistant. Use the available tools to search for datasets, get entity details, and trace lineage. Always include URNs in your answers., ) response agent.invoke({messages: [{role: user, content: Find all datasets about customers}]}) print(response[messages][-1].content)注意这里使用的是 LangChain 1.x 的新版create_agentAPI而非旧版initialize_agent。system_prompt是引导 Agent 正确使用工具的抓手明确告知它何时搜索、何时取详情、何时追血缘并要求答案中始终附带 URN可显著提升回答质量。第 3 步切换模型提供商以 AWS Bedrock 为例LangChain 集成的价值之一在于模型无关。工具、Agent 构建与调用逻辑完全不变只需替换模型对象import boto3 from langchain_aws import ChatBedrock # pip install langchain-aws boto3 bedrock_runtime boto3.client(service_namebedrock-runtime, region_nameus-west-2) llm ChatBedrock( clientbedrock_runtime, model_idus.anthropic.claude-haiku-4-5-20251001-v1:0, model_kwargs{max_tokens: 2048, temperature: 0}, )完整可运行示例见 simple_search.py该示例即采用 Bedrock Claude Haiku 组合是“最小可用代码”的参考实现。内置工具清单读与写的安全边界build_langchain_tools的完整工具清单可在 builder.py 中确认。默认include_mutationsFalse只暴露只读工具能力类别工具说明搜索search、search_documents、grep_documents全文检索数据集与上下文文档实体详情get_entities、list_schema_fields获取实体信息与 Schema 字段上下文get_me当前用户/上下文信息血缘get_lineage、get_lineage_paths_between查询血缘关系及两实体间的血缘路径查询与质量get_dataset_queries、get_dataset_assertions数据集采样查询与质量断言事件list_incidents列出数据集事件/事故当include_mutationsTrue时会追加以下写操作工具元数据编辑update_description、save_document、set_domains/remove_domains归属管理add_owners/remove_owners、add_tags/remove_tags、add_glossary_terms/remove_glossary_terms、add_structured_properties/remove_structured_properties事件处理raise_incident、resolve_incident安全建议面向生产环境的 Agent 一律先以只读模式运行验证工具调用行为后再按需开启写操作并对DataHubClient使用的 Token 施加最小权限。另外DataHub Cloud 专属能力Ask DataHub AI 对话通过build_langchain_cloud_tools(client, ask_datahubTrue)获取可与你本地工具集合并使用tools build_langchain_tools(client, include_mutationsTrue) tools build_langchain_cloud_tools(client, ask_datahubTrue)源码原理工具如何与 DataHub 上下文自动绑定每个工具之所以能直接访问DataHubClient依赖 utils.py 中的create_context_wrapper它在工具执行前通过contextvars将客户端写入上下文执行结束后复位从而避免在工具函数间显式传递客户端对象同时把工具抛出的ItemNotFoundError捕获并转换为结构化字典响应保证 Agent/LLM 拿到的是可用消息而非未处理的异常。在 builder.py 中每个 MCP 工具都通过tool(create_context_wrapper(func, client))包装后追加到工具列表返回类型为list[BaseTool]可直接被 LangChain 的create_agent消费。进阶将 LangChain Agent 注册进 DataHub Agent Registry如果你希望 DataHub 能“看见”你的 LangChain Agent工具、模型、所用数据集SDK 提供了两种注册模式示例见 autoregister_agent.py。方式一回调处理器自动注册——把DataHubCallbackHandler挂到 AgentExecutor 上首次链执行完成chain_end时自动将 Agent 结构注册进 DataHubfrom datahub_agent_context._registration_core import AgentRegistrar from datahub_agent_context.langchain_registration import DataHubCallbackHandler registrar AgentRegistrar( frameworkLangChain, framework_versionNone, platformlangchain, agent_idorders-assistant, agent_nameOrders Assistant, descriptionAnswers questions about order data., ) handler DataHubCallbackHandler(registrar, tools[search_datasets, get_order_stats]) # 将 handler 挂到 AgentExecutor.callbacks 即可注册在首次 chain_end 时触发方式二一次性快照注册——Agent 构建完成后显式调用register_langchain_agent不侵入执行器from datahub_agent_context.langchain_registration import register_langchain_agent register_langchain_agent( executor, agent_idorders-assistant-v2, agent_nameOrders Assistant v2, descriptionSnapshot-registered agent using register_langchain_agent., )该函数会自动读取executor.tools中的工具形状名称、描述、输入 JSON Schema、executor.agent.llm的模型名以及通过importlib.metadata获取的langchain版本号然后一次性上报。核心实现见 handler.py 与 decorator.py。若要标记工具关联的数据集与技能可使用datahub_tool(datasets[...], skilldata-discovery)装饰器这些元数据会随注册一并写入 DataHub 图谱便于在 DataHub 中追踪每个 Agent 工具到底操作了哪些数据集。完整交互式示例basic_agent.py 提供了一个功能更完整的参考实现包含两个值得关注的点通过 MCP 适配器加载工具使用MultiServerMCPClient连接 DataHub MCP HTTP Server环境变量DATAHUB_MCP_URL、可选DATAHUB_MCP_TOKEN并从中拉取工具这为不直接依赖DataHubClient的场景提供了替代路径流式输出与工具调用可视化通过agent.astream(..., stream_modemessages)逐 token 打印模型输出并高亮展示每次工具调用的名称、参数与结果摘要。其系统提示词覆盖了搜索、Schema 解析、血缘追踪与元数据更新四大能力并内置了三条示例提问可直接体验1. Find all datasets related to customers 2. What are the schemas of the top 2 customer datasets? 3. Show me the upstream lineage for the first customer dataset you found故障排查工具执行报错先用client.test_connection()验证 DataHub 连通性再检查 Token 权限是否覆盖所需读/写范围。Agent 不使用工具强化system_prompt中对工具使用时机与输出格式的约束或将temperature设为0以获得更确定性的工具调用行为。导入错误ImportError确认已执行pip install datahub-agent-context[langchain] langchain-openai如使用 Bedrock还需pip install langchain-aws boto3。延伸阅读Agent Context Kit 总览了解 Data Analytics、Data Quality、Data Steward 三类 Agent 的构建场景MCP Server 指南以 MCP 方式将任意 Agent 接入 DataHubPersonal Access Token 文档配置 DataHub 认证凭据本集成源码langchain_tools 模块、langchain_registration 模块、examples/langchain 目录【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
